From e227ff3bbfb303d1e7e4301d213377d6e06e6cdd Mon Sep 17 00:00:00 2001 From: shinyu Date: Fri, 24 Jul 2026 14:54:35 +0900 Subject: [PATCH 1/9] feat(memory): observability + opt-in watchdog with dashboard (Monitor/Recommend/Adjust) Mitigate the Bun/mimalloc committed-memory retention that can exhaust the system commit charge in long sessions. App code cannot free native-retained memory, so this ships mitigation (observe -> recommend -> safe restart), not a cure. Runtime: portable memory snapshot (memory-usage.ts) reads Windows private/committed via a guarded PowerShell probe (hard timeout, soft-fail), Linux /proc/self/status, RSS fallback elsewhere; all thresholds are fractions of total RAM so behavior is identical on every machine. memory-watchdog.ts adds detectSupervisor, pure evaluate, recommend, live applyWatchdogRuntimeConfig, and a requireSupervisor gate that suppresses auto-restart to warn-only when no supervisor is present. responseStateMetrics() is a side-effect-free seam to distinguish JS-heap vs native growth. API: GET /api/memory (read-only report) and PUT /api/memory/settings (validated, live-applied, persisted) behind the existing management auth + local-only gate. GUI: dashboard (beta) MemoryCard with three layered opt-in options; i18n keys added across en/de/ko/zh/ru/ja. Defaults: enabled, warn-only, autoRestart=false, requireSupervisor=true. Tests: memory-watchdog (34) + memory-api + responseStateMetrics. --- gui/src/components/MemoryCard.tsx | 293 +++++++++++++++++++ gui/src/i18n/de.ts | 28 ++ gui/src/i18n/en.ts | 28 ++ gui/src/i18n/ja.ts | 28 ++ gui/src/i18n/ko.ts | 28 ++ gui/src/i18n/ru.ts | 28 ++ gui/src/i18n/zh.ts | 28 ++ gui/src/pages/Dashboard.tsx | 3 + scripts/memory-stress-harness.ts | 356 +++++++++++++++++++++++ src/lib/memory-usage.ts | 175 ++++++++++++ src/responses/state.ts | 44 +++ src/server/index.ts | 7 + src/server/management-api.ts | 47 +++ src/server/memory-watchdog.ts | 456 ++++++++++++++++++++++++++++++ src/types.ts | 24 ++ tests/memory-api.test.ts | 144 ++++++++++ tests/memory-watchdog.test.ts | 383 +++++++++++++++++++++++++ tests/responses-state.test.ts | 42 +++ 18 files changed, 2142 insertions(+) create mode 100644 gui/src/components/MemoryCard.tsx create mode 100644 scripts/memory-stress-harness.ts create mode 100644 src/lib/memory-usage.ts create mode 100644 src/server/memory-watchdog.ts create mode 100644 tests/memory-api.test.ts create mode 100644 tests/memory-watchdog.test.ts diff --git a/gui/src/components/MemoryCard.tsx b/gui/src/components/MemoryCard.tsx new file mode 100644 index 00000000000..fa118b6e29a --- /dev/null +++ b/gui/src/components/MemoryCard.tsx @@ -0,0 +1,293 @@ +import { useCallback, useEffect, useRef, useState } from "react"; +import { useI18n } from "../i18n"; + +/* + * MemoryCard — dashboard (beta) panel for the memory watchdog. Three layered, opt-in options: + * 1. Monitor — read-only live memory pressure vs total RAM (zero risk). + * 2. Recommend — advisory thresholds from observed baseline (never applied automatically). + * 3. Adjust — toggle/tune thresholds + opt-in auto-restart (management-auth gated; auto-restart + * is further gated by supervisor detection so we never exit into "just dead"). + * Self-contained: it owns its own fetch/poll/PUT so it does not couple to the large Dashboard. + */ + +type WatchdogLevel = "ok" | "warn" | "critical"; + +interface WatchdogDecision { + level: WatchdogLevel; + action: string; + fraction: number; + pressureMb: number; + totalMb: number; + source: string; + growthMbPerHour: number | null; + reason: string; +} + +interface ResolvedConfig { + enabled: boolean; + intervalMs: number; + warnFraction: number; + criticalFraction: number; + autoRestart: boolean; + requireSupervisor: boolean; + minRestartIntervalMs: number; + maxRestarts: number; + growthWindowMs: number; +} + +interface Recommendation { + warnFraction: number; + criticalFraction: number; + autoRestart: boolean; + rationale: string; +} + +interface MemoryReport { + enabled: boolean; + decision: WatchdogDecision | null; + samplesCount?: number; + growthMbPerHour?: number | null; + resolvedConfig?: ResolvedConfig; + supervisor?: { supervised: boolean; hint: string }; + recommendation?: Recommendation; + restartCount?: number; +} + +interface SettingsPatch { + enabled?: boolean; + warnFraction?: number; + criticalFraction?: number; + autoRestart?: boolean; + requireSupervisor?: boolean; +} + +const LEVEL_COLOR: Record = { + ok: "var(--green)", + warn: "var(--amber, #d89614)", + critical: "var(--red)", +}; + +function pct(fraction: number): number { + return Math.round(fraction * 1000) / 10; +} + +/** Stateless fetch so the polling effect never calls setState synchronously in its body. */ +async function loadReport(apiBase: string): Promise { + try { + const res = await fetch(`${apiBase}/api/memory`); + if (!res.ok) return null; + return (await res.json()) as MemoryReport; + } catch { + return null; // old server / offline — keep the last report + } +} + +export default function MemoryCard({ apiBase }: { apiBase: string }) { + const { t, locale } = useI18n(); + const [report, setReport] = useState(null); + const [loaded, setLoaded] = useState(false); + const [saving, setSaving] = useState(false); + const [saveError, setSaveError] = useState(null); + // Draft threshold percents while dragging; committed on "Apply thresholds". + const [warnDraft, setWarnDraft] = useState(null); + const [critDraft, setCritDraft] = useState(null); + const editingRef = useRef(false); + + useEffect(() => { + let cancelled = false; + const refresh = async () => { + const data = await loadReport(apiBase); + if (cancelled) return; + if (data) setReport(data); + setLoaded(true); + }; + void refresh(); + const interval = setInterval(() => { + // Don't stomp the sliders the user is dragging. + if (!editingRef.current) void refresh(); + }, 5000); + return () => { cancelled = true; clearInterval(interval); }; + }, [apiBase]); + + const save = useCallback(async (patch: SettingsPatch) => { + setSaving(true); + setSaveError(null); + try { + const res = await fetch(`${apiBase}/api/memory/settings`, { + method: "PUT", + headers: { "content-type": "application/json" }, + body: JSON.stringify(patch), + }); + if (!res.ok) { + const body = (await res.json().catch(() => null)) as { error?: string } | null; + setSaveError(body?.error ?? t("memory.httpError", { status: res.status })); + } + } catch (err) { + setSaveError(err instanceof Error ? err.message : String(err)); + } finally { + setSaving(false); + editingRef.current = false; + const data = await loadReport(apiBase); + if (data) setReport(data); + } + }, [apiBase, t]); + + if (!loaded) return null; + if (!report || report.enabled === false) { + return ( +
+
+ {t("memory.title")} + beta +
+

{t("memory.disabledHint")}

+ + {saveError &&
{saveError}
} +
+ ); + } + + const cfg = report.resolvedConfig; + const decision = report.decision; + const rec = report.recommendation; + const sup = report.supervisor; + const level = decision?.level ?? "ok"; + const fraction = decision?.fraction ?? 0; + const warnPct = warnDraft ?? (cfg ? Math.round(cfg.warnFraction * 100) : 60); + const critPct = critDraft ?? (cfg ? Math.round(cfg.criticalFraction * 100) : 75); + const autoRestartBlocked = !!cfg && cfg.requireSupervisor && !(sup?.supervised); + const nf = (n: number) => n.toLocaleString(locale); + + const applyThresholds = () => { + // Server clamps + keeps critical > warn; send fractions. + void save({ warnFraction: warnPct / 100, criticalFraction: Math.max(critPct, warnPct + 1) / 100 }); + }; + const applyRecommended = () => { + if (!rec) return; + setWarnDraft(Math.round(rec.warnFraction * 100)); + setCritDraft(Math.round(rec.criticalFraction * 100)); + void save({ + warnFraction: rec.warnFraction, + criticalFraction: rec.criticalFraction, + autoRestart: rec.autoRestart && !autoRestartBlocked ? true : false, + }); + }; + + return ( +
+
+ {t("memory.title")} + beta + {level.toUpperCase()} +
+

{t("memory.intro")}

+ + {/* 1. Monitor */} +
+
+ {t("memory.pressure")} + + {decision ? t("memory.pressureValue", { used: nf(decision.pressureMb), total: nf(decision.totalMb), pct: pct(fraction) }) : "—"} + +
+
+
+ {cfg && } + {cfg && } +
+
+ {t("memory.sourceLabel")} {decision?.source ?? "—"} + {t("memory.growthLabel")} {report.growthMbPerHour == null ? t("memory.notAvailable") : t("memory.growthValue", { mb: nf(report.growthMbPerHour) })} + {t("memory.restartsLabel")} {nf(report.restartCount ?? 0)} + {t("memory.supervisorLabel")} {sup?.supervised ? t("memory.supervisorDetected", { hint: sup.hint }) : t("memory.supervisorNotDetected")} +
+
+ + {/* 2. Recommend */} + {rec && ( +
+
+ {t("memory.recommended")} + {t("memory.recSummary", { + warn: Math.round(rec.warnFraction * 100), + crit: Math.round(rec.criticalFraction * 100), + auto: rec.autoRestart ? t("memory.yes") : t("memory.no"), + })} + +
+
{rec.rationale}
+
+ )} + + {/* 3. Adjust */} +
+ + +
+ { editingRef.current = true; setWarnDraft(v); }} /> + { editingRef.current = true; setCritDraft(v); }} /> +
+ +
+
+ + + {autoRestartBlocked && ( +
{t("memory.autoRestartBlocked")}
+ )} + + +
+ {saving && {t("common.saving")}} + {!saving && saveError && {saveError}} +
+
+
+ ); +} + +const betaBadge: React.CSSProperties = { + background: "transparent", + color: "var(--muted)", + border: "1px solid var(--border, var(--muted))", + padding: "1px 7px", + borderRadius: "var(--radius-pill)", + fontSize: 11, + textTransform: "uppercase", + letterSpacing: 0.4, +}; + +const rowStyle: React.CSSProperties = { display: "flex", alignItems: "center", gap: 8, cursor: "pointer" }; + +function Marker({ at, color }: { at: number; color: string }) { + return
; +} + +function Slider({ label, value, min, max, onChange }: { label: string; value: number; min: number; max: number; onChange: (v: number) => void }) { + return ( + + ); +} diff --git a/gui/src/i18n/de.ts b/gui/src/i18n/de.ts index 287b83e9758..8f48ad443cb 100644 --- a/gui/src/i18n/de.ts +++ b/gui/src/i18n/de.ts @@ -1,5 +1,33 @@ // German — generated from en.ts. Must match TKey set (compile-checked). export const de = { + // memory watchdog (beta) + "memory.title": "Speicher-Watchdog", + "memory.disabledHint": "Watchdog deaktiviert. Aktivieren Sie ihn, um den Speicherdruck zu beobachten und Warnungen zu erhalten.", + "memory.enable": "Aktivieren", + "memory.intro": "Prozessspeicher im Verhältnis zum Gesamt-RAM beobachten. Standardmäßig nur Warnung; Auto-Neustart ist optional und wird nur ausgelöst, wenn ein Supervisor erkannt wird.", + "memory.pressure": "Speicherdruck", + "memory.pressureValue": "{used} / {total} MB ({pct}%)", + "memory.sourceLabel": "Quelle:", + "memory.growthLabel": "Wachstum:", + "memory.growthValue": "{mb} MB/h", + "memory.notAvailable": "n/v", + "memory.restartsLabel": "Neustarts:", + "memory.supervisorLabel": "Supervisor:", + "memory.supervisorDetected": "erkannt ({hint})", + "memory.supervisorNotDetected": "nicht erkannt", + "memory.recommended": "Empfohlen", + "memory.recSummary": "Warnung {warn}% · kritisch {crit}% · Auto-Neustart {auto}", + "memory.applyRecommended": "Empfehlung übernehmen", + "memory.enabledLabel": "Aktiviert", + "memory.warnLabel": "Warnung %", + "memory.criticalLabel": "Kritisch %", + "memory.applyThresholds": "Schwellen übernehmen", + "memory.autoRestartLabel": "Auto-Neustart bei kritisch (optional)", + "memory.autoRestartBlocked": "Kein Supervisor erkannt — Auto-Neustart bleibt aus, damit der Proxy nie einfach beendet wird. Setzen Sie OCX_SUPERVISED=1 (oder betreiben Sie ihn unter pm2/systemd) oder deaktivieren Sie „Supervisor erforderlich“.", + "memory.requireSupervisorLabel": "Supervisor vor Auto-Neustart erforderlich", + "memory.yes": "ja", + "memory.no": "nein", + "memory.httpError": "HTTP {status}", "nav.dashboard": "Übersicht", "nav.providers": "Anbieter", "nav.models": "Modelle", diff --git a/gui/src/i18n/en.ts b/gui/src/i18n/en.ts index 3430c5ba0ba..c6b7e283fd5 100644 --- a/gui/src/i18n/en.ts +++ b/gui/src/i18n/en.ts @@ -1,6 +1,34 @@ // English — source of truth. Its keys define the TKey type; ko/zh/ja must match (compile-checked). // Strings with {cmd} render a chip via ; {var} are plain interpolations. export const en = { + // memory watchdog (beta) + "memory.title": "Memory watchdog", + "memory.disabledHint": "Watchdog disabled. Enable it to observe memory pressure and receive warnings.", + "memory.enable": "Enable", + "memory.intro": "Observe process memory vs total RAM. Warn-only by default; auto-restart is opt-in and only fires when a supervisor is detected.", + "memory.pressure": "Memory pressure", + "memory.pressureValue": "{used} / {total} MB ({pct}%)", + "memory.sourceLabel": "source:", + "memory.growthLabel": "growth:", + "memory.growthValue": "{mb} MB/h", + "memory.notAvailable": "n/a", + "memory.restartsLabel": "restarts:", + "memory.supervisorLabel": "supervisor:", + "memory.supervisorDetected": "detected ({hint})", + "memory.supervisorNotDetected": "not detected", + "memory.recommended": "Recommended", + "memory.recSummary": "warn {warn}% · critical {crit}% · auto-restart {auto}", + "memory.applyRecommended": "Apply recommended", + "memory.enabledLabel": "Enabled", + "memory.warnLabel": "Warn %", + "memory.criticalLabel": "Critical %", + "memory.applyThresholds": "Apply thresholds", + "memory.autoRestartLabel": "Auto-restart on critical (opt-in)", + "memory.autoRestartBlocked": "No supervisor detected — auto-restart stays off so the proxy is never left dead. Set OCX_SUPERVISED=1 (or run under pm2/systemd), or turn off \"Require supervisor\".", + "memory.requireSupervisorLabel": "Require supervisor before auto-restart", + "memory.yes": "yes", + "memory.no": "no", + "memory.httpError": "HTTP {status}", // sidebar / nav / common "nav.dashboard": "Dashboard", "nav.providers": "Providers", diff --git a/gui/src/i18n/ja.ts b/gui/src/i18n/ja.ts index 0babac0bffc..f9253f3babb 100644 --- a/gui/src/i18n/ja.ts +++ b/gui/src/i18n/ja.ts @@ -1,6 +1,34 @@ import type { TKey } from "./en"; export const ja: Record = { + // memory watchdog (beta) + "memory.title": "メモリウォッチドッグ", + "memory.disabledHint": "ウォッチドッグは無効です。有効にするとメモリ負荷を監視し警告を受け取れます。", + "memory.enable": "有効化", + "memory.intro": "総RAMに対するプロセスメモリを監視します。デフォルトは警告のみ。自動再起動はオプトインで、スーパーバイザーが検出された場合にのみ実行されます。", + "memory.pressure": "メモリ負荷", + "memory.pressureValue": "{used} / {total} MB ({pct}%)", + "memory.sourceLabel": "ソース:", + "memory.growthLabel": "増加:", + "memory.growthValue": "{mb} MB/h", + "memory.notAvailable": "なし", + "memory.restartsLabel": "再起動:", + "memory.supervisorLabel": "スーパーバイザー:", + "memory.supervisorDetected": "検出 ({hint})", + "memory.supervisorNotDetected": "未検出", + "memory.recommended": "推奨", + "memory.recSummary": "警告 {warn}% · 危険 {crit}% · 自動再起動 {auto}", + "memory.applyRecommended": "推奨値を適用", + "memory.enabledLabel": "有効", + "memory.warnLabel": "警告 %", + "memory.criticalLabel": "危険 %", + "memory.applyThresholds": "しきい値を適用", + "memory.autoRestartLabel": "危険時に自動再起動（オプトイン）", + "memory.autoRestartBlocked": "スーパーバイザーが検出されません — プロキシがそのまま停止しないよう自動再起動はオフのままです。OCX_SUPERVISED=1 を設定する（または pm2/systemd 下で実行する）か、「スーパーバイザーを必須にする」をオフにしてください。", + "memory.requireSupervisorLabel": "自動再起動の前にスーパーバイザーを必須にする", + "memory.yes": "はい", + "memory.no": "いいえ", + "memory.httpError": "HTTP {status}", // sidebar / nav / common "nav.dashboard": "ダッシュボード", "nav.providers": "プロバイダー", diff --git a/gui/src/i18n/ko.ts b/gui/src/i18n/ko.ts index acbaf9750a4..dce951a0aa4 100644 --- a/gui/src/i18n/ko.ts +++ b/gui/src/i18n/ko.ts @@ -1,6 +1,34 @@ import type { TKey } from "./en"; export const ko: Record = { + // memory watchdog (beta) + "memory.title": "메모리 워치독", + "memory.disabledHint": "워치독이 비활성화되었습니다. 메모리 압력을 관찰하고 경고를 받으려면 활성화하세요.", + "memory.enable": "활성화", + "memory.intro": "전체 RAM 대비 프로세스 메모리를 관찰합니다. 기본적으로 경고만 하며, 자동 재시작은 선택 사항이고 supervisor가 감지된 경우에만 실행됩니다.", + "memory.pressure": "메모리 압력", + "memory.pressureValue": "{used} / {total} MB ({pct}%)", + "memory.sourceLabel": "소스:", + "memory.growthLabel": "증가:", + "memory.growthValue": "{mb} MB/h", + "memory.notAvailable": "해당 없음", + "memory.restartsLabel": "재시작:", + "memory.supervisorLabel": "supervisor:", + "memory.supervisorDetected": "감지됨 ({hint})", + "memory.supervisorNotDetected": "감지되지 않음", + "memory.recommended": "권장", + "memory.recSummary": "경고 {warn}% · 위험 {crit}% · 자동 재시작 {auto}", + "memory.applyRecommended": "권장값 적용", + "memory.enabledLabel": "활성화됨", + "memory.warnLabel": "경고 %", + "memory.criticalLabel": "위험 %", + "memory.applyThresholds": "임계값 적용", + "memory.autoRestartLabel": "위험 시 자동 재시작 (선택)", + "memory.autoRestartBlocked": "supervisor가 감지되지 않았습니다 — 프록시가 그냥 종료되지 않도록 자동 재시작이 꺼진 상태로 유지됩니다. OCX_SUPERVISED=1을 설정하거나(또는 pm2/systemd에서 실행) \"supervisor 필요\"를 끄세요.", + "memory.requireSupervisorLabel": "자동 재시작 전에 supervisor 필요", + "memory.yes": "예", + "memory.no": "아니오", + "memory.httpError": "HTTP {status}", // sidebar / nav / common "nav.dashboard": "대시보드", "nav.providers": "프로바이더", diff --git a/gui/src/i18n/ru.ts b/gui/src/i18n/ru.ts index a4872452021..f3407128220 100644 --- a/gui/src/i18n/ru.ts +++ b/gui/src/i18n/ru.ts @@ -1,6 +1,34 @@ import type { TKey } from "./en"; export const ru: Record = { + // memory watchdog (beta) + "memory.title": "Сторож памяти", + "memory.disabledHint": "Сторож отключён. Включите его, чтобы отслеживать давление памяти и получать предупреждения.", + "memory.enable": "Включить", + "memory.intro": "Отслеживание памяти процесса относительно общего ОЗУ. По умолчанию только предупреждение; автоперезапуск включается по желанию и срабатывает только при обнаружении супервизора.", + "memory.pressure": "Давление памяти", + "memory.pressureValue": "{used} / {total} МБ ({pct}%)", + "memory.sourceLabel": "источник:", + "memory.growthLabel": "рост:", + "memory.growthValue": "{mb} МБ/ч", + "memory.notAvailable": "н/д", + "memory.restartsLabel": "перезапуски:", + "memory.supervisorLabel": "супервизор:", + "memory.supervisorDetected": "обнаружен ({hint})", + "memory.supervisorNotDetected": "не обнаружен", + "memory.recommended": "Рекомендуется", + "memory.recSummary": "предупр. {warn}% · критич. {crit}% · автоперезапуск {auto}", + "memory.applyRecommended": "Применить рекомендуемое", + "memory.enabledLabel": "Включено", + "memory.warnLabel": "Предупр. %", + "memory.criticalLabel": "Критич. %", + "memory.applyThresholds": "Применить пороги", + "memory.autoRestartLabel": "Автоперезапуск при критическом (по желанию)", + "memory.autoRestartBlocked": "Супервизор не обнаружен — автоперезапуск остаётся выключенным, чтобы прокси не завершился без восстановления. Установите OCX_SUPERVISED=1 (или запустите под pm2/systemd) либо отключите «Требовать супервизор».", + "memory.requireSupervisorLabel": "Требовать супервизор перед автоперезапуском", + "memory.yes": "да", + "memory.no": "нет", + "memory.httpError": "HTTP {status}", // sidebar / nav / common "nav.dashboard": "Дашборд", "nav.providers": "Провайдеры", diff --git a/gui/src/i18n/zh.ts b/gui/src/i18n/zh.ts index d3ebea74206..61698ca2a80 100644 --- a/gui/src/i18n/zh.ts +++ b/gui/src/i18n/zh.ts @@ -1,6 +1,34 @@ import type { TKey } from "./en"; export const zh: Record = { + // memory watchdog (beta) + "memory.title": "内存看门狗", + "memory.disabledHint": "看门狗已禁用。启用后可观察内存压力并接收警告。", + "memory.enable": "启用", + "memory.intro": "观察进程内存与总内存的比例。默认仅警告；自动重启为可选项，且仅在检测到 supervisor 时触发。", + "memory.pressure": "内存压力", + "memory.pressureValue": "{used} / {total} MB ({pct}%)", + "memory.sourceLabel": "来源：", + "memory.growthLabel": "增长：", + "memory.growthValue": "{mb} MB/h", + "memory.notAvailable": "不可用", + "memory.restartsLabel": "重启：", + "memory.supervisorLabel": "supervisor：", + "memory.supervisorDetected": "已检测 ({hint})", + "memory.supervisorNotDetected": "未检测", + "memory.recommended": "推荐", + "memory.recSummary": "警告 {warn}% · 严重 {crit}% · 自动重启 {auto}", + "memory.applyRecommended": "应用推荐值", + "memory.enabledLabel": "已启用", + "memory.warnLabel": "警告 %", + "memory.criticalLabel": "严重 %", + "memory.applyThresholds": "应用阈值", + "memory.autoRestartLabel": "严重时自动重启（可选）", + "memory.autoRestartBlocked": "未检测到 supervisor — 自动重启保持关闭，以免代理直接退出。请设置 OCX_SUPERVISED=1（或在 pm2/systemd 下运行），或关闭\"需要 supervisor\"。", + "memory.requireSupervisorLabel": "自动重启前需要 supervisor", + "memory.yes": "是", + "memory.no": "否", + "memory.httpError": "HTTP {status}", // sidebar / nav / common "nav.dashboard": "仪表盘", "nav.providers": "提供方", diff --git a/gui/src/pages/Dashboard.tsx b/gui/src/pages/Dashboard.tsx index 2d2bc5d1462..459413ca878 100644 --- a/gui/src/pages/Dashboard.tsx +++ b/gui/src/pages/Dashboard.tsx @@ -5,6 +5,7 @@ import { Trans } from "../i18n/provider"; import { useI18n, type TKey } from "../i18n/shared"; import { formatTokens } from "../format-tokens"; import { EmptyState, Select } from "../ui"; +import MemoryCard from "../components/MemoryCard"; interface HealthData { status: string; version: string; uptime: number } interface ProviderInfo { name: string; adapter: string; baseUrl: string; defaultModel?: string; hasApiKey: boolean } @@ -617,6 +618,8 @@ export default function Dashboard({ apiBase }: { apiBase: string }) {
+ + {projectConfigWarnings.length > 0 && (
diff --git a/scripts/memory-stress-harness.ts b/scripts/memory-stress-harness.ts new file mode 100644 index 00000000000..3711f06bd85 --- /dev/null +++ b/scripts/memory-stress-harness.ts @@ -0,0 +1,356 @@ +/** + * Isolated memory-stress harness for the responses continuation store. + * + * WHY: OpenCodex on Bun exhibits the same "committed memory >> working set + system stutter" + * signature that a sibling Bun CLI (Claude Code #36132) traced to the mimalloc allocator leaking + * committed memory ~0.6-1 GB/h, independent of the JS heap. Before touching the request path we + * need machine-local evidence that separates two hypotheses: + * (A) the JS-side continuation store (this repo) grows RAM, vs + * (B) the Bun native allocator retains memory the JS heap has already released. + * + * WHAT IT DOES (safe, isolated, accelerated): + * - Scenario "chain": one long previous_response_id chain -> proves prefix reference sharing + * (heapUsed grows ~O(N) while serialized totalBytes grows ~O(N^2)). + * - Scenario "sessions": many rotating independent chains with large payloads -> shows how much + * RAM the current COUNT-only cap (MAX_STORED_RESPONSES) still permits. + * - Return-to-OS check: clear the store + force GC, then resample RSS. If RSS does not fall, + * that is the native-retention (hypothesis B) signal. + * + * SAFETY: + * - Runs the workload in a CHILD process (parent orchestrates the A/B). + * - Child self-aborts if its own RSS exceeds --rss-cap-mb (default 1200 MB). + * - Parent enforces a hard wall-clock timeout and SIGKILLs the child if it overruns. + * - Uses an isolated OPENCODEX_HOME temp dir so it never touches the real ~/.opencodex. + * + * USAGE: + * bun run scripts/memory-stress-harness.ts # parent: runs the A/B and prints a table + * bun run scripts/memory-stress-harness.ts --child # single child run (used internally) + * + * TUNING (env, all optional): + * OCX_STRESS_CHAIN_TURNS, OCX_STRESS_CHAIN_PAYLOAD_BYTES, + * OCX_STRESS_SESSIONS, OCX_STRESS_SESSION_PAYLOAD_BYTES, + * OCX_STRESS_SAMPLE_EVERY, OCX_STRESS_RSS_CAP_MB, OCX_STRESS_PARENT_TIMEOUT_MS + */ + +import { spawnSync } from "node:child_process"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { buildResponseJSON } from "../src/bridge"; +import { + clearResponseStateMemoryForTests, + expandPreviousResponseInput, + rememberResponseState, + responseStateMetrics, +} from "../src/responses/state"; + +const MODEL = "gpt-5.5"; +const MiB = 1024 * 1024; + +function envInt(name: string, fallback: number): number { + const raw = Number(process.env[name]); + return Number.isFinite(raw) && raw > 0 ? Math.floor(raw) : fallback; +} + +const CONFIG = { + chainTurns: envInt("OCX_STRESS_CHAIN_TURNS", 400), + chainPayloadBytes: envInt("OCX_STRESS_CHAIN_PAYLOAD_BYTES", 16 * 1024), + sessions: envInt("OCX_STRESS_SESSIONS", 1500), + sessionPayloadBytes: envInt("OCX_STRESS_SESSION_PAYLOAD_BYTES", 128 * 1024), + sampleEvery: envInt("OCX_STRESS_SAMPLE_EVERY", 50), + rssCapMb: envInt("OCX_STRESS_RSS_CAP_MB", 1200), + parentTimeoutMs: envInt("OCX_STRESS_PARENT_TIMEOUT_MS", 120_000), +}; + +// --------------------------------------------------------------------------- +// Shared helpers +// --------------------------------------------------------------------------- + +interface MemSample { + label: string; + rssMb: number; + heapUsedMb: number; + externalMb: number; + storeCount: number; + storeTotalMb: number; + storeLargestMb: number; +} + +function sample(label: string): MemSample { + const m = process.memoryUsage(); + const s = responseStateMetrics(); + return { + label, + rssMb: round(m.rss / MiB), + heapUsedMb: round(m.heapUsed / MiB), + externalMb: round(m.external / MiB), + storeCount: s.count, + storeTotalMb: round(s.totalBytes / MiB), + storeLargestMb: round(s.largestBytes / MiB), + }; +} + +function round(n: number): number { + return Math.round(n * 100) / 100; +} + +/** Best-effort force GC across runtimes: Bun.gc(true), then node --expose-gc global gc(). */ +function forceGc(): void { + const bun = (globalThis as { Bun?: { gc?: (sync: boolean) => void } }).Bun; + if (bun?.gc) bun.gc(true); + const g = (globalThis as { gc?: () => void }).gc; + if (g) g(); +} + +function overRssCap(): boolean { + return process.memoryUsage().rss / MiB > CONFIG.rssCapMb; +} + +function makePayload(bytes: number, tag: string): string { + // A recognizable prefix + filler; length dominates the serialized size. + const filler = "x".repeat(Math.max(0, bytes - tag.length)); + return `${tag}${filler}`; +} + +function buildTurnResponse(turnPayload: string): { id?: unknown; output?: unknown; status?: unknown } { + return buildResponseJSON( + [{ type: "text_delta", text: turnPayload }, { type: "done" }], + MODEL, + ) as { id?: unknown; output?: unknown; status?: unknown }; +} + +// --------------------------------------------------------------------------- +// Child role: run the workload, emit JSONL samples on stderr, SUMMARY on stdout +// --------------------------------------------------------------------------- + +interface ChildSummary { + label: string; + aborted: boolean; + abortReason?: string; + chain: { + turns: number; + heapUsedMb: number; + storeTotalMb: number; + approxUniqueContentMb: number; + // heapUsed / uniqueContent ~ 1 confirms sharing; storeTotal / heapUsed >> 1 confirms over-count. + heapToUniqueRatio: number; + storeToHeapRatio: number; + }; + sessions: { + created: number; + peakStoreCount: number; + peakStoreTotalMb: number; + peakRssMb: number; + }; + returnToOs: { + rssBeforeClearMb: number; + rssAfterClearGcMb: number; + reclaimedMb: number; + }; +} + +function emitSample(s: MemSample): void { + process.stderr.write(JSON.stringify({ kind: "sample", ...s }) + "\n"); +} + +function runChild(): void { + const label = process.env["OCX_STRESS_LABEL"] ?? "child"; + const summary: ChildSummary = { + label, + aborted: false, + chain: { + turns: 0, heapUsedMb: 0, storeTotalMb: 0, approxUniqueContentMb: 0, + heapToUniqueRatio: 0, storeToHeapRatio: 0, + }, + sessions: { created: 0, peakStoreCount: 0, peakStoreTotalMb: 0, peakRssMb: 0 }, + returnToOs: { rssBeforeClearMb: 0, rssAfterClearGcMb: 0, reclaimedMb: 0 }, + }; + + // --- Scenario A: one long chain (prefix reference sharing / C-type proof) --- + clearResponseStateMemoryForTests(); + forceGc(); + let prevId: string | undefined; + let turn = 0; + for (turn = 1; turn <= CONFIG.chainTurns; turn++) { + const payload = makePayload(CONFIG.chainPayloadBytes, `chain-t${turn}-`); + const body = prevId + ? { model: MODEL, previous_response_id: prevId, input: [{ role: "user", content: payload }], store: false } + : { model: MODEL, input: [{ role: "user", content: payload }], store: false }; + const expanded = expandPreviousResponseInput(body); + const resp = buildTurnResponse(payload); + rememberResponseState(expanded, resp, undefined, { force: true }); + prevId = resp.id as string; + if (turn % CONFIG.sampleEvery === 0) { + emitSample(sample(`chain@${turn}`)); + if (overRssCap()) { + summary.aborted = true; + summary.abortReason = `RSS cap ${CONFIG.rssCapMb}MB exceeded during chain at turn ${turn}`; + break; + } + } + } + forceGc(); + { + const s = sample(`chain-final@${turn}`); + emitSample(s); + // Unique content is created once per turn (input payload + output text of ~equal size), then + // shared by reference across every later entry: ~ 2 * turns * payload. + const approxUniqueContentMb = round((2 * (turn) * CONFIG.chainPayloadBytes) / MiB); + summary.chain = { + turns: turn, + heapUsedMb: s.heapUsedMb, + storeTotalMb: s.storeTotalMb, + approxUniqueContentMb, + heapToUniqueRatio: approxUniqueContentMb > 0 ? round(s.heapUsedMb / approxUniqueContentMb) : 0, + storeToHeapRatio: s.heapUsedMb > 0 ? round(s.storeTotalMb / s.heapUsedMb) : 0, + }; + } + + if (!summary.aborted) { + // --- Scenario B: many rotating independent sessions with large payloads --- + clearResponseStateMemoryForTests(); + forceGc(); + for (let i = 1; i <= CONFIG.sessions; i++) { + const payload = makePayload(CONFIG.sessionPayloadBytes, `sess-${i}-`); + const body = { model: MODEL, input: [{ role: "user", content: payload }], store: false }; + const resp = buildTurnResponse(payload); + rememberResponseState(body, resp, undefined, { force: true }); + summary.sessions.created = i; + if (i % CONFIG.sampleEvery === 0) { + const s = sample(`sessions@${i}`); + emitSample(s); + summary.sessions.peakStoreCount = Math.max(summary.sessions.peakStoreCount, s.storeCount); + summary.sessions.peakStoreTotalMb = Math.max(summary.sessions.peakStoreTotalMb, s.storeTotalMb); + summary.sessions.peakRssMb = Math.max(summary.sessions.peakRssMb, s.rssMb); + if (overRssCap()) { + summary.aborted = true; + summary.abortReason = `RSS cap ${CONFIG.rssCapMb}MB exceeded during sessions at ${i}`; + break; + } + } + } + } + + // --- Return-to-OS check: clear + GC, then see if RSS actually falls --- + const rssBeforeClear = round(process.memoryUsage().rss / MiB); + clearResponseStateMemoryForTests(); + forceGc(); + // Give the allocator a beat to purge (best-effort; synchronous script). + const spinUntil = Date.now() + 250; + while (Date.now() < spinUntil) { /* brief settle */ } + forceGc(); + const rssAfterClearGc = round(process.memoryUsage().rss / MiB); + summary.returnToOs = { + rssBeforeClearMb: rssBeforeClear, + rssAfterClearGcMb: rssAfterClearGc, + reclaimedMb: round(rssBeforeClear - rssAfterClearGc), + }; + + process.stdout.write("SUMMARY " + JSON.stringify(summary) + "\n"); +} + +// --------------------------------------------------------------------------- +// Parent role: orchestrate the A/B, enforce timeout, print a comparison table +// --------------------------------------------------------------------------- + +interface Overlay { label: string; env: Record; } + +function runParent(): void { + const home = mkdtempSync(join(tmpdir(), "ocx-stress-")); + // A/B overlays. The mimalloc knobs are best-effort: if Bun's bundled mimalloc ignores them the + // two rows will simply match, which is itself informative (rules the lever out). + const overlays: Overlay[] = [ + { label: "bun-default", env: {} }, + { label: "mimalloc-purge0", env: { MIMALLOC_PURGE_DELAY: "0", MIMALLOC_PAGE_RESET: "1" } }, + ]; + + console.log("== OpenCodex memory-stress harness =="); + console.log(`bun: ${process.versions?.bun ?? "?"} node-compat: ${process.version} platform: ${process.platform}`); + console.log("config:", JSON.stringify(CONFIG)); + console.log(`isolated OPENCODEX_HOME: ${home}`); + console.log(""); + + const summaries: ChildSummary[] = []; + for (const overlay of overlays) { + process.stdout.write(`-- running scenario [${overlay.label}] ...\n`); + const res = spawnSync( + process.execPath, + [__filename, "--child"], + { + env: { + ...process.env, + ...overlay.env, + OPENCODEX_HOME: home, + OCX_STRESS_LABEL: overlay.label, + }, + timeout: CONFIG.parentTimeoutMs, + killSignal: "SIGKILL", + encoding: "utf8", + maxBuffer: 64 * MiB, + }, + ); + if (res.error) { + console.log(` [${overlay.label}] spawn error: ${res.error.message}`); + continue; + } + if (res.signal) { + console.log(` [${overlay.label}] killed by ${res.signal} (likely parent timeout / safety)`); + } + const summary = parseSummary(res.stdout ?? ""); + if (summary) summaries.push(summary); + else console.log(` [${overlay.label}] no SUMMARY produced. stderr tail:\n${tail(res.stderr ?? "", 5)}`); + } + + rmSync(home, { recursive: true, force: true }); + printReport(summaries); +} + +function parseSummary(stdout: string): ChildSummary | null { + for (const line of stdout.split("\n")) { + if (line.startsWith("SUMMARY ")) { + try { return JSON.parse(line.slice("SUMMARY ".length)) as ChildSummary; } catch { return null; } + } + } + return null; +} + +function tail(text: string, lines: number): string { + return text.split("\n").filter(Boolean).slice(-lines).join("\n"); +} + +function printReport(summaries: ChildSummary[]): void { + console.log("\n================= RESULTS =================\n"); + for (const s of summaries) { + console.log(`### scenario: ${s.label}${s.aborted ? " (ABORTED: " + s.abortReason + ")" : ""}`); + console.log(" chain (prefix-sharing / C-type):"); + console.log(` turns=${s.chain.turns} heapUsed=${s.chain.heapUsedMb}MB storeTotal=${s.chain.storeTotalMb}MB approxUniqueContent=${s.chain.approxUniqueContentMb}MB`); + console.log(` heapUsed/uniqueContent=${s.chain.heapToUniqueRatio} (~1 => content shared by reference)`); + console.log(` storeTotal/heapUsed=${s.chain.storeToHeapRatio} (>>1 => serialized metric over-counts shared prefixes)`); + console.log(" sessions (count-only cap headroom):"); + console.log(` created=${s.sessions.created} peakStoreCount=${s.sessions.peakStoreCount} peakStoreTotal=${s.sessions.peakStoreTotalMb}MB peakRss=${s.sessions.peakRssMb}MB`); + console.log(" return-to-OS (native-retention signal):"); + console.log(` rssBeforeClear=${s.returnToOs.rssBeforeClearMb}MB rssAfterClear+GC=${s.returnToOs.rssAfterClearGcMb}MB reclaimed=${s.returnToOs.reclaimedMb}MB`); + console.log(""); + } + if (summaries.length >= 2) { + const [a, b] = summaries; + console.log("### A/B (allocator lever)"); + console.log(` ${a.label} peakRss=${a.sessions.peakRssMb}MB reclaimed=${a.returnToOs.reclaimedMb}MB`); + console.log(` ${b.label} peakRss=${b.sessions.peakRssMb}MB reclaimed=${b.returnToOs.reclaimedMb}MB`); + console.log(" (if reclaimed is small in BOTH, RSS is retained after JS release => hypothesis B / native allocator)"); + } + console.log("\n========================================="); + console.log("Interpretation guide:"); + console.log(" - heapUsed/uniqueContent ~ 1 => items are shared by reference; a delta-chain refactor buys little."); + console.log(" - storeTotal/heapUsed >> 1 => the JSON-serialized byte metric over-counts (upper bound), as designed."); + console.log(" - sessions peakRss high with peakStoreCount pinned at the count cap => a BYTE budget would help (hygiene)."); + console.log(" - reclaimed ~ 0 after clear+GC => native retention dominates => the fix is a watchdog + preemptive restart."); +} + +// --------------------------------------------------------------------------- + +if (process.argv.includes("--child")) { + runChild(); +} else { + runParent(); +} diff --git a/src/lib/memory-usage.ts b/src/lib/memory-usage.ts new file mode 100644 index 00000000000..743586f073f --- /dev/null +++ b/src/lib/memory-usage.ts @@ -0,0 +1,175 @@ +/** + * Portable process-memory snapshot for the memory watchdog. + * + * WHY: OpenCodex on Bun can hit the "committed memory >> working set" signature (field report: + * bun.exe private/committed ~79 GB while the working set stayed ~5.8 GB, freezing the box once the + * system commit charge was exhausted). The Bun/mimalloc native allocator retains committed memory + * the JS heap has already released — proven locally by the return-to-OS stress harness and mirrored + * upstream by Claude Code #36132. A watchdog that only reads RSS would MISS that case entirely, so + * we read the platform's private/committed counter where it is cheaply available and fall back to + * RSS elsewhere. + * + * SAFETY: the only platform that needs a child process to read committed bytes is Windows; that + * probe is guarded exactly like the icacls hardening path (hard timeout, soft-fail, no shell + * string) so a slow/hung query can never block the proxy. Linux reads /proc (no spawn); every other + * platform uses the in-process RSS fallback. All thresholds that consume this snapshot are relative + * to totalSystemBytes so the same code behaves identically on a 16 GB laptop and a 128 GB worker. + */ + +import { readFileSync } from "node:fs"; +import { platform as osPlatform, totalmem } from "node:os"; + +export type CommittedSource = "windows-private" | "proc-status" | "rss-fallback" | "none"; + +export interface MemorySnapshot { + /** Resident set size (physically mapped) in bytes. */ + rssBytes: number; + /** V8/JSC heap in use, bytes. */ + heapUsedBytes: number; + /** Off-heap native allocations tracked by the runtime, bytes. */ + externalBytes: number; + /** + * Private/committed bytes — the counter that actually blows up in the field case. `null` when the + * platform has no cheap probe (callers fall back to rssBytes for pressure math). + */ + committedBytes: number | null; + /** Where committedBytes came from (diagnostics; "none" when unavailable). */ + committedSource: CommittedSource; + /** Total physical RAM, bytes — the denominator for relative pressure thresholds. */ + totalSystemBytes: number; +} + +/** Total icacls-style budget for the Windows committed probe. A slow query soft-fails to RSS. */ +const WINDOWS_PROBE_TIMEOUT_MS = 2_000; + +type PlatformOverride = string | null; +let platformOverride: PlatformOverride = null; + +/** Test seam: force the platform gate (e.g. "linux") so probe selection is deterministic. */ +export function setMemoryPlatformForTests(value: PlatformOverride): void { + platformOverride = value; +} + +function effectivePlatform(): string { + return platformOverride ?? osPlatform(); +} + +// --------------------------------------------------------------------------- +// Windows private-bytes probe (guarded child process) +// --------------------------------------------------------------------------- + +export interface WindowsProbeResult { + privateBytes: number | null; + timedOut: boolean; +} + +type WindowsProbeRunner = (pid: number, timeoutMs: number) => WindowsProbeResult; + +function defaultWindowsProbeRunner(pid: number, timeoutMs: number): WindowsProbeResult { + // .NET Process.PrivateMemorySize64 == Windows private (committed) bytes for this PID. Arguments + // are passed as an array (no shell string) and the window is hidden, matching the icacls runner. + const spawnSync = (globalThis as { Bun?: { spawnSync?: typeof import("bun").spawnSync } }).Bun?.spawnSync; + if (!spawnSync) return { privateBytes: null, timedOut: false }; + const result = spawnSync( + [ + "powershell.exe", + "-NoProfile", + "-NonInteractive", + "-Command", + `(Get-Process -Id ${pid}).PrivateMemorySize64`, + ], + { stdin: "ignore", stdout: "pipe", stderr: "ignore", timeout: timeoutMs, windowsHide: true }, + ); + if (result.exitedDueToTimeout) return { privateBytes: null, timedOut: true }; + if (!result.success) return { privateBytes: null, timedOut: false }; + const text = result.stdout ? result.stdout.toString().trim() : ""; + const value = Number(text); + return { privateBytes: Number.isFinite(value) && value > 0 ? value : null, timedOut: false }; +} + +let windowsProbeRunner: WindowsProbeRunner = defaultWindowsProbeRunner; + +/** Test seam: replace the Windows probe runner. Pass null to restore the default. */ +export function setWindowsProbeRunnerForTests(runner: WindowsProbeRunner | null): void { + windowsProbeRunner = runner ?? defaultWindowsProbeRunner; +} + +// --------------------------------------------------------------------------- +// Linux /proc probe (no spawn) +// --------------------------------------------------------------------------- + +type ProcStatusReader = () => string | null; + +function defaultProcStatusReader(): string | null { + try { + return readFileSync("/proc/self/status", "utf8"); + } catch { + return null; + } +} + +let procStatusReader: ProcStatusReader = defaultProcStatusReader; + +/** Test seam: replace the /proc/self/status reader. Pass null to restore the default. */ +export function setProcStatusReaderForTests(reader: ProcStatusReader | null): void { + procStatusReader = reader ?? defaultProcStatusReader; +} + +/** Parse VmRSS + VmSwap (kB) out of /proc/self/status → private committed bytes, or null. */ +export function parseProcStatusCommitted(text: string): number | null { + const field = (name: string): number | null => { + const match = new RegExp(`^${name}:\\s*(\\d+)\\s*kB`, "m").exec(text); + return match ? Number(match[1]) * 1024 : null; + }; + const rss = field("VmRSS"); + if (rss === null) return null; + const swap = field("VmSwap") ?? 0; + return rss + swap; +} + +// --------------------------------------------------------------------------- +// Snapshot assembly +// --------------------------------------------------------------------------- + +function committedForPlatform(): { bytes: number | null; source: CommittedSource } { + const plat = effectivePlatform(); + if (plat === "win32") { + const result = windowsProbeRunner(process.pid, WINDOWS_PROBE_TIMEOUT_MS); + if (result.privateBytes !== null) return { bytes: result.privateBytes, source: "windows-private" }; + return { bytes: null, source: "none" }; + } + if (plat === "linux") { + const text = procStatusReader(); + const committed = text ? parseProcStatusCommitted(text) : null; + if (committed !== null) return { bytes: committed, source: "proc-status" }; + return { bytes: null, source: "none" }; + } + return { bytes: null, source: "none" }; +} + +/** + * Read a single portable memory snapshot. Never throws: probe failures degrade to committedBytes + * = null so the caller falls back to RSS. + */ +export function readMemorySnapshot(): MemorySnapshot { + const usage = process.memoryUsage(); + const committed = committedForPlatform(); + return { + rssBytes: usage.rss, + heapUsedBytes: usage.heapUsed, + externalBytes: usage.external, + committedBytes: committed.bytes, + committedSource: committed.source, + totalSystemBytes: totalmem(), + }; +} + +/** The bytes the watchdog treats as memory pressure: committed where known, else RSS. */ +export function pressureBytes(snapshot: MemorySnapshot): number { + return snapshot.committedBytes ?? snapshot.rssBytes; +} + +/** The source label after the RSS fallback is applied (for diagnostics). */ +export function pressureSource(snapshot: MemorySnapshot): CommittedSource { + return snapshot.committedBytes !== null ? snapshot.committedSource : "rss-fallback"; +} diff --git a/src/responses/state.ts b/src/responses/state.ts index ff4a8402454..d3de71be820 100644 --- a/src/responses/state.ts +++ b/src/responses/state.ts @@ -179,6 +179,50 @@ export function previousResponseProviderState(responseId: string | undefined): O return providers ? structuredClone(providers) : undefined; } +export interface ResponseStateMetrics { + /** Number of retained continuation entries currently in RAM. */ + count: number; + /** Sum of serialized item bytes across all entries. Because expanded previous_response_id chains + * share prefix item references, this is an UPPER bound on true heap (it re-counts shared history), + * never an under-count — safe to use as a memory-pressure signal. */ + totalBytes: number; + /** Serialized item bytes of the single largest entry (a long chain's head, or an image-heavy turn). */ + largestBytes: number; + /** Age of the oldest retained entry, in ms (0 when the store is empty). */ + oldestAgeMs: number; +} + +/** + * Observe-only snapshot of the in-RAM continuation store. Additive and side-effect free — it does + * NOT lazy-load the disk snapshot, prune, or evict — so a diagnostics endpoint or benchmark harness + * can sample it without perturbing request handling. This is the instrumentation seam for deciding + * whether RAM growth originates in this store (JS heap) or in the runtime allocator (native). + */ +export function responseStateMetrics(): ResponseStateMetrics { + const at = now(); + let totalBytes = 0; + let largestBytes = 0; + let oldestCreatedAt = at; + for (const state of states.values()) { + let bytes = 0; + try { + bytes = JSON.stringify(state.items).length; + if (state.providers) bytes += JSON.stringify(state.providers).length; + } catch { + /* unserializable entry (should not happen): count as 0 rather than throw in a metrics path */ + } + totalBytes += bytes; + if (bytes > largestBytes) largestBytes = bytes; + if (state.createdAt < oldestCreatedAt) oldestCreatedAt = state.createdAt; + } + return { + count: states.size, + totalBytes, + largestBytes, + oldestAgeMs: states.size > 0 ? at - oldestCreatedAt : 0, + }; +} + export function rememberResponseState( requestBody: unknown, response: { id?: unknown; output?: unknown; status?: unknown }, diff --git a/src/server/index.ts b/src/server/index.ts index c79896d1c61..7b895e7247c 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -114,6 +114,8 @@ export { jsonResponse, safeConfigDTO, } from "./auth-cors"; +import { startMemoryWatchdog } from "./memory-watchdog"; +export { memoryWatchdogSnapshot } from "./memory-watchdog"; import { disableResponsesRequestTimeout, handleResponses, handleResponsesCompact } from "./responses"; export { disableResponsesRequestTimeout, linkAbortSignal } from "./responses"; import { handleClaudeCountTokens, handleClaudeMessages } from "./claude-messages"; @@ -641,6 +643,11 @@ export function startServer(port?: number) { const actualPort = server.port ?? listenPort; setCorsOrigin(actualPort); + // Memory watchdog: warn (or, opt-in, gracefully restart) before the Bun/mimalloc native allocator + // can exhaust the system commit charge. Warn-only by default; the interval is unref'd so it never + // holds the process open on its own. Never throws — a watchdog failure must not block startup. + try { startMemoryWatchdog(config); } catch { /* observability is best-effort */ } + console.log(`🚀 opencodex proxy running on http://localhost:${actualPort}`); console.log(` POST /v1/responses → provider translation`); console.log(` POST /v1/chat/completions → OpenAI-compatible clients`); diff --git a/src/server/management-api.ts b/src/server/management-api.ts index 609d3dba139..49a22ef6d04 100644 --- a/src/server/management-api.ts +++ b/src/server/management-api.ts @@ -50,6 +50,12 @@ import { } from "../lib/debug-settings"; import type { OcxClaudeCodeConfig, OcxConfig, OcxCustomModel, OcxProviderConfig } from "../types"; import { drainAndShutdown } from "./lifecycle"; +import { + applyWatchdogRuntimeConfig, + memoryWatchdogReport, + startMemoryWatchdog, + stopMemoryWatchdog, +} from "./memory-watchdog"; import { filterRequestLogs, getRequestLogEntries, type RequestLogEntry } from "./request-log"; import { estimateComboCost, estimateRequestCost, normalizeCostTokens, tokensPerSecond } from "../usage/cost"; import type { PersistedUsageAttempt } from "../usage/log"; @@ -249,6 +255,47 @@ export async function handleManagementAPI(req: Request, url: URL, config: OcxCon return jsonResponse({ ok: true, codexAutoStart: codexAutoStartEnabled(config) }); } + if (url.pathname === "/api/memory" && req.method === "GET") { + // Read-only Monitor + Recommend report. Degrades to { enabled: false } on disabled/old servers. + return jsonResponse(memoryWatchdogReport() ?? { enabled: false }); + } + + if (url.pathname === "/api/memory/settings" && req.method === "PUT") { + let raw: unknown; + try { raw = await req.json(); } catch { return jsonResponse({ error: "invalid JSON body" }, 400); } + if (!isPlainRecord(raw)) return jsonResponse({ error: "body must be a JSON object" }, 400); + const b = raw as Record; + const isBool = (v: unknown) => typeof v === "boolean"; + const isPosNum = (v: unknown) => typeof v === "number" && Number.isFinite(v) && v > 0; + const isFrac = (v: unknown) => typeof v === "number" && Number.isFinite(v) && v > 0 && v < 1; + if (b.enabled !== undefined && !isBool(b.enabled)) return jsonResponse({ error: "enabled must be boolean" }, 400); + if (b.autoRestart !== undefined && !isBool(b.autoRestart)) return jsonResponse({ error: "autoRestart must be boolean" }, 400); + if (b.requireSupervisor !== undefined && !isBool(b.requireSupervisor)) return jsonResponse({ error: "requireSupervisor must be boolean" }, 400); + if (b.intervalMs !== undefined && !isPosNum(b.intervalMs)) return jsonResponse({ error: "intervalMs must be a positive number" }, 400); + if (b.warnFraction !== undefined && !isFrac(b.warnFraction)) return jsonResponse({ error: "warnFraction must be a number in (0,1)" }, 400); + if (b.criticalFraction !== undefined && !isFrac(b.criticalFraction)) return jsonResponse({ error: "criticalFraction must be a number in (0,1)" }, 400); + + const patch: NonNullable = {}; + if (b.enabled !== undefined) patch.enabled = b.enabled as boolean; + if (b.intervalMs !== undefined) patch.intervalMs = b.intervalMs as number; + if (b.warnFraction !== undefined) patch.warnFraction = b.warnFraction as number; + if (b.criticalFraction !== undefined) patch.criticalFraction = b.criticalFraction as number; + if (b.autoRestart !== undefined) patch.autoRestart = b.autoRestart as boolean; + if (b.requireSupervisor !== undefined) patch.requireSupervisor = b.requireSupervisor as boolean; + + config.memoryWatchdog = { ...config.memoryWatchdog, ...patch }; + saveConfig(config); + + // Apply live so the change takes effect without a proxy restart. + if (patch.enabled === false) { + stopMemoryWatchdog(); + } else if (applyWatchdogRuntimeConfig(patch) === null) { + // Not currently running (was disabled / first enable) — (re)start from the saved config. + startMemoryWatchdog(config); + } + return jsonResponse({ ok: true, memoryWatchdog: config.memoryWatchdog, report: memoryWatchdogReport() ?? { enabled: false } }); + } + if (url.pathname === "/api/diagnostics/project-config" && req.method === "GET") { const { getCachedProjectConfigDiagnostics } = await import("../codex/project-config-warnings"); const { warnings, grouped } = getCachedProjectConfigDiagnostics(); diff --git a/src/server/memory-watchdog.ts b/src/server/memory-watchdog.ts new file mode 100644 index 00000000000..a7a036f63f7 --- /dev/null +++ b/src/server/memory-watchdog.ts @@ -0,0 +1,456 @@ +/** + * Memory watchdog: observe + warn by default, opt-in graceful restart. + * + * WHY: the Bun/mimalloc native allocator can retain committed memory the JS heap already released + * (field: private/committed ~79 GB, working set ~5.8 GB, box frozen at the system commit limit; + * mirrored by Claude Code #36132). No in-app JS cache cap fixes native retention — the only proven + * mitigation is to notice the pressure and, optionally, hand off to the already-tested graceful + * restart path. This module NEVER decides thresholds in absolute GB: every trigger is a fraction of + * total system RAM, so one code path behaves the same on a 16 GB laptop and a 128 GB worker. + * + * SAFETY MODEL: + * - Default action is WARN ONLY. It cannot lose data or kill the process. + * - Auto-restart is opt-in (config.memoryWatchdog.autoRestart / env). When it fires it reuses + * drainAndShutdown() — which drains in-flight turns and flushes the response-state snapshot — + * then exits with a distinct code so a supervisor can respawn. + * - A cooldown (minRestartIntervalMs) and a max-restart guard prevent restart loops. + * - The decision core (evaluate) is pure and injectable (clock + reader + restart hook) so the + * logic is unit-tested without real timers or a real process exit. + */ + +import { pressureBytes, pressureSource, readMemorySnapshot, type MemorySnapshot } from "../lib/memory-usage"; +import { drainAndShutdown } from "./lifecycle"; +import type { OcxConfig } from "../types"; + +/** Exit code used for an intentional memory-driven restart (distinct from crash/normal exit). */ +export const MEMORY_RESTART_EXIT_CODE = 75; // EX_TEMPFAIL: "transient, retry" — a supervisor should respawn. + +const MiB = 1024 * 1024; + +export interface ResolvedWatchdogConfig { + enabled: boolean; + intervalMs: number; + warnFraction: number; + criticalFraction: number; + autoRestart: boolean; + /** When true, auto-restart only fires if a supervisor is detected (so we never exit into "just dead"). */ + requireSupervisor: boolean; + minRestartIntervalMs: number; + maxRestarts: number; + growthWindowMs: number; +} + +const DEFAULTS: ResolvedWatchdogConfig = { + enabled: true, + intervalMs: 60_000, + warnFraction: 0.60, + criticalFraction: 0.75, + autoRestart: false, + requireSupervisor: true, + minRestartIntervalMs: 600_000, // 10 min + maxRestarts: 3, + growthWindowMs: 600_000, // 10 min growth-rate window (diagnostic) +}; + +/** Parse a boolean-ish string; undefined when unset/unrecognized. */ +function parseFlagValue(raw: string | undefined): boolean | undefined { + const v = raw?.trim().toLowerCase(); + if (v === undefined || v === "") return undefined; + if (v === "1" || v === "true" || v === "yes" || v === "on") return true; + if (v === "0" || v === "false" || v === "no" || v === "off") return false; + return undefined; +} + +function envFlag(name: string): boolean | undefined { + return parseFlagValue(process.env[name]); +} + +function envNum(name: string): number | undefined { + const raw = process.env[name]?.trim(); + if (!raw) return undefined; + const value = Number(raw); + return Number.isFinite(value) && value > 0 ? value : undefined; +} + +function clampFraction(value: number | undefined, fallback: number): number { + if (value === undefined || !Number.isFinite(value)) return fallback; + return Math.min(0.99, Math.max(0.10, value)); +} + +/** + * Resolve config → runtime knobs. Precedence: env override > config file > default. Fractions are + * clamped to [0.10, 0.99]; a criticalFraction <= warnFraction is nudged above warn so the two + * levels never collapse. + */ +export function resolveWatchdogConfig(config: OcxConfig): ResolvedWatchdogConfig { + const c = config.memoryWatchdog ?? {}; + const enabled = envFlag("OCX_MEMORY_WATCHDOG_DISABLED") === true + ? false + : (envFlag("OCX_MEMORY_WATCHDOG_ENABLED") ?? c.enabled ?? DEFAULTS.enabled); + + const warnFraction = clampFraction( + envNum("OCX_MEMORY_WATCHDOG_WARN_FRACTION") ?? c.warnFraction, + DEFAULTS.warnFraction, + ); + let criticalFraction = clampFraction( + envNum("OCX_MEMORY_WATCHDOG_CRITICAL_FRACTION") ?? c.criticalFraction, + DEFAULTS.criticalFraction, + ); + if (criticalFraction <= warnFraction) criticalFraction = Math.min(0.99, warnFraction + 0.10); + + return { + enabled, + intervalMs: envNum("OCX_MEMORY_WATCHDOG_INTERVAL_MS") ?? c.intervalMs ?? DEFAULTS.intervalMs, + warnFraction, + criticalFraction, + autoRestart: envFlag("OCX_MEMORY_WATCHDOG_AUTO_RESTART") ?? c.autoRestart ?? DEFAULTS.autoRestart, + requireSupervisor: + envFlag("OCX_MEMORY_WATCHDOG_REQUIRE_SUPERVISOR") ?? c.requireSupervisor ?? DEFAULTS.requireSupervisor, + minRestartIntervalMs: + envNum("OCX_MEMORY_WATCHDOG_MIN_RESTART_INTERVAL_MS") ?? c.minRestartIntervalMs ?? DEFAULTS.minRestartIntervalMs, + maxRestarts: envNum("OCX_MEMORY_WATCHDOG_MAX_RESTARTS") ?? c.maxRestarts ?? DEFAULTS.maxRestarts, + growthWindowMs: DEFAULTS.growthWindowMs, + }; +} + +// --------------------------------------------------------------------------- +// Supervisor detection +// --------------------------------------------------------------------------- + +/** + * Best-effort detection of an external process supervisor that will respawn us after an intentional + * exit. Pure over an env map so it is unit-testable. Heuristics: explicit OCX_SUPERVISED flag, pm2 + * (pm_id / PM2_HOME), systemd (INVOCATION_ID / NOTIFY_SOCKET). When nothing matches we assume NO + * supervisor — the safe default, because exiting without a respawner is worse than staying up. + */ +export function detectSupervisor( + env: Record = process.env, +): { supervised: boolean; hint: string } { + const explicit = parseFlagValue(env.OCX_SUPERVISED); + if (explicit === true) return { supervised: true, hint: "OCX_SUPERVISED" }; + if (env.pm_id !== undefined || (env.PM2_HOME ?? "") !== "") return { supervised: true, hint: "pm2" }; + if ((env.INVOCATION_ID ?? "") !== "" || (env.NOTIFY_SOCKET ?? "") !== "") return { supervised: true, hint: "systemd" }; + if (explicit === false) return { supervised: false, hint: "OCX_SUPERVISED=off" }; + return { supervised: false, hint: "none" }; +} + +// --------------------------------------------------------------------------- +// Pure decision core +// --------------------------------------------------------------------------- + +export type WatchdogLevel = "ok" | "warn" | "critical"; +export type WatchdogAction = "none" | "warn" | "restart"; + +export interface WatchdogState { + samples: { t: number; bytes: number; fraction: number }[]; + /** Highest level already reported since the last drop back to ok — prevents per-interval log spam. */ + reportedLevel: WatchdogLevel; + restartCount: number; + lastRestartAt: number; +} + +export interface WatchdogDecision { + level: WatchdogLevel; + action: WatchdogAction; + fraction: number; + pressureMb: number; + totalMb: number; + source: string; + growthMbPerHour: number | null; + reason: string; +} + +export function createWatchdogState(): WatchdogState { + return { samples: [], reportedLevel: "ok", restartCount: 0, lastRestartAt: 0 }; +} + +function levelFor(fraction: number, cfg: ResolvedWatchdogConfig): WatchdogLevel { + if (fraction >= cfg.criticalFraction) return "critical"; + if (fraction >= cfg.warnFraction) return "warn"; + return "ok"; +} + +/** bytes/hour over the retained window, or null when there is not enough spread to be meaningful. */ +function growthBytesPerHour(state: WatchdogState): number | null { + if (state.samples.length < 2) return null; + const first = state.samples[0]!; + const last = state.samples[state.samples.length - 1]!; + const dtMs = last.t - first.t; + if (dtMs <= 0) return null; + return ((last.bytes - first.bytes) / dtMs) * 3_600_000; +} + +/** + * Pure evaluation of one sample. Mutates `state` (sample ring, report latch, restart bookkeeping) + * and returns the decision. Emits action "restart" only when level is critical AND auto-restart is + * enabled AND the cooldown + max-restart guards allow it; otherwise a critical level degrades to a + * "warn" action (loud log, no restart). + */ +export function evaluate( + state: WatchdogState, + snapshot: MemorySnapshot, + cfg: ResolvedWatchdogConfig, + nowMs: number, + supervised = true, +): WatchdogDecision { + const bytes = pressureBytes(snapshot); + const total = snapshot.totalSystemBytes > 0 ? snapshot.totalSystemBytes : bytes; + const fraction = total > 0 ? bytes / total : 0; + + state.samples.push({ t: nowMs, bytes, fraction }); + const cutoff = nowMs - cfg.growthWindowMs; + while (state.samples.length > 2 && state.samples[0]!.t < cutoff) state.samples.shift(); + + const level = levelFor(fraction, cfg); + const growth = growthBytesPerHour(state); + + let action: WatchdogAction = "none"; + let reason = ""; + + if (level === "ok") { + state.reportedLevel = "ok"; // recovered — re-arm warnings for the next crossing + } else if (level === "warn") { + if (state.reportedLevel === "ok") { action = "warn"; reason = "crossed warn fraction"; } + state.reportedLevel = state.reportedLevel === "critical" ? "critical" : "warn"; + } else { + // critical + const cooledDown = nowMs - state.lastRestartAt >= cfg.minRestartIntervalMs || state.lastRestartAt === 0; + const underCap = state.restartCount < cfg.maxRestarts; + const supervisorOk = !cfg.requireSupervisor || supervised; + if (cfg.autoRestart && supervisorOk && cooledDown && underCap) { + action = "restart"; + reason = "crossed critical fraction; auto-restart armed"; + state.restartCount += 1; + state.lastRestartAt = nowMs; + } else { + if (state.reportedLevel !== "critical") { action = "warn"; } + reason = !cfg.autoRestart + ? "critical; auto-restart disabled" + : !supervisorOk + ? "critical; auto-restart suppressed: no supervisor" + : underCap + ? "critical; restart deferred by cooldown" + : "critical; max-restart guard reached"; + } + state.reportedLevel = "critical"; + } + + return { + level, + action, + fraction, + pressureMb: Math.round(bytes / MiB), + totalMb: Math.round(total / MiB), + source: pressureSource(snapshot), + growthMbPerHour: growth === null ? null : Math.round(growth / MiB), + reason, + }; +} + +// --------------------------------------------------------------------------- +// Recommendation (advisory only — never mutates config or state) +// --------------------------------------------------------------------------- + +export interface WatchdogRecommendation { + warnFraction: number; + criticalFraction: number; + autoRestart: boolean; + rationale: string; +} + +/** + * Suggest thresholds from the observed steady-state. Advisory only: it computes numbers a human (or + * the Adjust UI) can choose to apply, and never touches config or running state. Warn is placed a + * margin above the observed peak fraction so it does not false-fire on normal usage; critical sits a + * further margin above warn. autoRestart is only suggested when a supervisor exists AND memory is + * trending up (a flat process does not benefit from restarts). With < 2 samples it echoes the + * current config and says so. + */ +export function recommend( + state: WatchdogState, + cfg: ResolvedWatchdogConfig, + supervised: boolean, +): WatchdogRecommendation { + if (state.samples.length < 2) { + return { + warnFraction: cfg.warnFraction, + criticalFraction: cfg.criticalFraction, + autoRestart: cfg.autoRestart, + rationale: "insufficient samples; keeping current thresholds", + }; + } + const observedMax = Math.max(...state.samples.map(s => s.fraction)); + const clamp = (v: number) => Math.min(0.99, Math.max(0.10, v)); + const warnFraction = clamp(observedMax + 0.10); + const criticalFraction = clamp(Math.max(warnFraction + 0.10, observedMax + 0.20)); + const growth = growthBytesPerHour(state); + const trendingUp = growth !== null && growth > 0; + const autoRestart = supervised && trendingUp; + const growthMbH = growth === null ? "n/a" : `${Math.round(growth / MiB)}MB/h`; + const rationale = + `observed peak ${(observedMax * 100).toFixed(1)}% of RAM over ${state.samples.length} samples; ` + + `growth ${growthMbH}; supervisor ${supervised ? "detected" : "not detected"} ` + + `→ autoRestart ${autoRestart ? "suggested" : "not suggested"}`; + return { warnFraction, criticalFraction, autoRestart, rationale }; +} + +// --------------------------------------------------------------------------- +// Runtime driver (timer + logging + restart hook) +// --------------------------------------------------------------------------- + +export interface WatchdogDeps { + now: () => number; + read: () => MemorySnapshot; + /** Whether an external supervisor will respawn us; gates auto-restart when requireSupervisor is on. */ + supervised: boolean; + /** Perform the graceful restart. Default: drain + flush snapshot, then exit with the restart code. */ + restart: (cfg: ResolvedWatchdogConfig) => void | Promise; + log: (line: string) => void; +} + +function defaultRestart(cfg: ResolvedWatchdogConfig): void { + void (async () => { + try { + await drainAndShutdown(undefined, 5_000); + } finally { + process.exit(MEMORY_RESTART_EXIT_CODE); + } + })(); + // Reference cfg so the signature stays honest for tests overriding it. + void cfg; +} + +const DEFAULT_DEPS: WatchdogDeps = { + now: Date.now, + read: readMemorySnapshot, + supervised: detectSupervisor().supervised, + restart: defaultRestart, + log: line => console.warn(line), +}; + +function formatLine(d: WatchdogDecision): string { + const pct = (d.fraction * 100).toFixed(1); + const growth = d.growthMbPerHour === null ? "n/a" : `${d.growthMbPerHour}MB/h`; + return `[opencodex] memory watchdog ${d.level.toUpperCase()}: ${d.pressureMb}MB / ${d.totalMb}MB (${pct}% of RAM, source=${d.source}, growth=${growth}) — ${d.reason}`; +} + +interface RunningWatchdog { + timer: ReturnType; + state: WatchdogState; + last: WatchdogDecision | null; + cfg: ResolvedWatchdogConfig; + deps: WatchdogDeps; +} + +let running: RunningWatchdog | null = null; + +/** One sampling tick: read → evaluate → act. Exported for deterministic tests. */ +export function tick(state: WatchdogState, cfg: ResolvedWatchdogConfig, deps: WatchdogDeps): WatchdogDecision { + const snapshot = deps.read(); + const decision = evaluate(state, snapshot, cfg, deps.now(), deps.supervised); + if (decision.action === "warn") { + deps.log(formatLine(decision)); + } else if (decision.action === "restart") { + deps.log(formatLine(decision)); + void deps.restart(cfg); + } + return decision; +} + +/** + * Start the watchdog. No-op (returns false) when disabled. Idempotent: a second call replaces the + * previous timer. The interval is unref'd so it never keeps the event loop alive on its own. + */ +export function startMemoryWatchdog(config: OcxConfig, deps: Partial = {}): boolean { + const cfg = resolveWatchdogConfig(config); + if (!cfg.enabled) return false; + stopMemoryWatchdog(); + const d: WatchdogDeps = { ...DEFAULT_DEPS, ...deps }; + const state = createWatchdogState(); + const timer = setInterval(() => { + try { + running!.last = tick(running!.state, running!.cfg, running!.deps); + } catch { + /* a sampling failure must never crash the proxy */ + } + }, cfg.intervalMs); + (timer as { unref?: () => void }).unref?.(); + running = { timer, state, last: null, cfg, deps: d }; + return true; +} + +/** + * Live-update knobs on a running watchdog without a restart (Adjust UI / management API). Fractions + * are clamped and critical is kept above warn, mirroring resolveWatchdogConfig. An intervalMs change + * re-arms the timer. Returns the applied config, or null when the watchdog is not running. + */ +export function applyWatchdogRuntimeConfig( + partial: Partial, +): ResolvedWatchdogConfig | null { + if (!running) return null; + const next: ResolvedWatchdogConfig = { ...running.cfg, ...partial }; + next.warnFraction = clampFraction(next.warnFraction, DEFAULTS.warnFraction); + next.criticalFraction = clampFraction(next.criticalFraction, DEFAULTS.criticalFraction); + if (next.criticalFraction <= next.warnFraction) { + next.criticalFraction = Math.min(0.99, next.warnFraction + 0.10); + } + const intervalChanged = next.intervalMs !== running.cfg.intervalMs && next.intervalMs > 0; + running.cfg = next; + if (intervalChanged) { + clearInterval(running.timer); + const timer = setInterval(() => { + try { + running!.last = tick(running!.state, running!.cfg, running!.deps); + } catch { + /* a sampling failure must never crash the proxy */ + } + }, next.intervalMs); + (timer as { unref?: () => void }).unref?.(); + running.timer = timer; + } + return next; +} + +/** Stop the watchdog timer (graceful shutdown / tests). Idempotent. */ +export function stopMemoryWatchdog(): void { + if (!running) return; + clearInterval(running.timer); + running = null; +} + +/** Last decision + guard state, for the management API / diagnostics. Null when never sampled. */ +export function memoryWatchdogSnapshot(): (WatchdogDecision & { restartCount: number }) | null { + if (!running || !running.last) return null; + return { ...running.last, restartCount: running.state.restartCount }; +} + +export interface WatchdogReport { + enabled: boolean; + decision: WatchdogDecision | null; + samplesCount: number; + growthMbPerHour: number | null; + resolvedConfig: ResolvedWatchdogConfig; + supervisor: { supervised: boolean; hint: string }; + recommendation: WatchdogRecommendation; + restartCount: number; +} + +/** + * Full observability report for the dashboard (Monitor + Recommend). Read-only: it never mutates + * config or state. Returns null when the watchdog is not running (disabled); the caller reports that + * as { enabled: false } so old/disabled servers degrade gracefully. + */ +export function memoryWatchdogReport(): WatchdogReport | null { + if (!running) return null; + const supervisor = detectSupervisor(); + return { + enabled: true, + decision: running.last, + samplesCount: running.state.samples.length, + growthMbPerHour: running.last?.growthMbPerHour ?? null, + resolvedConfig: running.cfg, + supervisor, + recommendation: recommend(running.state, running.cfg, supervisor.supervised), + restartCount: running.state.restartCount, + }; +} diff --git a/src/types.ts b/src/types.ts index 341cb73942c..b55d557c936 100644 --- a/src/types.ts +++ b/src/types.ts @@ -520,6 +520,30 @@ export interface OcxConfig { connectTimeoutMs?: number; /** Graceful shutdown drain timeout (ms). Active turns are aborted after this deadline. Default 5000. */ shutdownTimeoutMs?: number; + /** + * Memory watchdog: observe process memory pressure relative to total system RAM and warn (or, + * opt-in, gracefully restart) before the Bun/mimalloc native allocator can exhaust the system + * commit charge. Thresholds are fractions of total RAM so behavior is identical across machines. + * Default: enabled, warn-only (autoRestart false). + */ + memoryWatchdog?: { + /** Master switch. Default true (warn-only observation is safe). */ + enabled?: boolean; + /** Sampling interval (ms). Default 60000. */ + intervalMs?: number; + /** Warn when pressure (committed-or-RSS) / total RAM crosses this fraction. Default 0.60. */ + warnFraction?: number; + /** Critical threshold fraction; arms an opt-in restart. Default 0.75. */ + criticalFraction?: number; + /** Opt in to a graceful restart at the critical threshold. Default false (warn only). */ + autoRestart?: boolean; + /** When true (default), auto-restart only fires if a process supervisor is detected. */ + requireSupervisor?: boolean; + /** Minimum ms between auto-restarts (loop guard). Default 600000. */ + minRestartIntervalMs?: number; + /** Max auto-restarts before giving up and warning only (loop guard). Default 3. */ + maxRestarts?: number; + }; /** Advertise supports_websockets so Codex opens the WS endpoint. Default false; set true to opt in. */ websockets?: boolean; /** Generated API keys for external access to the proxy's /v1/responses endpoint. */ diff --git a/tests/memory-api.test.ts b/tests/memory-api.test.ts new file mode 100644 index 00000000000..f58e27813b9 --- /dev/null +++ b/tests/memory-api.test.ts @@ -0,0 +1,144 @@ +/** + * Management API surface for the memory watchdog (dashboard beta panel): + * GET /api/memory → read-only Monitor + Recommend report (Runs behind requireApiAuth). + * PUT /api/memory/settings → validate + persist (saveConfig) + apply live without a restart. + * The route handlers are exercised directly; auth/CORS gating is asserted elsewhere. saveConfig is + * pointed at an isolated OPENCODEX_HOME so persistence is checked without touching the real config. + */ +import { afterEach, describe, expect, test } from "bun:test"; +import { existsSync, mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { getConfigPath } from "../src/config"; +import { handleManagementAPI } from "../src/server/management-api"; +import { stopMemoryWatchdog } from "../src/server/memory-watchdog"; +import type { OcxConfig } from "../src/types"; + +const savedHome = process.env.OPENCODEX_HOME; +let tempHome: string | null = null; + +afterEach(() => { + stopMemoryWatchdog(); // never leak a running watchdog singleton across tests + if (savedHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = savedHome; + if (tempHome) { + try { rmSync(tempHome, { recursive: true, force: true }); } catch { /* best-effort */ } + tempHome = null; + } +}); + +function isolatedHome(): void { + tempHome = mkdtempSync(join(tmpdir(), "ocx-memory-")); + process.env.OPENCODEX_HOME = tempHome; +} + +function makeConfig(overrides: Partial = {}): OcxConfig { + return { port: 10100, providers: {}, defaultProvider: "openai", ...overrides } as OcxConfig; +} + +async function get(config: OcxConfig): Promise { + const req = new Request("http://localhost/api/memory"); + const res = await handleManagementAPI(req, new URL(req.url), config); + expect(res).not.toBeNull(); + return res!; +} + +async function put(config: OcxConfig, body: unknown): Promise { + const req = new Request("http://localhost/api/memory/settings", { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(body), + }); + const res = await handleManagementAPI(req, new URL(req.url), config); + expect(res).not.toBeNull(); + return res!; +} + +interface Report { + enabled: boolean; + resolvedConfig?: { warnFraction: number; criticalFraction: number; requireSupervisor: boolean; autoRestart: boolean }; + supervisor?: { supervised: boolean; hint: string }; + recommendation?: unknown; +} + +describe("GET /api/memory", () => { + test("degrades to { enabled: false } when the watchdog is not running", async () => { + isolatedHome(); + const config = makeConfig(); + const res = await get(config); + expect(res.status).toBe(200); + expect(await res.json()).toEqual({ enabled: false }); + }); + + test("after enabling, the report exposes the Monitor + Recommend shape", async () => { + isolatedHome(); + const config = makeConfig(); + await put(config, { enabled: true, warnFraction: 0.55, criticalFraction: 0.8 }); + const data = (await (await get(config)).json()) as Report; + expect(data.enabled).toBe(true); + expect(data.resolvedConfig?.warnFraction).toBe(0.55); + expect(data.resolvedConfig?.criticalFraction).toBe(0.8); + expect(data.supervisor).toBeDefined(); + expect(data.recommendation).toBeDefined(); + }); +}); + +describe("PUT /api/memory/settings", () => { + test("valid patch persists to disk and applies live", async () => { + isolatedHome(); + const config = makeConfig(); + const res = await put(config, { enabled: true, warnFraction: 0.5, criticalFraction: 0.85, requireSupervisor: false }); + expect(res.status).toBe(200); + const body = (await res.json()) as { ok: boolean; report: Report }; + expect(body.ok).toBe(true); + // in-memory config mutated + expect(config.memoryWatchdog).toMatchObject({ warnFraction: 0.5, criticalFraction: 0.85, requireSupervisor: false }); + // persisted to the isolated config file + const persisted = JSON.parse(readFileSync(getConfigPath(), "utf8")) as OcxConfig; + expect(persisted.memoryWatchdog).toMatchObject({ warnFraction: 0.5, criticalFraction: 0.85, requireSupervisor: false }); + // applied live: the running report reflects the new thresholds + expect(body.report.resolvedConfig?.warnFraction).toBe(0.5); + expect(body.report.resolvedConfig?.requireSupervisor).toBe(false); + }); + + test("a live threshold change updates the already-running watchdog", async () => { + isolatedHome(); + const config = makeConfig(); + await put(config, { enabled: true, warnFraction: 0.6, criticalFraction: 0.75 }); + const res = await put(config, { warnFraction: 0.4 }); + const body = (await res.json()) as { report: Report }; + expect(res.status).toBe(200); + expect(body.report.resolvedConfig?.warnFraction).toBe(0.4); + }); + + test("enabled:false stops the watchdog and the report degrades", async () => { + isolatedHome(); + const config = makeConfig(); + await put(config, { enabled: true }); + expect(((await (await get(config)).json()) as Report).enabled).toBe(true); + await put(config, { enabled: false }); + expect(await (await get(config)).json()).toEqual({ enabled: false }); + }); + + test.each([ + ["out-of-range warnFraction", { warnFraction: 2 }], + ["non-numeric criticalFraction", { criticalFraction: "high" }], + ["non-boolean autoRestart", { autoRestart: "yes" }], + ["non-boolean requireSupervisor", { requireSupervisor: 1 }], + ["non-positive intervalMs", { intervalMs: 0 }], + ] as const)("rejects %s with 400 and does not persist", async (_label, body) => { + isolatedHome(); + const config = makeConfig(); + const res = await put(config, body); + expect(res.status).toBe(400); + expect(config.memoryWatchdog).toBeUndefined(); + expect(existsSync(getConfigPath())).toBe(false); + }); + + test("rejects a non-object body with 400", async () => { + isolatedHome(); + const config = makeConfig(); + const res = await put(config, ["not", "an", "object"]); + expect(res.status).toBe(400); + }); +}); diff --git a/tests/memory-watchdog.test.ts b/tests/memory-watchdog.test.ts new file mode 100644 index 00000000000..fbf7fcb662c --- /dev/null +++ b/tests/memory-watchdog.test.ts @@ -0,0 +1,383 @@ +import { describe, expect, test } from "bun:test"; +import { + parseProcStatusCommitted, + pressureBytes, + pressureSource, + readMemorySnapshot, + setMemoryPlatformForTests, + setProcStatusReaderForTests, + setWindowsProbeRunnerForTests, + type MemorySnapshot, +} from "../src/lib/memory-usage"; +import { + applyWatchdogRuntimeConfig, + createWatchdogState, + detectSupervisor, + evaluate, + recommend, + resolveWatchdogConfig, + startMemoryWatchdog, + stopMemoryWatchdog, + tick, + type ResolvedWatchdogConfig, + type WatchdogDeps, +} from "../src/server/memory-watchdog"; +import type { OcxConfig } from "../src/types"; + +const MiB = 1024 * 1024; + +function snapshot(pressure: number, opts: Partial = {}): MemorySnapshot { + return { + rssBytes: opts.rssBytes ?? pressure, + heapUsedBytes: opts.heapUsedBytes ?? 10 * MiB, + externalBytes: opts.externalBytes ?? 0, + committedBytes: opts.committedBytes ?? pressure, + committedSource: opts.committedSource ?? "windows-private", + totalSystemBytes: opts.totalSystemBytes ?? 100 * MiB, + }; +} + +function cfg(over: Partial = {}): ResolvedWatchdogConfig { + return { + enabled: true, + intervalMs: 60_000, + warnFraction: 0.60, + criticalFraction: 0.75, + autoRestart: false, + requireSupervisor: true, + minRestartIntervalMs: 600_000, + maxRestarts: 3, + growthWindowMs: 600_000, + ...over, + }; +} + +describe("memory-usage snapshot", () => { + test("parseProcStatusCommitted sums VmRSS + VmSwap (kB → bytes)", () => { + const text = "Name:\tbun\nVmRSS:\t 1024 kB\nVmSwap:\t 512 kB\n"; + expect(parseProcStatusCommitted(text)).toBe((1024 + 512) * 1024); + }); + + test("parseProcStatusCommitted tolerates a missing VmSwap and rejects missing VmRSS", () => { + expect(parseProcStatusCommitted("VmRSS:\t 2048 kB\n")).toBe(2048 * 1024); + expect(parseProcStatusCommitted("VmData:\t 100 kB\n")).toBeNull(); + }); + + test("linux snapshot reports committed from /proc/self/status", () => { + setMemoryPlatformForTests("linux"); + setProcStatusReaderForTests(() => "VmRSS:\t 4096 kB\nVmSwap:\t 0 kB\n"); + try { + const snap = readMemorySnapshot(); + expect(snap.committedSource).toBe("proc-status"); + expect(snap.committedBytes).toBe(4096 * 1024); + expect(pressureSource(snap)).toBe("proc-status"); + } finally { + setMemoryPlatformForTests(null); + setProcStatusReaderForTests(null); + } + }); + + test("windows snapshot uses the private-bytes probe", () => { + setMemoryPlatformForTests("win32"); + setWindowsProbeRunnerForTests(() => ({ privateBytes: 79 * 1024 * MiB, timedOut: false })); + try { + const snap = readMemorySnapshot(); + expect(snap.committedSource).toBe("windows-private"); + expect(snap.committedBytes).toBe(79 * 1024 * MiB); + } finally { + setMemoryPlatformForTests(null); + setWindowsProbeRunnerForTests(null); + } + }); + + test("a timed-out windows probe soft-fails to the RSS fallback", () => { + setMemoryPlatformForTests("win32"); + setWindowsProbeRunnerForTests(() => ({ privateBytes: null, timedOut: true })); + try { + const snap = readMemorySnapshot(); + expect(snap.committedBytes).toBeNull(); + expect(snap.committedSource).toBe("none"); + // pressure falls back to RSS so the watchdog still functions. + expect(pressureBytes(snap)).toBe(snap.rssBytes); + expect(pressureSource(snap)).toBe("rss-fallback"); + } finally { + setMemoryPlatformForTests(null); + setWindowsProbeRunnerForTests(null); + } + }); +}); + +describe("resolveWatchdogConfig", () => { + test("defaults are warn-only and enabled", () => { + const r = resolveWatchdogConfig({} as OcxConfig); + expect(r.enabled).toBe(true); + expect(r.autoRestart).toBe(false); + expect(r.warnFraction).toBe(0.60); + expect(r.criticalFraction).toBe(0.75); + }); + + test("config file values override defaults", () => { + const r = resolveWatchdogConfig({ + memoryWatchdog: { enabled: true, warnFraction: 0.5, criticalFraction: 0.8, autoRestart: true }, + } as OcxConfig); + expect(r.warnFraction).toBe(0.5); + expect(r.criticalFraction).toBe(0.8); + expect(r.autoRestart).toBe(true); + }); + + test("critical <= warn is nudged above warn so the levels never collapse", () => { + const r = resolveWatchdogConfig({ + memoryWatchdog: { warnFraction: 0.7, criticalFraction: 0.6 }, + } as OcxConfig); + expect(r.criticalFraction).toBeGreaterThan(r.warnFraction); + }); + + test("fractions are clamped into [0.10, 0.99]", () => { + const r = resolveWatchdogConfig({ + memoryWatchdog: { warnFraction: 0.01, criticalFraction: 5 }, + } as OcxConfig); + expect(r.warnFraction).toBe(0.10); + expect(r.criticalFraction).toBe(0.99); + }); +}); + +describe("evaluate (pure decision core)", () => { + test("stays ok below the warn fraction", () => { + const s = createWatchdogState(); + const d = evaluate(s, snapshot(50 * MiB), cfg(), 1_000); + expect(d.level).toBe("ok"); + expect(d.action).toBe("none"); + }); + + test("warns once on the first crossing, then latches", () => { + const s = createWatchdogState(); + const first = evaluate(s, snapshot(65 * MiB), cfg(), 1_000); + expect(first.level).toBe("warn"); + expect(first.action).toBe("warn"); + const second = evaluate(s, snapshot(66 * MiB), cfg(), 2_000); + expect(second.level).toBe("warn"); + expect(second.action).toBe("none"); // latched — no repeat log + }); + + test("recovering to ok re-arms the warning", () => { + const s = createWatchdogState(); + evaluate(s, snapshot(65 * MiB), cfg(), 1_000); + evaluate(s, snapshot(50 * MiB), cfg(), 2_000); // back to ok + const again = evaluate(s, snapshot(65 * MiB), cfg(), 3_000); + expect(again.action).toBe("warn"); + }); + + test("critical with auto-restart disabled warns but never restarts", () => { + const s = createWatchdogState(); + const d = evaluate(s, snapshot(80 * MiB), cfg({ autoRestart: false }), 1_000); + expect(d.level).toBe("critical"); + expect(d.action).toBe("warn"); + expect(d.reason).toContain("auto-restart disabled"); + expect(s.restartCount).toBe(0); + }); + + test("critical with auto-restart armed restarts once, then defers by cooldown", () => { + const s = createWatchdogState(); + const c = cfg({ autoRestart: true, minRestartIntervalMs: 600_000 }); + const first = evaluate(s, snapshot(80 * MiB), c, 1_000); + expect(first.action).toBe("restart"); + expect(s.restartCount).toBe(1); + + // Still critical a moment later — cooldown must suppress a second restart. + const second = evaluate(s, snapshot(85 * MiB), c, 2_000); + expect(second.action).not.toBe("restart"); + expect(second.reason).toContain("cooldown"); + expect(s.restartCount).toBe(1); + }); + + test("max-restart guard stops restarting after the cap", () => { + const s = createWatchdogState(); + const c = cfg({ autoRestart: true, minRestartIntervalMs: 0, maxRestarts: 2 }); + // Each call is critical and past the (zero) cooldown. + expect(evaluate(s, snapshot(80 * MiB), c, 10_000).action).toBe("restart"); + expect(evaluate(s, snapshot(80 * MiB), c, 20_000).action).toBe("restart"); + const third = evaluate(s, snapshot(80 * MiB), c, 30_000); + expect(third.action).not.toBe("restart"); + expect(third.reason).toContain("max-restart"); + expect(s.restartCount).toBe(2); + }); + + test("computes growth rate across the sample window", () => { + const s = createWatchdogState(); + evaluate(s, snapshot(50 * MiB), cfg(), 0); + const d = evaluate(s, snapshot(60 * MiB), cfg(), 3_600_000); // +10 MiB over exactly 1h + expect(d.growthMbPerHour).toBe(10); + }); +}); + +describe("tick (driver wiring)", () => { + function deps(over: Partial): WatchdogDeps { + return { + now: () => 1_000, + read: () => snapshot(50 * MiB), + supervised: true, + restart: () => {}, + log: () => {}, + ...over, + }; + } + + test("logs on a warn action and invokes restart on a restart action", () => { + const logs: string[] = []; + let restarts = 0; + const s = createWatchdogState(); + const c = cfg({ autoRestart: true, minRestartIntervalMs: 0 }); + + // warn crossing + tick(s, c, deps({ read: () => snapshot(65 * MiB), log: l => logs.push(l) })); + expect(logs.length).toBe(1); + expect(logs[0]).toContain("WARN"); + + // critical crossing → restart + const d = tick(s, c, deps({ + read: () => snapshot(80 * MiB), + log: l => logs.push(l), + restart: () => { restarts += 1; }, + now: () => 2_000, + })); + expect(d.action).toBe("restart"); + expect(restarts).toBe(1); + }); + + test("a read failure inside tick propagates as a thrown error only to the caller's try", () => { + // startMemoryWatchdog wraps tick in try/catch; here we assert tick itself surfaces the error + // so the wrapper (not the timer) owns swallowing it. + const s = createWatchdogState(); + expect(() => tick(s, cfg(), deps({ read: () => { throw new Error("probe boom"); } }))).toThrow("probe boom"); + }); +}); + +describe("detectSupervisor", () => { + test("explicit OCX_SUPERVISED=1 wins", () => { + expect(detectSupervisor({ OCX_SUPERVISED: "1" })).toEqual({ supervised: true, hint: "OCX_SUPERVISED" }); + }); + + test("pm2 markers are detected", () => { + expect(detectSupervisor({ pm_id: "0" })).toEqual({ supervised: true, hint: "pm2" }); + expect(detectSupervisor({ PM2_HOME: "/home/app/.pm2" })).toEqual({ supervised: true, hint: "pm2" }); + }); + + test("systemd markers are detected", () => { + expect(detectSupervisor({ INVOCATION_ID: "abc123" })).toEqual({ supervised: true, hint: "systemd" }); + expect(detectSupervisor({ NOTIFY_SOCKET: "/run/systemd/notify" })).toEqual({ supervised: true, hint: "systemd" }); + }); + + test("nothing set → not supervised (safe default)", () => { + expect(detectSupervisor({})).toEqual({ supervised: false, hint: "none" }); + }); + + test("explicit OCX_SUPERVISED=off reports the reason", () => { + expect(detectSupervisor({ OCX_SUPERVISED: "off" })).toEqual({ supervised: false, hint: "OCX_SUPERVISED=off" }); + }); +}); + +describe("recommend (advisory only)", () => { + test("with < 2 samples it echoes current config and says so", () => { + const s = createWatchdogState(); + const r = recommend(s, cfg({ warnFraction: 0.5, criticalFraction: 0.7 }), true); + expect(r.warnFraction).toBe(0.5); + expect(r.criticalFraction).toBe(0.7); + expect(r.rationale).toContain("insufficient samples"); + }); + + test("places warn above the observed peak and keeps critical > warn", () => { + const s = createWatchdogState(); + const c = cfg(); + evaluate(s, snapshot(40 * MiB), c, 0); // fraction 0.40 + evaluate(s, snapshot(50 * MiB), c, 1_000); // fraction 0.50 (observed peak) + const r = recommend(s, c, true); + expect(r.warnFraction).toBeGreaterThan(0.50); + expect(r.criticalFraction).toBeGreaterThan(r.warnFraction); + }); + + test("clamps into [0.10, 0.99] even at very high pressure", () => { + const s = createWatchdogState(); + const c = cfg(); + evaluate(s, snapshot(95 * MiB), c, 0); + evaluate(s, snapshot(98 * MiB), c, 1_000); + const r = recommend(s, c, true); + expect(r.warnFraction).toBeLessThanOrEqual(0.99); + expect(r.criticalFraction).toBeLessThanOrEqual(0.99); + }); + + test("suggests auto-restart only when supervised AND trending up", () => { + const s = createWatchdogState(); + const c = cfg(); + evaluate(s, snapshot(40 * MiB), c, 0); + evaluate(s, snapshot(50 * MiB), c, 3_600_000); // growing + expect(recommend(s, c, true).autoRestart).toBe(true); + expect(recommend(s, c, false).autoRestart).toBe(false); // no supervisor → never suggested + }); + + test("does not suggest auto-restart for a flat process", () => { + const s = createWatchdogState(); + const c = cfg(); + evaluate(s, snapshot(50 * MiB), c, 0); + evaluate(s, snapshot(50 * MiB), c, 3_600_000); // no growth + expect(recommend(s, c, true).autoRestart).toBe(false); + }); +}); + +describe("evaluate — requireSupervisor gate", () => { + test("critical + autoRestart but no supervisor → warns instead of restarting", () => { + const s = createWatchdogState(); + const c = cfg({ autoRestart: true, requireSupervisor: true, minRestartIntervalMs: 0 }); + const d = evaluate(s, snapshot(80 * MiB), c, 1_000, /* supervised */ false); + expect(d.level).toBe("critical"); + expect(d.action).toBe("warn"); + expect(d.reason).toContain("no supervisor"); + expect(s.restartCount).toBe(0); + }); + + test("critical + autoRestart + supervised → restarts", () => { + const s = createWatchdogState(); + const c = cfg({ autoRestart: true, requireSupervisor: true, minRestartIntervalMs: 0 }); + const d = evaluate(s, snapshot(80 * MiB), c, 1_000, /* supervised */ true); + expect(d.action).toBe("restart"); + expect(s.restartCount).toBe(1); + }); + + test("requireSupervisor=false lets auto-restart fire without a supervisor", () => { + const s = createWatchdogState(); + const c = cfg({ autoRestart: true, requireSupervisor: false, minRestartIntervalMs: 0 }); + const d = evaluate(s, snapshot(80 * MiB), c, 1_000, /* supervised */ false); + expect(d.action).toBe("restart"); + }); +}); + +describe("applyWatchdogRuntimeConfig (live tuning)", () => { + test("returns null when the watchdog is not running", () => { + stopMemoryWatchdog(); + expect(applyWatchdogRuntimeConfig({ warnFraction: 0.5 })).toBeNull(); + }); + + test("live-updates fractions with clamping and critical > warn", () => { + try { + startMemoryWatchdog({} as OcxConfig, { read: () => snapshot(50 * MiB), now: () => 0, log: () => {} }); + const applied = applyWatchdogRuntimeConfig({ warnFraction: 0.01, criticalFraction: 5 }); + expect(applied).not.toBeNull(); + expect(applied!.warnFraction).toBe(0.10); // clamped up + expect(applied!.criticalFraction).toBe(0.99); // clamped down + const collapsed = applyWatchdogRuntimeConfig({ warnFraction: 0.7, criticalFraction: 0.6 }); + expect(collapsed!.criticalFraction).toBeGreaterThan(collapsed!.warnFraction); + } finally { + stopMemoryWatchdog(); + } + }); + + test("updates requireSupervisor / autoRestart in place", () => { + try { + startMemoryWatchdog({} as OcxConfig, { read: () => snapshot(50 * MiB), now: () => 0, log: () => {} }); + const applied = applyWatchdogRuntimeConfig({ autoRestart: true, requireSupervisor: false }); + expect(applied!.autoRestart).toBe(true); + expect(applied!.requireSupervisor).toBe(false); + } finally { + stopMemoryWatchdog(); + } + }); +}); diff --git a/tests/responses-state.test.ts b/tests/responses-state.test.ts index 35c9f4f49ce..3bb3d9628aa 100644 --- a/tests/responses-state.test.ts +++ b/tests/responses-state.test.ts @@ -14,6 +14,7 @@ import { previousResponseConversationId, previousResponseProviderState, rememberResponseState, + responseStateMetrics, } from "../src/responses/state"; describe("Responses previous_response_id state", () => { @@ -360,4 +361,45 @@ describe("Responses previous_response_id state", () => { clearPrior: newlyCompactedRequest.contextUsageReset === true, }).carryForwardTokens).toBeUndefined(); }); + + test("responseStateMetrics reports an empty store as zeroed", () => { + expect(responseStateMetrics()).toEqual({ + count: 0, + totalBytes: 0, + largestBytes: 0, + oldestAgeMs: 0, + }); + }); + + test("responseStateMetrics counts entries and tracks the largest by serialized bytes", () => { + const small = buildResponseJSON([{ type: "text_delta", text: "hi" }, { type: "done" }], "gpt-5.5"); + rememberResponseState({ model: "gpt-5.5", input: "small" }, small); + + const bigText = "y".repeat(200 * 1024); + const big = buildResponseJSON([{ type: "text_delta", text: bigText }, { type: "done" }], "gpt-5.5"); + rememberResponseState({ model: "gpt-5.5", input: "big" }, big); + + const metrics = responseStateMetrics(); + expect(metrics.count).toBe(2); + expect(metrics.largestBytes).toBeGreaterThan(200 * 1024); + expect(metrics.totalBytes).toBeGreaterThanOrEqual(metrics.largestBytes); + expect(metrics.oldestAgeMs).toBeGreaterThanOrEqual(0); + }); + + test("responseStateMetrics is side-effect free (does not lazy-load the disk snapshot)", () => { + // Seed a snapshot on disk, then wipe memory. A pure metrics read must NOT resurrect it: it + // reflects live RAM only, so a diagnostics probe never perturbs the store or triggers a load. + const first = buildResponseJSON([{ type: "text_delta", text: "persisted" }, { type: "done" }], "gpt-5.5"); + rememberResponseState({ model: "gpt-5.5", input: "hi" }, first); + flushResponseState(); + clearResponseStateMemoryForTests(); + + expect(responseStateMetrics().count).toBe(0); + + // A real read path still loads it, proving the snapshot was intact and metrics simply abstained. + expect((expandPreviousResponseInput({ + model: "gpt-5.5", previous_response_id: first.id, input: "next", + }) as { input: unknown[] }).input).toHaveLength(3); + expect(responseStateMetrics().count).toBe(1); + }); }); From 59daab43143f701392f2a84603f07fffaea793ed Mon Sep 17 00:00:00 2001 From: shinyu Date: Fri, 24 Jul 2026 14:54:36 +0900 Subject: [PATCH 2/9] feat(memory): quiet-window watchdog restart via configurable restartGraceMs Replace the hard-coded 5s drain before a memory-driven restart with a configurable restartGraceMs (default 30s). drainAndShutdown already rejects new turns (503) and returns the moment the in-flight set empties, so the restart now lands on a natural idle gap and only aborts turns that outlive the budget. Exposed via config (memoryWatchdog.restartGraceMs), env (OCX_MEMORY_WATCHDOG_RESTART_GRACE_MS), and PUT /api/memory/settings (validated as a positive number). Live-tunable through applyWatchdogRuntimeConfig. --- src/server/management-api.ts | 2 ++ src/server/memory-watchdog.ts | 18 +++++++++++++++--- src/types.ts | 4 ++++ tests/memory-watchdog.test.ts | 15 ++++++++++++--- 4 files changed, 33 insertions(+), 6 deletions(-) diff --git a/src/server/management-api.ts b/src/server/management-api.ts index 49a22ef6d04..ff10794c8c6 100644 --- a/src/server/management-api.ts +++ b/src/server/management-api.ts @@ -272,12 +272,14 @@ export async function handleManagementAPI(req: Request, url: URL, config: OcxCon if (b.autoRestart !== undefined && !isBool(b.autoRestart)) return jsonResponse({ error: "autoRestart must be boolean" }, 400); if (b.requireSupervisor !== undefined && !isBool(b.requireSupervisor)) return jsonResponse({ error: "requireSupervisor must be boolean" }, 400); if (b.intervalMs !== undefined && !isPosNum(b.intervalMs)) return jsonResponse({ error: "intervalMs must be a positive number" }, 400); + if (b.restartGraceMs !== undefined && !isPosNum(b.restartGraceMs)) return jsonResponse({ error: "restartGraceMs must be a positive number" }, 400); if (b.warnFraction !== undefined && !isFrac(b.warnFraction)) return jsonResponse({ error: "warnFraction must be a number in (0,1)" }, 400); if (b.criticalFraction !== undefined && !isFrac(b.criticalFraction)) return jsonResponse({ error: "criticalFraction must be a number in (0,1)" }, 400); const patch: NonNullable = {}; if (b.enabled !== undefined) patch.enabled = b.enabled as boolean; if (b.intervalMs !== undefined) patch.intervalMs = b.intervalMs as number; + if (b.restartGraceMs !== undefined) patch.restartGraceMs = b.restartGraceMs as number; if (b.warnFraction !== undefined) patch.warnFraction = b.warnFraction as number; if (b.criticalFraction !== undefined) patch.criticalFraction = b.criticalFraction as number; if (b.autoRestart !== undefined) patch.autoRestart = b.autoRestart as boolean; diff --git a/src/server/memory-watchdog.ts b/src/server/memory-watchdog.ts index a7a036f63f7..0e69f544a82 100644 --- a/src/server/memory-watchdog.ts +++ b/src/server/memory-watchdog.ts @@ -13,6 +13,12 @@ * - Auto-restart is opt-in (config.memoryWatchdog.autoRestart / env). When it fires it reuses * drainAndShutdown() — which drains in-flight turns and flushes the response-state snapshot — * then exits with a distinct code so a supervisor can respawn. + * - Quiet-window restart: the drain budget for a memory-driven restart is restartGraceMs (default + * 30s), much longer than a generic shutdown. Because drainAndShutdown rejects new turns and exits + * the instant the in-flight set empties, the restart normally lands on a natural idle gap — so a + * running turn is only cut when it outlives the whole grace window (bounded so the restart always + * fires). A mid-stream turn that is still cut is regenerated on retry (context is preserved via + * the flushed previous_response_id snapshot); the provider generation itself cannot be resumed. * - A cooldown (minRestartIntervalMs) and a max-restart guard prevent restart loops. * - The decision core (evaluate) is pure and injectable (clock + reader + restart hook) so the * logic is unit-tested without real timers or a real process exit. @@ -37,6 +43,9 @@ export interface ResolvedWatchdogConfig { requireSupervisor: boolean; minRestartIntervalMs: number; maxRestarts: number; + /** Drain budget (ms) for a memory-driven restart: wait up to this long for in-flight turns to + * finish before aborting, so the restart lands on a natural idle gap (quiet-window). */ + restartGraceMs: number; growthWindowMs: number; } @@ -49,6 +58,7 @@ const DEFAULTS: ResolvedWatchdogConfig = { requireSupervisor: true, minRestartIntervalMs: 600_000, // 10 min maxRestarts: 3, + restartGraceMs: 30_000, // quiet-window: wait up to 30s for in-flight turns before aborting growthWindowMs: 600_000, // 10 min growth-rate window (diagnostic) }; @@ -109,6 +119,8 @@ export function resolveWatchdogConfig(config: OcxConfig): ResolvedWatchdogConfig minRestartIntervalMs: envNum("OCX_MEMORY_WATCHDOG_MIN_RESTART_INTERVAL_MS") ?? c.minRestartIntervalMs ?? DEFAULTS.minRestartIntervalMs, maxRestarts: envNum("OCX_MEMORY_WATCHDOG_MAX_RESTARTS") ?? c.maxRestarts ?? DEFAULTS.maxRestarts, + restartGraceMs: + envNum("OCX_MEMORY_WATCHDOG_RESTART_GRACE_MS") ?? c.restartGraceMs ?? DEFAULTS.restartGraceMs, growthWindowMs: DEFAULTS.growthWindowMs, }; } @@ -311,13 +323,13 @@ export interface WatchdogDeps { function defaultRestart(cfg: ResolvedWatchdogConfig): void { void (async () => { try { - await drainAndShutdown(undefined, 5_000); + // Quiet-window: drainAndShutdown rejects new turns and returns the moment the in-flight set + // empties, so this waits for a natural idle gap and only aborts turns that outlive the budget. + await drainAndShutdown(undefined, cfg.restartGraceMs); } finally { process.exit(MEMORY_RESTART_EXIT_CODE); } })(); - // Reference cfg so the signature stays honest for tests overriding it. - void cfg; } const DEFAULT_DEPS: WatchdogDeps = { diff --git a/src/types.ts b/src/types.ts index b55d557c936..45707481d24 100644 --- a/src/types.ts +++ b/src/types.ts @@ -543,6 +543,10 @@ export interface OcxConfig { minRestartIntervalMs?: number; /** Max auto-restarts before giving up and warning only (loop guard). Default 3. */ maxRestarts?: number; + /** Quiet-window drain budget (ms) for a memory-driven restart: wait up to this long for + * in-flight turns to finish before aborting, so the restart lands on a natural idle gap. + * Default 30000. */ + restartGraceMs?: number; }; /** Advertise supports_websockets so Codex opens the WS endpoint. Default false; set true to opt in. */ websockets?: boolean; diff --git a/tests/memory-watchdog.test.ts b/tests/memory-watchdog.test.ts index fbf7fcb662c..b3da5c63975 100644 --- a/tests/memory-watchdog.test.ts +++ b/tests/memory-watchdog.test.ts @@ -47,6 +47,7 @@ function cfg(over: Partial = {}): ResolvedWatchdogConfig requireSupervisor: true, minRestartIntervalMs: 600_000, maxRestarts: 3, + restartGraceMs: 30_000, growthWindowMs: 600_000, ...over, }; @@ -139,6 +140,12 @@ describe("resolveWatchdogConfig", () => { expect(r.warnFraction).toBe(0.10); expect(r.criticalFraction).toBe(0.99); }); + + test("restartGraceMs defaults to the quiet-window budget and is config-overridable", () => { + expect(resolveWatchdogConfig({} as OcxConfig).restartGraceMs).toBe(30_000); + const r = resolveWatchdogConfig({ memoryWatchdog: { restartGraceMs: 120_000 } } as OcxConfig); + expect(r.restartGraceMs).toBe(120_000); + }); }); describe("evaluate (pure decision core)", () => { @@ -225,23 +232,25 @@ describe("tick (driver wiring)", () => { test("logs on a warn action and invokes restart on a restart action", () => { const logs: string[] = []; let restarts = 0; + let restartGrace = 0; const s = createWatchdogState(); - const c = cfg({ autoRestart: true, minRestartIntervalMs: 0 }); + const c = cfg({ autoRestart: true, minRestartIntervalMs: 0, restartGraceMs: 45_000 }); // warn crossing tick(s, c, deps({ read: () => snapshot(65 * MiB), log: l => logs.push(l) })); expect(logs.length).toBe(1); expect(logs[0]).toContain("WARN"); - // critical crossing → restart + // critical crossing → restart, and the running cfg (incl. the quiet-window budget) is threaded through const d = tick(s, c, deps({ read: () => snapshot(80 * MiB), log: l => logs.push(l), - restart: () => { restarts += 1; }, + restart: rc => { restarts += 1; restartGrace = rc.restartGraceMs; }, now: () => 2_000, })); expect(d.action).toBe("restart"); expect(restarts).toBe(1); + expect(restartGrace).toBe(45_000); }); test("a read failure inside tick propagates as a thrown error only to the caller's try", () => { From 1ff459dc956c7b6e66229aa878e4ebe2efb6390a Mon Sep 17 00:00:00 2001 From: shinyu Date: Fri, 24 Jul 2026 14:54:36 +0900 Subject: [PATCH 3/9] feat(gui): honest restart-behavior note under auto-restart toggle Add a calm, non-alarming muted note beneath the auto-restart control explaining what happens when it fires: in-flight requests drain first, a reply still streaming at that instant is regenerated on retry, the conversation context is preserved, and restarts are rate-limited. Adds memory.autoRestartNote to the en source of truth plus de/ko/zh/ru/ja translations (compile-checked key set). --- gui/src/components/MemoryCard.tsx | 1 + gui/src/i18n/de.ts | 1 + gui/src/i18n/en.ts | 1 + gui/src/i18n/ja.ts | 1 + gui/src/i18n/ko.ts | 1 + gui/src/i18n/ru.ts | 1 + gui/src/i18n/zh.ts | 1 + 7 files changed, 7 insertions(+) diff --git a/gui/src/components/MemoryCard.tsx b/gui/src/components/MemoryCard.tsx index fa118b6e29a..78ab45d48b8 100644 --- a/gui/src/components/MemoryCard.tsx +++ b/gui/src/components/MemoryCard.tsx @@ -251,6 +251,7 @@ export default function MemoryCard({ apiBase }: { apiBase: string }) { {autoRestartBlocked && (
{t("memory.autoRestartBlocked")}
)} +
{t("memory.autoRestartNote")}
+ {mem && ( +
+ {/* Actual Private Bytes vs the pressure value in the gauge above — deliberately distinct. */} + {t("memory.privateLabel")} {mem.processPrivateBytes === null ? t("memory.notAvailable") : t("memory.mbValue", { mb: nf(Math.round(mem.processPrivateBytes / MIB)) })} + {t("memory.commitLabel")} {mem.systemCommitAvailable && mem.systemCommittedBytes !== null && mem.systemCommitLimitBytes !== null + ? t("memory.commitValue", { used: gb1(mem.systemCommittedBytes), limit: gb1(mem.systemCommitLimitBytes), pct: mem.systemCommitFraction === null ? "?" : Math.round(mem.systemCommitFraction * 100) }) + : t("memory.notAvailable")} + {rs && ( + {t("memory.responseCacheLabel")} {t("memory.responseCacheValue", { count: nf(rs.count), mb: nf(Math.round(rs.totalBytes / MIB)) })} + )} + {t("memory.probeLabel")} {probeAgeS === null ? t("memory.notAvailable") : t("memory.probeAgo", { s: probeAgeS })}{mem.probeError && ( + {t("memory.probeDegraded", { code: mem.probeError })} + )} +
+ )} {/* 2. Recommend */} diff --git a/gui/src/i18n/de.ts b/gui/src/i18n/de.ts index c7d9f7de7f2..1593f98640b 100644 --- a/gui/src/i18n/de.ts +++ b/gui/src/i18n/de.ts @@ -12,6 +12,15 @@ export const de = { "memory.growthValue": "{mb} MB/h", "memory.notAvailable": "n/v", "memory.restartsLabel": "Neustarts:", + "memory.privateLabel": "Privat:", + "memory.commitLabel": "System-Commit:", + "memory.commitValue": "{used} / {limit} GB ({pct}%)", + "memory.responseCacheLabel": "Antwort-Cache:", + "memory.responseCacheValue": "{count} Einträge · {mb} MB", + "memory.probeLabel": "Messung:", + "memory.probeAgo": "vor {s}s", + "memory.probeDegraded": "eingeschränkt ({code})", + "memory.mbValue": "{mb} MB", "memory.supervisorLabel": "Supervisor:", "memory.supervisorDetected": "erkannt ({hint})", "memory.supervisorNotDetected": "nicht erkannt", diff --git a/gui/src/i18n/en.ts b/gui/src/i18n/en.ts index a39f6c2f796..41b0fc480b7 100644 --- a/gui/src/i18n/en.ts +++ b/gui/src/i18n/en.ts @@ -13,6 +13,15 @@ export const en = { "memory.growthValue": "{mb} MB/h", "memory.notAvailable": "n/a", "memory.restartsLabel": "restarts:", + "memory.privateLabel": "private:", + "memory.commitLabel": "system commit:", + "memory.commitValue": "{used} / {limit} GB ({pct}%)", + "memory.responseCacheLabel": "response cache:", + "memory.responseCacheValue": "{count} items · {mb} MB", + "memory.probeLabel": "probe:", + "memory.probeAgo": "{s}s ago", + "memory.probeDegraded": "degraded ({code})", + "memory.mbValue": "{mb} MB", "memory.supervisorLabel": "supervisor:", "memory.supervisorDetected": "detected ({hint})", "memory.supervisorNotDetected": "not detected", diff --git a/gui/src/i18n/ja.ts b/gui/src/i18n/ja.ts index ec055bcbb37..e5c4b9dc8fd 100644 --- a/gui/src/i18n/ja.ts +++ b/gui/src/i18n/ja.ts @@ -13,6 +13,15 @@ export const ja: Record = { "memory.growthValue": "{mb} MB/h", "memory.notAvailable": "なし", "memory.restartsLabel": "再起動:", + "memory.privateLabel": "プライベート:", + "memory.commitLabel": "システムコミット:", + "memory.commitValue": "{used} / {limit} GB（{pct}%）", + "memory.responseCacheLabel": "応答キャッシュ:", + "memory.responseCacheValue": "{count} 件 · {mb} MB", + "memory.probeLabel": "計測:", + "memory.probeAgo": "{s}秒前", + "memory.probeDegraded": "制限あり（{code}）", + "memory.mbValue": "{mb} MB", "memory.supervisorLabel": "スーパーバイザー:", "memory.supervisorDetected": "検出 ({hint})", "memory.supervisorNotDetected": "未検出", diff --git a/gui/src/i18n/ko.ts b/gui/src/i18n/ko.ts index 3fa19948ff3..0fc9bc0bf2f 100644 --- a/gui/src/i18n/ko.ts +++ b/gui/src/i18n/ko.ts @@ -13,6 +13,15 @@ export const ko: Record = { "memory.growthValue": "{mb} MB/h", "memory.notAvailable": "해당 없음", "memory.restartsLabel": "재시작:", + "memory.privateLabel": "프라이빗:", + "memory.commitLabel": "시스템 커밋:", + "memory.commitValue": "{used} / {limit} GB ({pct}%)", + "memory.responseCacheLabel": "응답 캐시:", + "memory.responseCacheValue": "{count}개 · {mb} MB", + "memory.probeLabel": "측정:", + "memory.probeAgo": "{s}초 전", + "memory.probeDegraded": "제한됨 ({code})", + "memory.mbValue": "{mb} MB", "memory.supervisorLabel": "supervisor:", "memory.supervisorDetected": "감지됨 ({hint})", "memory.supervisorNotDetected": "감지되지 않음", diff --git a/gui/src/i18n/ru.ts b/gui/src/i18n/ru.ts index 9c2b6ffd538..4ddd3e458ff 100644 --- a/gui/src/i18n/ru.ts +++ b/gui/src/i18n/ru.ts @@ -13,6 +13,15 @@ export const ru: Record = { "memory.growthValue": "{mb} МБ/ч", "memory.notAvailable": "н/д", "memory.restartsLabel": "перезапуски:", + "memory.privateLabel": "приватная:", + "memory.commitLabel": "системный коммит:", + "memory.commitValue": "{used} / {limit} ГБ ({pct}%)", + "memory.responseCacheLabel": "кэш ответов:", + "memory.responseCacheValue": "{count} записей · {mb} МБ", + "memory.probeLabel": "замер:", + "memory.probeAgo": "{s} с назад", + "memory.probeDegraded": "ограничен ({code})", + "memory.mbValue": "{mb} МБ", "memory.supervisorLabel": "супервизор:", "memory.supervisorDetected": "обнаружен ({hint})", "memory.supervisorNotDetected": "не обнаружен", diff --git a/gui/src/i18n/zh.ts b/gui/src/i18n/zh.ts index bd5544ecd6a..92fde29f277 100644 --- a/gui/src/i18n/zh.ts +++ b/gui/src/i18n/zh.ts @@ -13,6 +13,15 @@ export const zh: Record = { "memory.growthValue": "{mb} MB/h", "memory.notAvailable": "不可用", "memory.restartsLabel": "重启：", + "memory.privateLabel": "专用内存：", + "memory.commitLabel": "系统提交：", + "memory.commitValue": "{used} / {limit} GB（{pct}%）", + "memory.responseCacheLabel": "响应缓存：", + "memory.responseCacheValue": "{count} 项 · {mb} MB", + "memory.probeLabel": "探测：", + "memory.probeAgo": "{s} 秒前", + "memory.probeDegraded": "受限（{code}）", + "memory.mbValue": "{mb} MB", "memory.supervisorLabel": "supervisor：", "memory.supervisorDetected": "已检测 ({hint})", "memory.supervisorNotDetected": "未检测", From 9942240de0ab38e729ef55fa0690476103005084 Mon Sep 17 00:00:00 2001 From: shinyu Date: Fri, 24 Jul 2026 14:54:38 +0900 Subject: [PATCH 8/9] fix(memory): harden probe loop against auto-restart race + capture-throw fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four fixes from the joint post-implementation audit (terra-high + qwen-ultimate), all verified against the code before applying: - #1 (blocker): when an auto-restart fired, tick()'s restart hook synchronously nulled the global `running` (defaultRestart calls stopMemoryWatchdog before draining), and probeOnce's trailing `scheduleNextProbe(generation, running.cfg.intervalMs)` then evaluated `running.cfg` on null — an unhandled TypeError at the exact moment of every real auto-restart (surfacing as crash-guard noise, and a dead loop for injected restart hooks). probeOnce now pins its instance in a local `current` and re-checks identity against the global after capture, after tick, and before rescheduling: a stopped/replaced watchdog is left untouched and never rescheduled — the restart owns the exit. - #2: an unexpected exception escaping capture() no longer silences the whole cycle; probeOnce evaluates exactly once on a new exported rssFallbackSnapshot(now, "capture-threw") instead of skipping evaluation. - #3: the PowerShell probe now emits labeled values (P=/C=/L=/A=) parsed by the new pure parseWindowsProbeOutput() — a $null first value previously rendered as a blank line that trim() removed, shifting positional indexes and misassigning commit values to privateBytes. Order-independent, loss-explicit, unit-tested; first-resolution-wins/timeout/kill handling untouched. - #4: applyWatchdogRuntimeConfig now finite-guards systemCommitHighWater and maxRestarts (same pattern as resolveWatchdogConfig): a programmatic partial carrying undefined/NaN would otherwise make `fraction >= NaN` permanently false and silently disable the commit warning. Regression tests: auto-restart-stops-watchdog mid-tick (validated to fail on the pre-fix code via bun's unhandled-rejection surfacing), capture-throw RSS fallback, NaN/undefined live re-clamp, and label-parser unit cases. 77 pass. --- src/lib/memory-usage.ts | 56 +++++++++++++++++------ src/server/memory-watchdog.ts | 64 +++++++++++++++++--------- tests/memory-watchdog.test.ts | 85 +++++++++++++++++++++++++++++++++++ 3 files changed, 171 insertions(+), 34 deletions(-) diff --git a/src/lib/memory-usage.ts b/src/lib/memory-usage.ts index 60ea375a66c..987c2dafe2f 100644 --- a/src/lib/memory-usage.ts +++ b/src/lib/memory-usage.ts @@ -133,7 +133,11 @@ async function defaultWindowsProbeRunner(pid: number, timeoutMs: number, signal? "-NoProfile", "-NonInteractive", "-Command", - `$p=(Get-Process -Id ${pid}).PrivateMemorySize64; $m=Get-CimInstance Win32_PerfRawData_PerfOS_Memory; Write-Output $p $m.CommittedBytes $m.CommitLimit $m.AvailableBytes`, + // Labeled output: a $null value would render as an empty line that trim() removes, shifting + // positional indexes — labels make the parse order-independent and loss-explicit. + `$p=(Get-Process -Id ${pid}).PrivateMemorySize64; ` + + `$m=Get-CimInstance Win32_PerfRawData_PerfOS_Memory; ` + + `Write-Output "P=$p" "C=$($m.CommittedBytes)" "L=$($m.CommitLimit)" "A=$($m.AvailableBytes)"`, ], { stdin: "ignore", stdout: "pipe", stderr: "ignore", windowsHide: true }, ); @@ -172,18 +176,7 @@ async function defaultWindowsProbeRunner(pid: number, timeoutMs: number, signal? finish({ ...EMPTY_PROBE, timedOut: false, error: `probe-exit-${proc.exitCode ?? "unknown"}` }); return; } - const nums = text.trim().split(/\r?\n/).map(line => Number(line.trim())); - const val = (n: number | undefined): number | null => - n !== undefined && Number.isFinite(n) && n > 0 ? n : null; - const result: WindowsProbeResult = { - privateBytes: val(nums[0]), - systemCommittedBytes: val(nums[1]), - systemCommitLimitBytes: val(nums[2]), - availablePhysicalBytes: val(nums[3]), - timedOut: false, - }; - if (result.privateBytes === null && result.systemCommittedBytes === null) result.error = "parse-failed"; - finish(result); + finish(parseWindowsProbeOutput(text)); } catch { finish({ ...EMPTY_PROBE, timedOut: false, error: "probe-failed" }); } @@ -191,6 +184,31 @@ async function defaultWindowsProbeRunner(pid: number, timeoutMs: number, signal? }); } +/** + * Parse the labeled probe output (`P=` private, `C=` committed, `L=` limit, `A=` available). + * Order-independent and tolerant of missing/blank labels — a value PowerShell rendered as empty + * (`P=`) parses to NaN and degrades to null instead of shifting the other fields. Exported as a + * pure function so the parse path is unit-testable without spawning a child. + */ +export function parseWindowsProbeOutput(text: string): WindowsProbeResult { + const fields = new Map(); + for (const line of text.trim().split(/\r?\n/)) { + const match = /^([PCLA])=(.*)$/.exec(line.trim()); + if (match) fields.set(match[1]!, Number(match[2])); + } + const val = (n: number | undefined): number | null => + n !== undefined && Number.isFinite(n) && n > 0 ? n : null; + const result: WindowsProbeResult = { + privateBytes: val(fields.get("P")), + systemCommittedBytes: val(fields.get("C")), + systemCommitLimitBytes: val(fields.get("L")), + availablePhysicalBytes: val(fields.get("A")), + timedOut: false, + }; + if (result.privateBytes === null && result.systemCommittedBytes === null) result.error = "parse-failed"; + return result; +} + let windowsProbeRunner: WindowsProbeRunner = defaultWindowsProbeRunner; /** Test seam: replace the Windows probe runner. Pass null to restore the default. */ @@ -253,6 +271,18 @@ function baseSnapshot(capturedAt: number): MemorySnapshot { }; } +/** + * RSS-fallback snapshot for callers that must evaluate SOMETHING even when capture itself failed. + * captureMemorySnapshot is contractually non-throwing, but the probe loop uses this as a safety + * net so an unexpected runtime exception never silences a whole evaluation cycle (audit #2). + * probeError takes sanitized codes only — never raw command output or exception text. + */ +export function rssFallbackSnapshot(nowMs: number = Date.now(), probeError?: string): MemorySnapshot { + const snapshot = baseSnapshot(nowMs); + if (probeError !== undefined) snapshot.probeError = probeError; + return snapshot; +} + /** * Capture one memory snapshot. ASYNC because the Windows path awaits a child-process probe; it * never throws and never blocks the event loop. Probe failures degrade to the RSS fallback with a diff --git a/src/server/memory-watchdog.ts b/src/server/memory-watchdog.ts index d9cb53f3440..97cd91b3e42 100644 --- a/src/server/memory-watchdog.ts +++ b/src/server/memory-watchdog.ts @@ -42,7 +42,7 @@ * logic is unit-tested without real timers or a real process exit. */ -import { captureMemorySnapshot, type MemorySnapshot } from "../lib/memory-usage"; +import { captureMemorySnapshot, rssFallbackSnapshot, type MemorySnapshot } from "../lib/memory-usage"; import { responseStateMetrics, type ResponseStateMetrics } from "../responses/state"; import { drainAndShutdown } from "./lifecycle"; import { loadMemoryRestartHistory, recordMemoryRestart } from "./memory-restart-history"; @@ -556,33 +556,46 @@ export function tick( * stopMemoryWatchdog()/a restart replaced the singleton, so a late probe can never revive state. */ async function probeOnce(generation: number): Promise { - const r = running; - if (!r || r.generation !== generation) return; - r.timer = null; - r.abort = new AbortController(); - let snapshot: MemorySnapshot | null = null; + // tick()'s restart hook can synchronously null the global `running` (defaultRestart calls + // stopMemoryWatchdog before draining) — so pin the instance this probe owns in a local and, + // from here on, decide "am I still current?" by IDENTITY against the global (audit #1). + const current = running; + if (!current || current.generation !== generation) return; + current.timer = null; + current.abort = new AbortController(); + + let snapshot: MemorySnapshot; try { - snapshot = await r.deps.capture(r.deps.now(), r.abort.signal); + snapshot = await current.deps.capture(current.deps.now(), current.abort.signal); } catch { - snapshot = null; // capture is contractually non-throwing; treat a bug as a skipped cycle + // Audit #2: captureMemorySnapshot is contractually non-throwing, but an unexpected runtime + // exception must not silence the cycle — evaluate exactly once on the RSS fallback instead. + snapshot = rssFallbackSnapshot(current.deps.now(), "capture-threw"); } - if (!running || running.generation !== generation) return; // stopped mid-probe — drop late result - running.abort = null; + + // Stopped/replaced mid-capture: drop the late result (global is null or a different instance). + if (running !== current || running.generation !== generation) return; + current.abort = null; + try { - if (snapshot) { - running.last = tick(running.state, running.cfg, running.deps, snapshot); - running.lastSnapshot = snapshot; - // §7: COMPLETION time, not capture start — a slow probe must read as fresh, not stale. - const completedAt = running.deps.now(); - running.lastProbeAt = completedAt; - if (snapshot.systemCommitAvailable) running.lastSuccessfulSystemProbeAt = completedAt; - // Response-state metrics serialize every entry — compute once per probe, not per poll. - try { running.responseState = responseStateMetrics(); } catch { running.responseState = null; } - } + const decision = tick(current.state, current.cfg, current.deps, snapshot); + // If tick fired an auto-restart, its hook already stopped (or replaced) the watchdog: leave + // the dead/foreign singleton untouched and do not reschedule — the restart owns the exit. + if (running !== current || running.generation !== generation) return; + current.last = decision; + current.lastSnapshot = snapshot; + // §7: COMPLETION time, not capture start — a slow probe must read as fresh, not stale. + const completedAt = current.deps.now(); + current.lastProbeAt = completedAt; + if (snapshot.systemCommitAvailable) current.lastSuccessfulSystemProbeAt = completedAt; + // Response-state metrics serialize every entry — compute once per probe, not per poll. + try { current.responseState = responseStateMetrics(); } catch { current.responseState = null; } } catch { /* an evaluation failure must never crash the proxy */ } - scheduleNextProbe(generation, running.cfg.intervalMs); + + if (running !== current || running.generation !== generation) return; + scheduleNextProbe(generation, current.cfg.intervalMs); } function scheduleNextProbe(generation: number, delayMs: number): void { @@ -646,6 +659,15 @@ export function applyWatchdogRuntimeConfig( } next.restartGraceMs = clampMs(next.restartGraceMs, RESTART_GRACE_MIN_MS, RESTART_GRACE_MAX_MS, running.cfg.restartGraceMs); next.minRestartIntervalMs = posNumOr(next.minRestartIntervalMs, running.cfg.minRestartIntervalMs); + // Audit #4: not PUT-exposed, but a programmatic partial carrying undefined/NaN would otherwise + // survive the spread — an NaN high-water makes `fraction >= NaN` permanently false and silently + // disables the commit warning. Same finite-guard pattern as resolveWatchdogConfig's maxRestarts. + next.systemCommitHighWater = Number.isFinite(next.systemCommitHighWater) + ? Math.min(0.99, Math.max(0.50, next.systemCommitHighWater)) + : running.cfg.systemCommitHighWater; + next.maxRestarts = Number.isFinite(next.maxRestarts) && next.maxRestarts >= 0 + ? Math.floor(next.maxRestarts) + : running.cfg.maxRestarts; normalizeRestartTiming(next); next.intervalMs = Math.max(1_000, posNumOr(next.intervalMs, running.cfg.intervalMs)); const intervalChanged = next.intervalMs !== running.cfg.intervalMs; diff --git a/tests/memory-watchdog.test.ts b/tests/memory-watchdog.test.ts index 08297ff5e07..2129ffdfd1b 100644 --- a/tests/memory-watchdog.test.ts +++ b/tests/memory-watchdog.test.ts @@ -6,6 +6,7 @@ import { setRestartHistoryPathForTests } from "../src/server/memory-restart-hist import { captureMemorySnapshot, parseProcStatusCommitted, + parseWindowsProbeOutput, setMemoryPlatformForTests, setProcStatusReaderForTests, setWindowsProbeRunnerForTests, @@ -170,6 +171,26 @@ describe("memory-usage snapshot", () => { } }); + test("labeled probe output parses order-independently and tolerates missing values", () => { + const full = parseWindowsProbeOutput("P=100\nC=200\nL=300\nA=400\n"); + expect(full).toMatchObject({ privateBytes: 100, systemCommittedBytes: 200, systemCommitLimitBytes: 300, availablePhysicalBytes: 400 }); + expect(full.error).toBeUndefined(); + + // Shuffled line order must not reassign fields (the old positional parser's failure mode). + const shuffled = parseWindowsProbeOutput("A=400\r\nL=300\r\nP=100\r\nC=200"); + expect(shuffled).toMatchObject({ privateBytes: 100, systemCommittedBytes: 200 }); + + // A $null private renders as "P=" — private degrades to null WITHOUT shifting commit values. + const noPrivate = parseWindowsProbeOutput("P=\nC=200\nL=300\nA=400"); + expect(noPrivate.privateBytes).toBeNull(); + expect(noPrivate.systemCommittedBytes).toBe(200); + expect(noPrivate.error).toBeUndefined(); + + // Garbage / empty output is an explicit parse failure, not a zero-filled snapshot. + expect(parseWindowsProbeOutput("").error).toBe("parse-failed"); + expect(parseWindowsProbeOutput("access denied\nblah").error).toBe("parse-failed"); + }); + // Real end-to-end probe: only meaningful (and only run) on an actual Windows host. Verifies the // §4 field/unit assumptions live: private bytes for this PID plus system CommittedBytes / // CommitLimit in bytes, all from a single async PowerShell spawn that never blocks the loop. @@ -747,6 +768,70 @@ describe("startMemoryWatchdog — probe-driven loop (§2, §8)", () => { stopMemoryWatchdog(); expect(aborted).toBe(true); }); + + test("audit #1: an auto-restart that stops the watchdog mid-tick does not crash or reschedule", async () => { + const root = mkdtempSync(join(tmpdir(), "ocx-watchdog-restart-race-")); + let captures = 0; + try { + setRestartHistoryPathForTests(join(root, "absent.json")); // fresh guards regardless of runner env + startMemoryWatchdog( + { memoryWatchdog: { enabled: true, autoRestart: true, requireSupervisor: false } } as OcxConfig, + { + capture: async () => { captures += 1; return snapshot(90 * MiB); }, // 90/100 = critical (>0.75) + now: () => 1_000, + supervised: true, + restart: () => stopMemoryWatchdog(), // exactly what defaultRestart does synchronously + recordRestart: () => {}, // keep the test disk-free + log: () => {}, + }, + ); + await Bun.sleep(0); // let the immediate probe's microtasks complete + // With the pre-fix code this turn died on `running.cfg` (unhandled TypeError at reschedule). + expect(memoryWatchdogReport()).toBeNull(); // the restart's stop was respected — nothing revived + expect(captures).toBe(1); // and no extra capture was scheduled by the dead loop + } finally { + stopMemoryWatchdog(); + setRestartHistoryPathForTests(null); + rmSync(root, { recursive: true, force: true }); + } + }); + + test("audit #2: a throwing capture still evaluates exactly once via the RSS fallback", async () => { + try { + startMemoryWatchdog({} as OcxConfig, { + capture: async () => { throw new Error("unexpected boom"); }, + now: () => 5_000, + log: () => {}, + }); + await Bun.sleep(0); + const report = memoryWatchdogReport(); + expect(report).not.toBeNull(); + expect(report!.decision).not.toBeNull(); // the cycle was evaluated, not silenced + expect(report!.samplesCount).toBeGreaterThan(0); + expect(report!.lastProbeAt).toBe(5_000); + expect(report!.memory!.probeError).toBe("capture-threw"); + expect(report!.memory!.processSource).toBe("rss-fallback"); + } finally { + stopMemoryWatchdog(); + } + }); + + test("audit #4: applyWatchdogRuntimeConfig defends systemCommitHighWater against NaN/undefined", () => { + try { + startMemoryWatchdog({} as OcxConfig, { capture: async () => snapshot(50 * MiB), now: () => 0, log: () => {} }); + const nan = applyWatchdogRuntimeConfig({ systemCommitHighWater: Number.NaN }); + expect(Number.isFinite(nan!.systemCommitHighWater)).toBe(true); // previous value kept, not NaN + expect(nan!.systemCommitHighWater).toBe(0.90); + const undef = applyWatchdogRuntimeConfig({ systemCommitHighWater: undefined as unknown as number }); + expect(undef!.systemCommitHighWater).toBe(0.90); + const clamped = applyWatchdogRuntimeConfig({ systemCommitHighWater: 5 }); + expect(clamped!.systemCommitHighWater).toBe(0.99); // clamped down into [0.50, 0.99] + const junkRestarts = applyWatchdogRuntimeConfig({ maxRestarts: Number.NaN }); + expect(junkRestarts!.maxRestarts).toBe(3); // finite-guarded like resolveWatchdogConfig + } finally { + stopMemoryWatchdog(); + } + }); }); describe("startMemoryWatchdog — cross-process history seeding", () => { From a9590de76298a4e8156298f1ec9ab782de92eaf9 Mon Sep 17 00:00:00 2001 From: shinyu Date: Fri, 24 Jul 2026 14:54:39 +0900 Subject: [PATCH 9/9] =?UTF-8?q?docs(structure):=20memory=20watchdog=20SOT?= =?UTF-8?q?=20=E2=80=94=20design=20rationale,=20platform=20scope,=20audit?= =?UTF-8?q?=20trail?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Permanent record for the PR: why the watchdog exists (79GB committed / 97% system commit field incident), the async-collect/sync-decide architecture and its defended invariants, the platform matrix (cross-platform core, Windows- first measurement fidelity; v1 system-commit axis is Windows-only), why the commit axis is observe-only in v1 (it cannot know the pressure is ours — and honestly does NOT auto-mitigate the original incident yet), the quiet-window restart + honestly-scoped loop guards, verification evidence incl. live measurements, the three audit rounds, and deferred follow-ups. --- structure/09_memory-watchdog.md | 129 ++++++++++++++++++++++++++++++++ 1 file changed, 129 insertions(+) create mode 100644 structure/09_memory-watchdog.md diff --git a/structure/09_memory-watchdog.md b/structure/09_memory-watchdog.md new file mode 100644 index 00000000000..668da61a3eb --- /dev/null +++ b/structure/09_memory-watchdog.md @@ -0,0 +1,129 @@ +# Memory Watchdog SOT + +## Why this exists + +Field incident: OpenCodex on Bun reached **~79 GB private/committed memory while the working set +stayed ~5.8 GB**, freezing the box once the **system commit charge hit ~97% of the commit limit**. +The Bun/mimalloc native allocator retains committed memory the JS heap has already released +(mirrored upstream by Claude Code #36132); no in-app JS cache cap can fix native retention. The +locally-run stress harness reproduces the detectability signature on a healthy machine +(`afterClear private/RSS = 4.4–4.8` after clearing the JS store and forcing GC). + +The only proven mitigation is to **observe the pressure and, opt-in, hand off to a graceful +restart**. Everything below follows from that. + +## Platform scope (PR reviewers: read this first) + +The watchdog is **cross-platform**; only the measurement *fidelity* differs per platform. This is +NOT a Windows-only patch, but the motivating incident, the highest-fidelity probe, and the v1 +system-commit axis are Windows-specific. + +| Layer | win32 | linux | other | +| --- | --- | --- | --- | +| Process pressure source | Private Bytes (async PowerShell probe) | VmRSS+VmSwap (`/proc/self/status`, no spawn) | RSS (in-process) | +| System commit axis (v1) | measured (CommittedBytes/CommitLimit) | not measured (`systemCommitAvailable=false`; `/proc/meminfo` `Committed_AS` is a documented follow-up) | not measured | +| Decision core (warn/critical, cooldown, max-restarts, history seeding) | identical | identical | identical | +| Quiet-window drain + exit-75 restart, supervisor gating | identical | identical | identical | +| Dashboard / management API / i18n | identical | identical | identical | + +The commit-axis logic itself is platform-neutral and null-safe: on platforms without measurement +it simply skips (no NaN propagation, latch held), so enabling a Linux collector later is additive. + +## Architecture: async collect, sync decide + +```text +self-rescheduling probe loop (unref'd; first probe fires IMMEDIATELY at start) + └─ captureMemorySnapshot() — async, never throws, never blocks the event loop + win32: ONE hidden PowerShell child (array args, -NoProfile -NonInteractive) returns + labeled values P=/C=/L=/A= (private, committed, commit limit, available physical; + Win32_PerfRawData_PerfOS_Memory — all bytes, verified live). 15s budget + (measured cold spawn: 4.0–5.9s); timeout kills the child; first-resolution-wins + drops late results; failures degrade to an RSS fallback with a sanitized probeError. + └─ tick(state, cfg, deps, snapshot) — sync, pure-core evaluate(), EXACTLY once per probe + axis 1 PROCESS: pressure/physicalRAM vs warn(0.60)/critical(0.75) → opt-in restart path + axis 2 SYSTEM COMMIT: observe-only high-water warning (see below), NEVER restarts + └─ report cache (memory block, responseStateMetrics, lastProbeAt/lastSuccessfulSystemProbeAt) + computed once per probe — a 5s dashboard poll never re-serializes the response store +``` + +Key invariants, each defended by a regression test: + +- **The probe never blocks the event loop.** The original implementation used `Bun.spawnSync` + (up to 2s synchronous stall per tick on Windows) — flagged as a deploy blocker by audit and + replaced. A streaming proxy must not pause SSE for measurement. +- **Exactly one evaluation per probe.** Timeout → RSS-fallback evaluation, and a late success for + the same probe is dropped; an unexpected `capture()` exception evaluates once on + `rssFallbackSnapshot(..., "capture-threw")` instead of silencing the cycle. +- **A restart owns the exit.** `tick()`'s restart hook synchronously stops the watchdog; + `probeOnce` pins its instance in a local and re-checks IDENTITY against the global after + capture, after tick, and before rescheduling, so the stopped/replaced singleton is never + touched or rescheduled (pre-fix this was an unhandled TypeError on every real auto-restart). +- **Stop kills the child.** `stopMemoryWatchdog()` aborts the in-flight capture via + AbortController and a generation guard drops any result that still arrives. + +## System-commit axis: why observe-only (v1) + +`systemCommittedBytes / systemCommitLimitBytes >= 0.90` (env-only override +`OCX_MEMORY_WATCHDOG_COMMIT_HIGH_WATER`, clamped [0.50, 0.99]) logs ONE latched warning; the latch +re-arms only on a MEASURED recovery — measurement loss (probe failure) holds it, so a flapping +probe cannot re-warn each time it returns. + +It never contributes to the restart decision because the commit pressure may come from **another +process**: restarting OpenCodex would not free it, and an automated restart loop against an +external cause is worse than a loud warning. Honest consequence, stated up front: **v1 does not +auto-mitigate the original incident** (79 GB / 128 GB RAM ≈ 62% process pressure < 0.75 critical; +system commit 97% → warning only). Promoting the commit axis into the restart decision, and any +UI-configurable thresholds, are deliberately deferred until field measurement validates the axis. + +## Restart path (quiet-window) and honest loop guards + +- `restartGraceMs` (default 30s, clamped [1s, 10min] on every entry path — env, config file, + management API which 400s out-of-range values): draining rejects new turns with 503+Retry-After + and returns the moment the in-flight set empties, so the restart lands on a natural idle gap and + the value is a DEADLINE, not a delay. +- `minRestartIntervalMs` is normalized to ≥ `restartGraceMs` so a second restart can never arm + inside the first drain window. +- Cooldown/max-restarts counters die with the process a restart ends, so they are re-seeded from + a best-effort timestamps-only history file (rolling ~6h window, atomic-rename write, every + failure swallowed). This is **not** a permanent cross-process guarantee: the supervisor's own + restart limit/backoff (pm2 `max_restarts`, systemd `StartLimitBurst`) remains the outer layer, + and exit code 75 is only a *request* to respawn. NSSM / Windows services are not auto-detected — + set `OCX_SUPERVISED=1` there. + +## Verification evidence (as merged) + +- Unit/integration: `tests/memory-watchdog.test.ts` (66), `tests/memory-restart-history.test.ts` + (11), `tests/memory-api.test.ts` (16) — includes a REAL Windows end-to-end probe test asserting + private bytes and a sane commit fraction from the live labeled command. +- Live measurements on a Windows 11 box: probe cold spawn 4.0–5.9s (hence the 15s async budget); + `Win32_PerfRawData_PerfOS_Memory` values confirmed to be bytes; harness detectability + `afterClear private/RSS = 4.4–4.8`. +- Gates: `typecheck`, `lint:gui`, `build:gui`, `privacy:scan` all exit 0. `probeError` carries + sanitized codes only (never raw command output/paths) — privacy-scan enforced. +- Full-suite caveat: on the development Windows box the suite exits 1 due to pre-existing + environment failures (NTFS ACL hardening `EACCES` clusters in auth tests + branch-dependent + release-helper tests). Evidence that these are unrelated: the failure set at this branch's HEAD + is a strict subset (321 ⊂ 345, zero new files, zero memory-file failures) of the same run at the + pre-change baseline commit under identical conditions. + +## Audit trail + +Three independent audit rounds shaped this design (summarized honestly rather than re-litigated): + +1. Initial audit (deploy blockers): synchronous `spawnSync` probe stall, missing system-commit + measurement, 60s startup blind spot, RSS-only stress assertions, unwired response-state + metrics → all five confirmed against code and fixed by the async rewrite. +2. Design review: chose "cached-free" probe-completion-driven evaluation over defer-until-idle + and blue-green alternatives (complexity/regression cost vs. benefit), kept warn-only defaults. +3. Post-implementation audit: auto-restart race in `probeOnce` (unhandled TypeError), capture-throw + silence, positional PowerShell parsing, live re-clamp gaps → all four fixed with regression + tests (the race test was validated to fail against the pre-fix code). + +## Follow-ups (out of scope here) + +- Field measurement, then possible commit-axis restart integration + UI thresholds (v2). +- Linux system commit via `/proc/meminfo` (`Committed_AS`/`CommitLimit`, no spawn). +- docs-site ja/zh-cn translations of the memoryWatchdog reference section. +- Separate issue: `windows-secret-acl` hardening is non-atomic under icacls timeouts + (inheritance stripped before grant), which is the root cause of the environmental `EACCES` + test failures on slow-ACL machines.