summaryrefslogtreecommitdiffhomepage
path: root/packages/host-bin
diff options
context:
space:
mode:
Diffstat (limited to 'packages/host-bin')
-rw-r--r--packages/host-bin/src/main.ts40
-rw-r--r--packages/host-bin/src/mem-telemetry.test.ts225
-rw-r--r--packages/host-bin/src/mem-telemetry.ts160
3 files changed, 424 insertions, 1 deletions
diff --git a/packages/host-bin/src/main.ts b/packages/host-bin/src/main.ts
index aa114d5..f892fe8 100644
--- a/packages/host-bin/src/main.ts
+++ b/packages/host-bin/src/main.ts
@@ -27,7 +27,11 @@ import { extension as messageQueueExt } from "@dispatch/message-queue";
import { extension as providerConcurrencyExt } from "@dispatch/provider-concurrency";
import { extension as providerOpenaiCompatExt } from "@dispatch/provider-openai-compat";
import { extension as providerUmansExt } from "@dispatch/provider-umans";
-import { extension as sessionOrchestratorExt } from "@dispatch/session-orchestrator";
+import {
+ type MemorySample,
+ extension as sessionOrchestratorExt,
+ sessionOrchestratorHandle,
+} from "@dispatch/session-orchestrator";
import { extension as skillsExt } from "@dispatch/skills";
import { extension as sshExt } from "@dispatch/ssh";
import { createSqliteStorage, extension as storageSqliteExt } from "@dispatch/storage-sqlite";
@@ -49,6 +53,7 @@ import type { ChildHandle } from "./collector-supervisor.js";
import { createCollectorSupervisor } from "./collector-supervisor.js";
import { configMapToAccess, envToConfigMap } from "./config.js";
import { loadExternalExtensions } from "./load-external.js";
+import { startMemoryTelemetry } from "./mem-telemetry.js";
function createEmptySecrets(): SecretsAccess {
return {
@@ -227,10 +232,43 @@ async function boot(): Promise<void> {
}
}
+ // Periodic memory telemetry — leak-localization edge effect (AGENTS.md:
+ // timers are edge effects owned by host-bin, the composition root, NOT the
+ // kernel). Logs process.memoryUsage() every 60s tagged with the active-
+ // conversation count, and every 5 min runs Bun.gc(true) + logs RSS
+ // before/after to distinguish live retained objects from GC fragmentation.
+ // The per-turn before/after sampling lives in session-orchestrator; this
+ // owns the PERIODIC baseline. All effects are injected (no ambient state);
+ // stop() is cleared on shutdown so timers never leak across a restart.
+ let memoryTelemetry: { stop: () => void } | undefined;
+ try {
+ const orchestrator = host.getHostAPI().getService(sessionOrchestratorHandle);
+ memoryTelemetry = startMemoryTelemetry({
+ logger: logger.child({ extensionId: "mem-telemetry" }),
+ sampleMemory: (): MemorySample => {
+ const m = process.memoryUsage();
+ return {
+ rss: m.rss,
+ heapUsed: m.heapUsed,
+ heapTotal: m.heapTotal,
+ external: m.external,
+ arrayBuffers: m.arrayBuffers,
+ };
+ },
+ gc: () => Bun.gc(true),
+ getActiveConversationCount: () => orchestrator.getActiveConversationCount(),
+ });
+ } catch (err) {
+ logger.error("Memory telemetry not started (session-orchestrator unavailable)", {
+ err,
+ });
+ }
+
let shuttingDown = false;
const shutdown = async () => {
if (shuttingDown) return;
shuttingDown = true;
+ memoryTelemetry?.stop();
logger.info("Shutting down — deactivating extensions");
await host.deactivate();
logger.info("Draining collector");
diff --git a/packages/host-bin/src/mem-telemetry.test.ts b/packages/host-bin/src/mem-telemetry.test.ts
new file mode 100644
index 0000000..20237ef
--- /dev/null
+++ b/packages/host-bin/src/mem-telemetry.test.ts
@@ -0,0 +1,225 @@
+import type { Attributes, Logger } from "@dispatch/kernel";
+import type { MemorySample } from "@dispatch/session-orchestrator";
+import { describe, expect, it } from "vitest";
+import {
+ buildGcAttributes,
+ buildPeriodicAttributes,
+ type MemoryTelemetryDeps,
+ startMemoryTelemetry,
+} from "./mem-telemetry.js";
+
+/** Minimal capturing logger — records every info/debug/warn/error call. */
+interface CapturedLog {
+ readonly level: string;
+ readonly msg: string;
+ readonly attrs?: Attributes;
+}
+
+function capturingLogger(): { logger: Logger; logs: CapturedLog[] } {
+ const logs: CapturedLog[] = [];
+ const record = (level: string) => (msg: string, attrs?: Attributes) => {
+ logs.push({ level, msg, attrs });
+ };
+ const logger: Logger = {
+ debug: record("debug"),
+ info: record("info"),
+ warn: record("warn"),
+ error: () => {},
+ child: () => logger,
+ span: () => ({
+ id: "s",
+ log: logger,
+ setAttributes: () => {},
+ addLink: () => {},
+ child: () => ({}) as never,
+ end: () => {},
+ }),
+ };
+ return { logger, logs };
+}
+
+const SAMPLE_A: MemorySample = {
+ rss: 100 * 1024 * 1024,
+ heapUsed: 40 * 1024 * 1024,
+ heapTotal: 60 * 1024 * 1024,
+ external: 5 * 1024 * 1024,
+ arrayBuffers: 2 * 1024 * 1024,
+};
+const SAMPLE_B: MemorySample = {
+ rss: 300 * 1024 * 1024,
+ heapUsed: 80 * 1024 * 1024,
+ heapTotal: 60 * 1024 * 1024,
+ external: 5 * 1024 * 1024,
+ arrayBuffers: 6 * 1024 * 1024,
+};
+
+describe("buildPeriodicAttributes", () => {
+ it("formats the sample as MB and tags the active-conversation count", () => {
+ const attrs = buildPeriodicAttributes(SAMPLE_A, 3);
+ expect(attrs).toEqual({
+ rssMB: 100,
+ heapUsedMB: 40,
+ heapTotalMB: 60,
+ externalMB: 5,
+ arrayBuffersMB: 2,
+ activeConversations: 3,
+ });
+ });
+});
+
+describe("buildGcAttributes", () => {
+ it("carries absolute after values plus reclaimed (before-after) delta", () => {
+ const attrs = buildGcAttributes(SAMPLE_A, SAMPLE_B);
+ // Absolute "after" values (SAMPLE_B).
+ expect(attrs.rssMB).toBe(300);
+ expect(attrs.heapUsedMB).toBe(80);
+ // Reclaimed delta: before - after is negative here (memory GREW), so
+ // reclaimedRssMB = round((100-300) MB) = -200.
+ expect(attrs.reclaimedRssMB).toBe(-200);
+ expect(attrs.reclaimedHeapUsedMB).toBe(-40);
+ expect(attrs.reclaimedHeapTotalMB).toBe(0);
+ expect(attrs.reclaimedArrayBuffersMB).toBe(-4);
+ });
+
+ it("shows a positive reclaimed value when GC freed memory", () => {
+ const attrs = buildGcAttributes(SAMPLE_B, SAMPLE_A);
+ expect(attrs.reclaimedRssMB).toBe(200);
+ });
+});
+
+describe("startMemoryTelemetry", () => {
+ function fakeTimers(): {
+ setInterval: MemoryTelemetryDeps["setInterval"];
+ clearInterval: MemoryTelemetryDeps["clearInterval"];
+ tick: (name: "sample" | "gc") => void;
+ cleared: { sample: number; gc: number };
+ } {
+ // Store the callbacks keyed by interval so we can drive either one. The
+ // gc interval is always larger than the sample interval, so route the
+ // larger ms to the gc callback.
+ let sampleCb: (() => void) | undefined;
+ let gcCb: (() => void) | undefined;
+ let firstMs = 0;
+ const cleared = { sample: 0, gc: 0 };
+ return {
+ setInterval: (fn, ms) => {
+ if (firstMs === 0) {
+ firstMs = ms;
+ sampleCb = fn;
+ return "sample" as never;
+ }
+ // The second timer registered is the gc one (larger interval).
+ gcCb = fn;
+ return "gc" as never;
+ },
+ clearInterval: (handle) => {
+ if (handle === "sample") cleared.sample++;
+ else if (handle === "gc") cleared.gc++;
+ },
+ tick: (name) => {
+ if (name === "sample") sampleCb?.();
+ else gcCb?.();
+ },
+ cleared,
+ };
+ }
+
+ it("logs a periodic sample tagged with the active-conversation count", () => {
+ const { logger, logs } = capturingLogger();
+ let active = 2;
+ const timers = fakeTimers();
+ const handle = startMemoryTelemetry({
+ logger,
+ sampleMemory: () => SAMPLE_A,
+ gc: () => {},
+ getActiveConversationCount: () => active,
+ setInterval: timers.setInterval,
+ clearInterval: timers.clearInterval,
+ });
+
+ timers.tick("sample");
+ const periodic = logs.find((l) => l.msg === "memory:periodic");
+ expect(periodic).toBeDefined();
+ expect(periodic?.attrs?.rssMB).toBe(100);
+ expect(periodic?.attrs?.activeConversations).toBe(2);
+
+ // The count is read live (re-evaluated each tick).
+ active = 5;
+ timers.tick("sample");
+ const periodic2 = logs.filter((l) => l.msg === "memory:periodic");
+ expect(periodic2).toHaveLength(2);
+ expect(periodic2[1]?.attrs?.activeConversations).toBe(5);
+
+ handle.stop();
+ expect(timers.cleared.sample).toBe(1);
+ expect(timers.cleared.gc).toBe(1);
+ });
+
+ it("runs gc and logs RSS before/after on the gc interval", () => {
+ const { logger, logs } = capturingLogger();
+ let calls = 0;
+ const samples = [SAMPLE_A, SAMPLE_B];
+ const timers = fakeTimers();
+ const handle = startMemoryTelemetry({
+ logger,
+ sampleMemory: () => samples[calls++] as MemorySample,
+ gc: () => {},
+ getActiveConversationCount: () => 0,
+ setInterval: timers.setInterval,
+ clearInterval: timers.clearInterval,
+ });
+
+ timers.tick("gc");
+ const gcLog = logs.find((l) => l.msg === "memory:gc");
+ expect(gcLog).toBeDefined();
+ // before=SAMPLE_A (call 0), after=SAMPLE_B (call 1) — rss grew 100→300.
+ expect(gcLog?.attrs?.rssMB).toBe(300);
+ expect(gcLog?.attrs?.reclaimedRssMB).toBe(-200);
+
+ handle.stop();
+ });
+
+ it("actually calls the injected gc function", () => {
+ const { logger } = capturingLogger();
+ let gcCalls = 0;
+ const timers = fakeTimers();
+ const handle = startMemoryTelemetry({
+ logger,
+ sampleMemory: () => SAMPLE_A,
+ gc: () => {
+ gcCalls++;
+ },
+ getActiveConversationCount: () => 0,
+ setInterval: timers.setInterval,
+ clearInterval: timers.clearInterval,
+ });
+
+ timers.tick("gc");
+ expect(gcCalls).toBe(1);
+ // Periodic ticks do NOT trigger gc.
+ timers.tick("sample");
+ expect(gcCalls).toBe(1);
+
+ handle.stop();
+ });
+
+ it("stop is idempotent (clearing twice is harmless)", () => {
+ const { logger } = capturingLogger();
+ const timers = fakeTimers();
+ const handle = startMemoryTelemetry({
+ logger,
+ sampleMemory: () => SAMPLE_A,
+ gc: () => {},
+ getActiveConversationCount: () => 0,
+ setInterval: timers.setInterval,
+ clearInterval: timers.clearInterval,
+ });
+
+ handle.stop();
+ handle.stop();
+ // Both timers cleared exactly once each (second stop is a no-op on the
+ // same handles — clear count stays at 1 per handle).
+ expect(timers.cleared.sample).toBe(1);
+ expect(timers.cleared.gc).toBe(1);
+ });
+});
diff --git a/packages/host-bin/src/mem-telemetry.ts b/packages/host-bin/src/mem-telemetry.ts
new file mode 100644
index 0000000..7f0bb42
--- /dev/null
+++ b/packages/host-bin/src/mem-telemetry.ts
@@ -0,0 +1,160 @@
+/**
+ * Periodic memory telemetry — leak-localization edge effect.
+ *
+ * Owns the timers (setInterval) that periodically log process.memoryUsage()
+ * so RSS growth can be correlated with active conversations/turns and the
+ * leaking subsystem pinpointed. This is the composition-root edge effect
+ * (AGENTS.md: timers are edge effects owned by host-bin, NOT the kernel).
+ *
+ * All effects are injected (sampler, GC, clock, logger, active-conversation
+ * count) so the timer logic is fully testable — tests pass fakes and assert
+ * the emitted log calls, never waiting on a real wall clock. No ambient
+ * state (P3): the timers are owned explicitly and returned as a `stop()`
+ * handle that the composition root clears on shutdown.
+ *
+ * The per-turn before/after sampling lives in session-orchestrator (it wraps
+ * the stream boundary); this module owns the PERIODIC baseline + GC logging.
+ */
+
+import type { Logger } from "@dispatch/kernel";
+import {
+ type MemorySample,
+ memoryDelta,
+ memorySampleAttributes,
+} from "@dispatch/session-orchestrator";
+
+/** Default periodic sample interval: every 60s. */
+export const DEFAULT_MEMORY_SAMPLE_INTERVAL_MS = 60_000;
+
+/** Default GC interval: every 5 min (longer than the sample interval). */
+export const DEFAULT_GC_INTERVAL_MS = 5 * 60_000;
+
+/**
+ * Deps injected into {@link startMemoryTelemetry}. Every effect is explicit so
+ * the timer logic is reproducible from its inputs (no ambient state, P3) and
+ * testable without a real clock or process.
+ */
+export interface MemoryTelemetryDeps {
+ /** Logger (auto-scoped to host-bin by the composition root). */
+ readonly logger: Logger;
+ /** Edge effect: capture a {@link MemorySample} now (process.memoryUsage()). */
+ readonly sampleMemory: () => MemorySample;
+ /**
+ * Edge effect: run a full GC cycle. In production this is `() =>
+ * Bun.gc(true)`; tests pass a no-op or a counting fake. Used on the longer
+ * GC interval to distinguish live retained objects from GC fragmentation.
+ */
+ readonly gc: () => void;
+ /**
+ * The number of conversations currently driving a turn (from the
+ * session-orchestrator's activeConversations set). Tags each periodic
+ * sample so growth can be attributed to the streaming/turn path vs an idle
+ * baseline.
+ */
+ readonly getActiveConversationCount: () => number;
+ /** Periodic sample interval (ms). Defaults to 60s. */
+ readonly sampleIntervalMs?: number;
+ /** GC interval (ms). Defaults to 5 min. */
+ readonly gcIntervalMs?: number;
+ /**
+ * Injected timer scheduler (defaults to global setInterval). Tests pass a
+ * fake to drive ticks deterministically without real wall-clock waits. The
+ * handle type is opaque (the same type {@link clearInterval} accepts).
+ */
+ readonly setInterval?: (fn: () => void, ms: number) => MemoryTimerHandle;
+ /** Injected timer clearer (defaults to global clearInterval). */
+ readonly clearInterval?: (handle: MemoryTimerHandle | undefined) => void;
+}
+
+/** Opaque timer handle shared by {@link MemoryTelemetryDeps.setInterval} / clearInterval. */
+export type MemoryTimerHandle = ReturnType<typeof globalThis.setInterval>;
+
+/** Handle returned by {@link startMemoryTelemetry} to stop the timers. */
+export interface MemoryTelemetryHandle {
+ /** Stop both timers. Idempotent. Called by the composition root on shutdown. */
+ readonly stop: () => void;
+}
+
+/**
+ * Start periodic memory telemetry. Logs process.memoryUsage() every
+ * `sampleIntervalMs` (default 60s) tagged with the active-conversation count,
+ * and every `gcIntervalMs` (default 5 min) runs `gc()` and logs RSS
+ * before/after to distinguish live retained objects from GC fragmentation.
+ *
+ * Returns a `stop()` handle that clears both timers. The composition root
+ * (host-bin main) owns this handle and calls `stop()` on shutdown so the
+ * timers never leak across a restart.
+ *
+ * Pure decision logic: {@link buildPeriodicAttributes} /
+ * {@link buildGcAttributes} are exported separately for unit testing.
+ */
+export function startMemoryTelemetry(deps: MemoryTelemetryDeps): MemoryTelemetryHandle {
+ const sampleIntervalMs = deps.sampleIntervalMs ?? DEFAULT_MEMORY_SAMPLE_INTERVAL_MS;
+ const gcIntervalMs = deps.gcIntervalMs ?? DEFAULT_GC_INTERVAL_MS;
+ const setIntervalFn = deps.setInterval ?? globalThis.setInterval;
+ const clearIntervalFn = deps.clearInterval ?? globalThis.clearInterval;
+
+ let gcHandle: MemoryTimerHandle | undefined;
+
+ // Periodic sample: log rss/heap/external/arrayBuffers + active-conversation
+ // count every 60s. Correlates RSS growth with active turns so the leak can
+ // be attributed to the streaming path vs an idle baseline.
+ const sampleHandle: MemoryTimerHandle | undefined = setIntervalFn(() => {
+ const sample = deps.sampleMemory();
+ const activeConversations = deps.getActiveConversationCount();
+ deps.logger.info("memory:periodic", buildPeriodicAttributes(sample, activeConversations));
+ }, sampleIntervalMs);
+
+ // GC log: on a longer interval, force a full GC and log RSS before/after.
+ // A small/no drop after gc means the memory is LIVE (retained objects — the
+ // leak); a large drop means it was GC fragmentation (reclaimable). This
+ // distinguishes the two failure modes the crash investigation flagged.
+ gcHandle = setIntervalFn(() => {
+ const before = deps.sampleMemory();
+ deps.gc();
+ const after = deps.sampleMemory();
+ deps.logger.info("memory:gc", buildGcAttributes(before, after));
+ }, gcIntervalMs);
+
+ let stopped = false;
+ return {
+ stop() {
+ if (stopped) return; // idempotent — safe to call on every shutdown path
+ stopped = true;
+ clearIntervalFn(sampleHandle);
+ clearIntervalFn(gcHandle);
+ },
+ };
+}
+
+/**
+ * Pure: build the logger attributes for a periodic sample. Exported for unit
+ * testing (no I/O, no clock). The `activeConversations` count tags the sample
+ * so growth can be attributed to the streaming/turn path vs idle baseline.
+ */
+export function buildPeriodicAttributes(
+ sample: MemorySample,
+ activeConversations: number,
+): ReturnType<typeof memorySampleAttributes> & { activeConversations: number } {
+ return { ...memorySampleAttributes(sample), activeConversations };
+}
+
+/**
+ * Pure: build the logger attributes for a GC sample. Exported for unit testing.
+ * Carries the absolute after-sample plus the `reclaimed` delta
+ * (`before - after`): a POSITIVE `reclaimedRssMB` means GC freed memory
+ * (fragmentation, reclaimable); near-zero/negative means the memory is LIVE
+ * (retained objects — the leak). This distinguishes the two failure modes the
+ * crash investigation flagged.
+ */
+export function buildGcAttributes(
+ before: MemorySample,
+ after: MemorySample,
+): ReturnType<typeof memorySampleAttributes> {
+ // reclaimed = before - after (how much GC freed). memoryDelta(a, b) = b - a,
+ // so pass (after, before) to get before - after.
+ return {
+ ...memorySampleAttributes(after),
+ ...memorySampleAttributes(memoryDelta(after, before), "reclaimed"),
+ };
+}
generated by cgit v1.2.3 (git 2.39.1) at 2026-08-13 18:18:53 +0000