From 1cd66da48f8c0a35b4208202d07bcd9f20fbc2c2 Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Sun, 28 Jun 2026 08:43:29 +0900 Subject: feat(observability): periodic memory-usage logging to localize leak source --- packages/host-bin/src/main.ts | 40 ++++- packages/host-bin/src/mem-telemetry.test.ts | 225 ++++++++++++++++++++++++++++ packages/host-bin/src/mem-telemetry.ts | 160 ++++++++++++++++++++ 3 files changed, 424 insertions(+), 1 deletion(-) create mode 100644 packages/host-bin/src/mem-telemetry.test.ts create mode 100644 packages/host-bin/src/mem-telemetry.ts (limited to 'packages/host-bin/src') 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 { } } + // 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; + +/** 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 & { 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 { + // 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"), + }; +} -- cgit v1.2.3 From 73ff84c606f5307e5f40e649cae9f93484c0d99d Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Sun, 28 Jun 2026 08:57:17 +0900 Subject: fix: disable LSP + change memory telemetry interval to 15s - Disable LSP extension (import + CORE_EXTENSIONS) due to crashes - Make transport-http tolerate LSP being absent (optional getService) - Remove lsp from transport-http dependsOn manifest - Change memory telemetry sample interval from 60s to 15s --- packages/host-bin/src/main.ts | 6 ++++-- packages/host-bin/src/mem-telemetry.ts | 10 +++++----- packages/transport-http/src/extension.ts | 13 ++++++++++--- 3 files changed, 19 insertions(+), 10 deletions(-) (limited to 'packages/host-bin/src') diff --git a/packages/host-bin/src/main.ts b/packages/host-bin/src/main.ts index f892fe8..29e402e 100644 --- a/packages/host-bin/src/main.ts +++ b/packages/host-bin/src/main.ts @@ -21,7 +21,9 @@ import { type SecretsAccess, type StorageNamespace, } from "@dispatch/kernel"; -import { extension as lspExt } from "@dispatch/lsp"; +// LSP temporarily disabled — crashes (unhandled JSON parse, ENOENT on +// transient .old_modules dirs) and a memory leak. Re-enable after fix. +// import { extension as lspExt } from "@dispatch/lsp"; import { extension as mcpExt } from "@dispatch/mcp"; import { extension as messageQueueExt } from "@dispatch/message-queue"; import { extension as providerConcurrencyExt } from "@dispatch/provider-concurrency"; @@ -106,7 +108,7 @@ const CORE_EXTENSIONS: readonly Extension[] = [ skillsExt, systemPromptExt, cacheWarmingExt, - lspExt, + // lspExt, // LSP temporarily disabled — see import above // ssh declares `dependsOn: ["exec-backend"]` and PROVIDES the remote // exec-backend factory + the ComputerService the HTTP routes delegate to. // Its lookups are lazy (tool-/request-time), but it is placed after diff --git a/packages/host-bin/src/mem-telemetry.ts b/packages/host-bin/src/mem-telemetry.ts index 7f0bb42..576347b 100644 --- a/packages/host-bin/src/mem-telemetry.ts +++ b/packages/host-bin/src/mem-telemetry.ts @@ -23,8 +23,8 @@ import { memorySampleAttributes, } from "@dispatch/session-orchestrator"; -/** Default periodic sample interval: every 60s. */ -export const DEFAULT_MEMORY_SAMPLE_INTERVAL_MS = 60_000; +/** Default periodic sample interval: every 15s. */ +export const DEFAULT_MEMORY_SAMPLE_INTERVAL_MS = 15_000; /** Default GC interval: every 5 min (longer than the sample interval). */ export const DEFAULT_GC_INTERVAL_MS = 5 * 60_000; @@ -52,7 +52,7 @@ export interface MemoryTelemetryDeps { * baseline. */ readonly getActiveConversationCount: () => number; - /** Periodic sample interval (ms). Defaults to 60s. */ + /** Periodic sample interval (ms). Defaults to 15s. */ readonly sampleIntervalMs?: number; /** GC interval (ms). Defaults to 5 min. */ readonly gcIntervalMs?: number; @@ -77,7 +77,7 @@ export interface MemoryTelemetryHandle { /** * Start periodic memory telemetry. Logs process.memoryUsage() every - * `sampleIntervalMs` (default 60s) tagged with the active-conversation count, + * `sampleIntervalMs` (default 15s) 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. * @@ -97,7 +97,7 @@ export function startMemoryTelemetry(deps: MemoryTelemetryDeps): MemoryTelemetry 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 + // count every 15s. 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(); diff --git a/packages/transport-http/src/extension.ts b/packages/transport-http/src/extension.ts index f46fca5..5d394b8 100644 --- a/packages/transport-http/src/extension.ts +++ b/packages/transport-http/src/extension.ts @@ -11,6 +11,7 @@ import { credentialStoreHandle, heartbeatServiceHandle, lspServiceHandle, + type LspService, mcpServiceHandle, sessionOrchestratorHandle, systemPromptHandle, @@ -27,7 +28,6 @@ export const manifest: Manifest = { "conversation-store", "credential-store", "heartbeat", - "lsp", "mcp", "session-orchestrator", "throughput-store", @@ -95,7 +95,14 @@ export function createTransportHttpExtension(): Extension & { const throughputStore = host.getService(throughputStoreHandle); const warmService = host.getService(cacheWarmHandle); const compactionService = host.getService(compactionHandle); - const lspService = host.getService(lspServiceHandle); + // Optional: the `lsp` extension may be disabled (hot-fix). Wrapped because + // getService throws for an unregistered handle — degrades to no diagnostics. + let lspService: LspService | undefined; + try { + lspService = host.getService(lspServiceHandle); + } catch { + lspService = undefined; + } const mcpService = host.getService(mcpServiceHandle); const systemPromptService = host.getService(systemPromptHandle); const heartbeatService = host.getService(heartbeatServiceHandle); @@ -128,7 +135,7 @@ export function createTransportHttpExtension(): Extension & { throughputStore, warmService, compactionService, - lspService, + ...(lspService !== undefined ? { lspService } : {}), mcpService, systemPromptService, heartbeatService, -- cgit v1.2.3 From d09a4f8ae041536dc7d37a384971058248d7b995 Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Sun, 28 Jun 2026 12:31:18 +0900 Subject: fix(ssh,host-bin): permanent pooled-client error listener + uncaughtException/unhandledRejection guards MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause of the live production crash (exit-1 'Timed out while waiting for handshake'): the pooled ssh2.Client had no permanent 'error' listener after connect, so a post-connect ssh2 error escaped as an uncaught EventEmitter 'error' with no process-level guard. See notes/crash-investigation-findings.md §1. - packages/ssh/src/pool.ts: attach a permanent 'error' listener to the pooled client in buildConnection that sets state=error, logs (alias, message, level), and does not throw; cleanup() no longer removes it. - packages/host-bin/src/main.ts: add process.on('uncaughtException') (graceful shutdown after logging) and process.on('unhandledRejection') (log + continue), both logging message/stack, memory snapshot, activeConversations count, and timestamp so the failure site is observable. --- packages/host-bin/src/main.ts | 35 +++++++++ packages/ssh/src/pool.test.ts | 176 ++++++++++++++++++++++++++++++++++++++++++ packages/ssh/src/pool.ts | 18 +++++ 3 files changed, 229 insertions(+) create mode 100644 packages/ssh/src/pool.test.ts (limited to 'packages/host-bin/src') diff --git a/packages/host-bin/src/main.ts b/packages/host-bin/src/main.ts index 29e402e..70d1cb2 100644 --- a/packages/host-bin/src/main.ts +++ b/packages/host-bin/src/main.ts @@ -31,6 +31,7 @@ import { extension as providerOpenaiCompatExt } from "@dispatch/provider-openai- import { extension as providerUmansExt } from "@dispatch/provider-umans"; import { type MemorySample, + memorySampleAttributes, extension as sessionOrchestratorExt, sessionOrchestratorHandle, } from "@dispatch/session-orchestrator"; @@ -243,8 +244,10 @@ async function boot(): Promise { // 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; + let activeConvCountFn: (() => number) | undefined; try { const orchestrator = host.getHostAPI().getService(sessionOrchestratorHandle); + activeConvCountFn = () => orchestrator.getActiveConversationCount(); memoryTelemetry = startMemoryTelemetry({ logger: logger.child({ extensionId: "mem-telemetry" }), sampleMemory: (): MemorySample => { @@ -280,6 +283,38 @@ async function boot(): Promise { process.on("SIGINT", shutdown); process.on("SIGTERM", shutdown); + const memorySnapshot = () => { + const m = process.memoryUsage(); + return memorySampleAttributes({ + rss: m.rss, + heapUsed: m.heapUsed, + heapTotal: m.heapTotal, + external: m.external, + arrayBuffers: m.arrayBuffers, + }); + }; + + process.on("unhandledRejection", (reason) => { + logger.error("unhandledRejection", { + err: reason, + handler: "unhandledRejection", + activeConversations: activeConvCountFn?.() ?? "unavailable", + timestamp: new Date().toISOString(), + ...memorySnapshot(), + }); + }); + + process.on("uncaughtException", (err) => { + logger.error("uncaughtException", { + err, + handler: "uncaughtException", + activeConversations: activeConvCountFn?.() ?? "unavailable", + timestamp: new Date().toISOString(), + ...memorySnapshot(), + }); + void shutdown(); + }); + logger.info("Dispatch booted"); console.info("Dispatch booted"); } diff --git a/packages/ssh/src/pool.test.ts b/packages/ssh/src/pool.test.ts new file mode 100644 index 0000000..b5d72c8 --- /dev/null +++ b/packages/ssh/src/pool.test.ts @@ -0,0 +1,176 @@ +/** + * Unit tests for the pooled ssh2.Client error lifecycle in pool.ts. + * + * The ssh2.Client (the outermost network edge) is faked via EventEmitter — + * permitted by the constitution (mocking the outermost edge is fine; only + * `@dispatch/*` may not be mocked). These tests prove the permanent + * `'error'` listener survives `cleanup()` (which only removes the connect-time + * onReady/onError) and that a post-connect ssh2 error is captured, logged, and + * non-throwing, with the connection reconnecting on the next acquire. + */ + +import { EventEmitter } from "node:events"; +import type { Logger } from "@dispatch/kernel"; +import type { Computer } from "@dispatch/wire"; +import type { Client } from "ssh2"; +import { afterEach, describe, expect, it } from "vitest"; +import { createSshConnectionPool } from "./pool.js"; + +const computer: Computer = { + alias: "testremote", + hostName: "localhost", + port: 22, + user: "testuser", + identityFile: null, + knownHost: false, +}; + +interface CapturedError { + msg: string; + err: unknown; + alias?: string; + message?: string; + level?: string; + from?: string; + to?: string; +} + +function capturingLogger(calls: CapturedError[]): Logger { + const log: Logger = { + debug: () => undefined, + info: () => undefined, + warn: () => undefined, + error: (msg, attrs) => { + calls.push({ + msg, + err: attrs?.err, + alias: attrs?.alias as string | undefined, + message: attrs?.message as string | undefined, + level: attrs?.level as string | undefined, + from: attrs?.from as string | undefined, + to: attrs?.to as string | undefined, + }); + }, + child: () => log, + span: (name) => ({ + id: name, + log, + setAttributes: () => undefined, + addLink: () => undefined, + child: (n) => ({ id: n, log }) as never, + end: () => undefined, + }), + }; + return log; +} + +/** A fake ssh2.Client: EventEmitter + connect/emd/sftp. `connect` emits 'ready'. */ +function fakeClient(): { client: Client; ee: EventEmitter } { + const ee = new EventEmitter(); + const api = { + connect: () => { + process.nextTick(() => ee.emit("ready")); + }, + end: () => undefined, + sftp: (cb: (err: Error | null, sftp: unknown) => void) => cb(null, {}), + }; + return { client: Object.assign(ee, api) as unknown as Client, ee }; +} + +function makeDeps(client: Client, logger: Logger) { + return { + logger, + homeDir: "/tmp", + knownHostsPath: "/tmp/known_hosts", + readFileText: async (p: string): Promise => { + if (p === "/tmp/known_hosts") return ""; + return "ssh-ed25519 not-encrypted-key"; + }, + appendKnownHosts: async () => undefined, + pathExists: async () => true, + newClient: () => client, + resolveComputer: async () => computer, + }; +} + +describe("SshConnectionPool — permanent pooled-client error listener", () => { + let pool: ReturnType; + + afterEach(async () => { + await pool?.closeAll(); + }); + + it("captures a post-connect 'error' without throwing, sets state=error, and logs", async () => { + const errorCalls: CapturedError[] = []; + const { client, ee } = fakeClient(); + pool = createSshConnectionPool(makeDeps(client, capturingLogger(errorCalls))); + + const conn = await pool.acquire("testremote"); + expect(conn.state).toBe("connected"); + + const sshErr = Object.assign(new Error("Timed out while waiting for handshake"), { + level: "socket", + }); + + expect(() => ee.emit("error", sshErr)).not.toThrow(); + + expect(conn.state).toBe("error"); + expect(conn.error).toBe("Timed out while waiting for handshake"); + expect(errorCalls).toHaveLength(1); + expect(errorCalls[0]?.msg).toBe("ssh: pooled client error"); + expect(errorCalls[0]?.alias).toBe("testremote"); + expect(errorCalls[0]?.message).toBe("Timed out while waiting for handshake"); + expect(errorCalls[0]?.level).toBe("socket"); + expect(errorCalls[0]?.from).toBe("connected"); + expect(errorCalls[0]?.to).toBe("error"); + }); + + it("does NOT remove the permanent listener on successful connect (survives cleanup)", async () => { + const errorCalls: CapturedError[] = []; + const { client, ee } = fakeClient(); + pool = createSshConnectionPool(makeDeps(client, capturingLogger(errorCalls))); + + const conn = await pool.acquire("testremote"); + expect(conn.state).toBe("connected"); + + ee.emit("error", new Error("post-connect drop")); + expect(conn.state).toBe("error"); + expect(errorCalls).toHaveLength(1); + }); + + it("reconnects on the next acquire after a post-connect error", async () => { + const { client, ee } = fakeClient(); + pool = createSshConnectionPool(makeDeps(client, capturingLogger([]))); + + const conn = await pool.acquire("testremote"); + expect(conn.state).toBe("connected"); + + ee.emit("error", new Error("post-connect drop")); + expect(conn.state).toBe("error"); + + const conn2 = await pool.acquire("testremote"); + expect(conn2.state).toBe("connected"); + expect(conn2.error).toBeUndefined(); + }); + + it("logs the connect-time error too (permanent listener fires during handshake)", async () => { + const errorCalls: CapturedError[] = []; + const ee = new EventEmitter(); + const api = { + connect: () => { + process.nextTick(() => ee.emit("error", new Error("connect refused"))); + }, + end: () => undefined, + sftp: (cb: (err: Error | null, sftp: unknown) => void) => cb(null, {}), + }; + const client = Object.assign(ee, api) as unknown as Client; + pool = createSshConnectionPool(makeDeps(client, capturingLogger(errorCalls))); + + const conn = await pool.acquire("testremote"); + expect(conn.state).toBe("error"); + expect(conn.error).toBe("connect refused"); + expect(errorCalls).toHaveLength(1); + expect(errorCalls[0]?.msg).toBe("ssh: pooled client error"); + expect(errorCalls[0]?.message).toBe("connect refused"); + }); +}); diff --git a/packages/ssh/src/pool.ts b/packages/ssh/src/pool.ts index 9acae0e..5d1eedd 100644 --- a/packages/ssh/src/pool.ts +++ b/packages/ssh/src/pool.ts @@ -112,6 +112,24 @@ export function createSshConnectionPool(deps: SshPoolDeps): SshConnectionPool { let sftp: import("ssh2").SFTPWrapper | null = null; let connectPromise: Promise | null = null; + // Permanent error listener — without it, a post-connect ssh2 'error' + // (re-key/keepalive timeout) escapes uncaught → process crash. cleanup() + // in doConnect only removes the connect-time onReady/onError; this persists. + client.on("error", (err: unknown) => { + const from = state.value; + state.value = "error"; + state.error = err instanceof Error ? err.message : String(err); + connectPromise = null; + deps.logger.error("ssh: pooled client error", { + err, + alias, + message: state.error, + level: (err as { level?: string } | null)?.level, + from, + to: "error", + }); + }); + const touch = (): void => { const e = entries.get(alias); if (e !== undefined) e.lastUsedAt = Date.now(); -- cgit v1.2.3