summaryrefslogtreecommitdiffhomepage
path: root/packages/host-bin
diff options
context:
space:
mode:
Diffstat (limited to 'packages/host-bin')
-rw-r--r--packages/host-bin/package.json73
-rw-r--r--packages/host-bin/src/collector-supervisor.test.ts574
-rw-r--r--packages/host-bin/src/collector-supervisor.ts244
-rw-r--r--packages/host-bin/src/config.test.ts152
-rw-r--r--packages/host-bin/src/config.ts108
-rw-r--r--packages/host-bin/src/load-external.test.ts70
-rw-r--r--packages/host-bin/src/load-external.ts46
-rw-r--r--packages/host-bin/src/main.ts436
-rw-r--r--packages/host-bin/src/mem-telemetry.test.ts225
-rw-r--r--packages/host-bin/src/mem-telemetry.ts160
-rw-r--r--packages/host-bin/tsconfig.json132
11 files changed, 1352 insertions, 868 deletions
diff --git a/packages/host-bin/package.json b/packages/host-bin/package.json
index 64f436e..7d3b38c 100644
--- a/packages/host-bin/package.json
+++ b/packages/host-bin/package.json
@@ -1,37 +1,40 @@
{
- "name": "@dispatch/host-bin",
- "version": "0.0.0",
- "type": "module",
- "private": true,
- "dependencies": {
- "@dispatch/kernel": "workspace:*",
- "@dispatch/storage-sqlite": "workspace:*",
- "@dispatch/conversation-store": "workspace:*",
- "@dispatch/auth-apikey": "workspace:*",
- "@dispatch/cache-warming": "workspace:*",
- "@dispatch/credential-store": "workspace:*",
- "@dispatch/exec-backend": "workspace:*",
- "@dispatch/provider-openai-compat": "workspace:*",
- "@dispatch/provider-umans": "workspace:*",
- "@dispatch/message-queue": "workspace:*",
- "@dispatch/mcp": "workspace:*",
- "@dispatch/session-orchestrator": "workspace:*",
- "@dispatch/skills": "workspace:*",
- "@dispatch/ssh": "workspace:*",
- "@dispatch/throughput-store": "workspace:*",
- "@dispatch/todo": "workspace:*",
- "@dispatch/transport-http": "workspace:*",
- "@dispatch/tool-read-file": "workspace:*",
- "@dispatch/tool-shell": "workspace:*",
- "@dispatch/tool-edit-file": "workspace:*",
- "@dispatch/tool-write-file": "workspace:*",
- "@dispatch/tool-web-search": "workspace:*",
- "@dispatch/tool-youtube-transcript": "workspace:*",
- "@dispatch/journal-sink": "workspace:*",
- "@dispatch/lsp": "workspace:*",
- "@dispatch/surface-loaded-extensions": "workspace:*",
- "@dispatch/surface-registry": "workspace:*",
- "@dispatch/transport-ws": "workspace:*",
- "@dispatch/system-prompt": "workspace:*"
- }
+ "name": "@dispatch/host-bin",
+ "version": "0.0.0",
+ "type": "module",
+ "private": true,
+ "dependencies": {
+ "@dispatch/kernel": "workspace:*",
+ "@dispatch/storage-sqlite": "workspace:*",
+ "@dispatch/conversation-store": "workspace:*",
+ "@dispatch/auth-apikey": "workspace:*",
+ "@dispatch/cache-warming": "workspace:*",
+ "@dispatch/credential-store": "workspace:*",
+ "@dispatch/exec-backend": "workspace:*",
+ "@dispatch/heartbeat": "workspace:*",
+ "@dispatch/provider-openai-compat": "workspace:*",
+ "@dispatch/provider-concurrency": "workspace:*",
+ "@dispatch/provider-umans": "workspace:*",
+ "@dispatch/message-queue": "workspace:*",
+ "@dispatch/mcp": "workspace:*",
+ "@dispatch/session-orchestrator": "workspace:*",
+ "@dispatch/skills": "workspace:*",
+ "@dispatch/ssh": "workspace:*",
+ "@dispatch/throughput-store": "workspace:*",
+ "@dispatch/todo": "workspace:*",
+ "@dispatch/transport-http": "workspace:*",
+ "@dispatch/tool-read-file": "workspace:*",
+ "@dispatch/tool-shell": "workspace:*",
+ "@dispatch/tool-edit-file": "workspace:*",
+ "@dispatch/tool-write-file": "workspace:*",
+ "@dispatch/tool-web-search": "workspace:*",
+ "@dispatch/tool-youtube-transcript": "workspace:*",
+ "@dispatch/journal-sink": "workspace:*",
+ "@dispatch/lsp": "workspace:*",
+ "@dispatch/surface-loaded-extensions": "workspace:*",
+ "@dispatch/surface-registry": "workspace:*",
+ "@dispatch/transport-ws": "workspace:*",
+ "@dispatch/system-prompt": "workspace:*",
+ "@dispatch/vision-handoff": "workspace:*"
+ }
}
diff --git a/packages/host-bin/src/collector-supervisor.test.ts b/packages/host-bin/src/collector-supervisor.test.ts
index 8c1f104..f590806 100644
--- a/packages/host-bin/src/collector-supervisor.test.ts
+++ b/packages/host-bin/src/collector-supervisor.test.ts
@@ -2,300 +2,300 @@ import { describe, expect, it } from "vitest";
import { type ChildHandle, createCollectorSupervisor } from "./collector-supervisor.js";
interface FakeChild {
- readonly handle: ChildHandle;
- resolveExit: (code: number) => void;
- readonly signals: string[];
+ readonly handle: ChildHandle;
+ resolveExit: (code: number) => void;
+ readonly signals: string[];
}
function createFakeChild(code = 0): FakeChild {
- let resolveExit!: (code: number) => void;
- const exited = new Promise<number>((r) => {
- resolveExit = r;
- });
- const signals: string[] = [];
- const handle: ChildHandle = {
- kill: (signal?: string) => {
- signals.push(signal ?? "SIGTERM");
- if (signal === "SIGKILL") resolveExit(code);
- },
- exited,
- };
- return { handle, resolveExit, signals };
+ let resolveExit!: (code: number) => void;
+ const exited = new Promise<number>((r) => {
+ resolveExit = r;
+ });
+ const signals: string[] = [];
+ const handle: ChildHandle = {
+ kill: (signal?: string) => {
+ signals.push(signal ?? "SIGTERM");
+ if (signal === "SIGKILL") resolveExit(code);
+ },
+ exited,
+ };
+ return { handle, resolveExit, signals };
}
function createFakeLogger() {
- const msgs: Array<{ level: string; msg: string }> = [];
- return {
- msgs,
- debug: () => {},
- info: (msg: string) => msgs.push({ level: "info", msg }),
- warn: (msg: string) => msgs.push({ level: "warn", msg }),
- error: () => {},
- child: () => createFakeLogger(),
- span: () => ({
- id: "s",
- log: createFakeLogger(),
- setAttributes: () => {},
- addLink: () => {},
- child: () => ({}) as never,
- end: () => {},
- }),
- };
+ const msgs: Array<{ level: string; msg: string }> = [];
+ return {
+ msgs,
+ debug: () => {},
+ info: (msg: string) => msgs.push({ level: "info", msg }),
+ warn: (msg: string) => msgs.push({ level: "warn", msg }),
+ error: () => {},
+ child: () => createFakeLogger(),
+ span: () => ({
+ id: "s",
+ log: createFakeLogger(),
+ setAttributes: () => {},
+ addLink: () => {},
+ child: () => ({}) as never,
+ end: () => {},
+ }),
+ };
}
describe("createCollectorSupervisor", () => {
- const DEFAULTS = {
- journalPath: "/tmp/journal.ndjson",
- dbPath: "/tmp/traces.db",
- };
-
- it("start() spawns with the correct command and args", () => {
- let capturedCmd: string[] = [];
- const children: FakeChild[] = [];
- const spawn = (cmd: string[]) => {
- capturedCmd = cmd;
- const child = createFakeChild();
- children.push(child);
- return child.handle;
- };
-
- const supervisor = createCollectorSupervisor({
- ...DEFAULTS,
- spawn,
- logger: createFakeLogger() as never,
- });
- supervisor.start();
-
- expect(capturedCmd).toEqual([
- "bun",
- "packages/observability-collector/src/main.ts",
- "--journal",
- "/tmp/journal.ndjson",
- "--db",
- "/tmp/traces.db",
- ]);
- });
-
- it("start() passes --interval when provided", () => {
- let capturedCmd: string[] = [];
- const spawn = (cmd: string[]) => {
- capturedCmd = cmd;
- return createFakeChild().handle;
- };
-
- const supervisor = createCollectorSupervisor({
- ...DEFAULTS,
- interval: 500,
- spawn,
- logger: createFakeLogger() as never,
- });
- supervisor.start();
-
- expect(capturedCmd).toContain("--interval");
- expect(capturedCmd).toContain("500");
- });
-
- it("unexpected child exit respawns the collector", async () => {
- const children: FakeChild[] = [];
- let spawnCount = 0;
- const spawn = () => {
- spawnCount++;
- const child = createFakeChild();
- children.push(child);
- return child.handle;
- };
-
- const time = 0;
- const now = () => time;
- const delayResolvers: Array<() => void> = [];
- const delay = (_ms: number) =>
- new Promise<void>((r) => {
- delayResolvers.push(r);
- });
-
- const supervisor = createCollectorSupervisor({
- ...DEFAULTS,
- spawn,
- logger: createFakeLogger() as never,
- now,
- delay,
- });
- supervisor.start();
-
- expect(spawnCount).toBe(1);
-
- // Simulate unexpected exit
- children[0]?.resolveExit(1);
- await Promise.resolve();
- await Promise.resolve();
-
- // Trigger the backoff delay resolver
- expect(delayResolvers.length).toBe(1);
- delayResolvers[0]?.();
- await Promise.resolve();
- await Promise.resolve();
-
- expect(spawnCount).toBe(2);
- });
-
- it("restart guard caps respawns in a tight loop", async () => {
- const children: FakeChild[] = [];
- let spawnCount = 0;
- const spawn = () => {
- spawnCount++;
- const child = createFakeChild();
- children.push(child);
- return child.handle;
- };
-
- const time = 0;
- const now = () => time;
- const delayResolvers: Array<() => void> = [];
- const delay = (_ms: number) =>
- new Promise<void>((r) => {
- delayResolvers.push(r);
- });
-
- const logger = createFakeLogger();
- const supervisor = createCollectorSupervisor({
- ...DEFAULTS,
- spawn,
- logger: logger as never,
- now,
- delay,
- });
- supervisor.start();
-
- // Simulate rapid crashes (within the restart window)
- for (let i = 0; i < 5; i++) {
- children[i]?.resolveExit(1);
- await Promise.resolve();
- await Promise.resolve();
- if (delayResolvers.length > i) {
- delayResolvers[i]?.();
- await Promise.resolve();
- await Promise.resolve();
- }
- }
-
- // Should have spawned 6 times (1 initial + 5 restarts)
- expect(spawnCount).toBe(6);
-
- // 6th child also dies — should NOT respawn (cap reached)
- children[5]?.resolveExit(1);
- await Promise.resolve();
- await Promise.resolve();
-
- // spawnCount should still be 6
- expect(spawnCount).toBe(6);
- expect(logger.msgs.some((m) => m.msg === "Collector restart cap reached; giving up")).toBe(
- true,
- );
- });
-
- it("stop() sends SIGTERM and does not respawn", async () => {
- const child = createFakeChild();
- const spawn = () => child.handle;
-
- const supervisor = createCollectorSupervisor({
- ...DEFAULTS,
- spawn,
- logger: createFakeLogger() as never,
- });
- supervisor.start();
-
- // Resolve exit after SIGTERM (simulating graceful shutdown)
- const stopPromise = supervisor.stop();
- child.resolveExit(0);
- await stopPromise;
-
- expect(child.signals).toContain("SIGTERM");
- });
-
- it("stop() sends SIGKILL when child does not exit in time", async () => {
- const child = createFakeChild();
- const spawn = () => child.handle;
-
- const time = 0;
- const now = () => time;
- const delayResolvers: Array<() => void> = [];
- const delay = (_ms: number) =>
- new Promise<void>((r) => {
- delayResolvers.push(r);
- });
-
- const supervisor = createCollectorSupervisor({
- ...DEFAULTS,
- spawn,
- logger: createFakeLogger() as never,
- now,
- delay,
- });
- supervisor.start();
-
- const stopPromise = supervisor.stop();
-
- // Don't resolve exit — simulate hung child
- // Resolve the timeout delay instead
- expect(delayResolvers.length).toBe(1);
- delayResolvers[0]?.();
- await stopPromise;
-
- expect(child.signals).toContain("SIGTERM");
- expect(child.signals).toContain("SIGKILL");
- });
-
- it("stop() does not respawn after unexpected exit during stop", async () => {
- const children: FakeChild[] = [];
- let spawnCount = 0;
- const spawn = () => {
- spawnCount++;
- const child = createFakeChild();
- children.push(child);
- return child.handle;
- };
-
- const supervisor = createCollectorSupervisor({
- ...DEFAULTS,
- spawn,
- logger: createFakeLogger() as never,
- });
- supervisor.start();
- expect(spawnCount).toBe(1);
-
- // Child exits during stop — supervisor is already stopping
- const stopPromise = supervisor.stop();
- children[0]?.resolveExit(1);
- await stopPromise;
-
- // Should NOT have respawned
- expect(spawnCount).toBe(1);
- });
-
- it("spawn throwing does not throw to caller", () => {
- const spawn = () => {
- throw new Error("spawn failed");
- };
-
- const logger = createFakeLogger();
- const supervisor = createCollectorSupervisor({
- ...DEFAULTS,
- spawn,
- logger: logger as never,
- });
-
- expect(() => supervisor.start()).not.toThrow();
- expect(logger.msgs.some((m) => m.msg === "Failed to spawn collector")).toBe(true);
- });
-
- it("stop() is safe to call when no child was started", async () => {
- const spawn = () => createFakeChild().handle;
- const supervisor = createCollectorSupervisor({
- ...DEFAULTS,
- spawn,
- logger: createFakeLogger() as never,
- });
-
- await expect(supervisor.stop()).resolves.toBeUndefined();
- });
+ const DEFAULTS = {
+ journalPath: "/tmp/journal.ndjson",
+ dbPath: "/tmp/traces.db",
+ };
+
+ it("start() spawns with the correct command and args", () => {
+ let capturedCmd: string[] = [];
+ const children: FakeChild[] = [];
+ const spawn = (cmd: string[]) => {
+ capturedCmd = cmd;
+ const child = createFakeChild();
+ children.push(child);
+ return child.handle;
+ };
+
+ const supervisor = createCollectorSupervisor({
+ ...DEFAULTS,
+ spawn,
+ logger: createFakeLogger() as never,
+ });
+ supervisor.start();
+
+ expect(capturedCmd).toEqual([
+ "bun",
+ "packages/observability-collector/src/main.ts",
+ "--journal",
+ "/tmp/journal.ndjson",
+ "--db",
+ "/tmp/traces.db",
+ ]);
+ });
+
+ it("start() passes --interval when provided", () => {
+ let capturedCmd: string[] = [];
+ const spawn = (cmd: string[]) => {
+ capturedCmd = cmd;
+ return createFakeChild().handle;
+ };
+
+ const supervisor = createCollectorSupervisor({
+ ...DEFAULTS,
+ interval: 500,
+ spawn,
+ logger: createFakeLogger() as never,
+ });
+ supervisor.start();
+
+ expect(capturedCmd).toContain("--interval");
+ expect(capturedCmd).toContain("500");
+ });
+
+ it("unexpected child exit respawns the collector", async () => {
+ const children: FakeChild[] = [];
+ let spawnCount = 0;
+ const spawn = () => {
+ spawnCount++;
+ const child = createFakeChild();
+ children.push(child);
+ return child.handle;
+ };
+
+ const time = 0;
+ const now = () => time;
+ const delayResolvers: Array<() => void> = [];
+ const delay = (_ms: number) =>
+ new Promise<void>((r) => {
+ delayResolvers.push(r);
+ });
+
+ const supervisor = createCollectorSupervisor({
+ ...DEFAULTS,
+ spawn,
+ logger: createFakeLogger() as never,
+ now,
+ delay,
+ });
+ supervisor.start();
+
+ expect(spawnCount).toBe(1);
+
+ // Simulate unexpected exit
+ children[0]?.resolveExit(1);
+ await Promise.resolve();
+ await Promise.resolve();
+
+ // Trigger the backoff delay resolver
+ expect(delayResolvers.length).toBe(1);
+ delayResolvers[0]?.();
+ await Promise.resolve();
+ await Promise.resolve();
+
+ expect(spawnCount).toBe(2);
+ });
+
+ it("restart guard caps respawns in a tight loop", async () => {
+ const children: FakeChild[] = [];
+ let spawnCount = 0;
+ const spawn = () => {
+ spawnCount++;
+ const child = createFakeChild();
+ children.push(child);
+ return child.handle;
+ };
+
+ const time = 0;
+ const now = () => time;
+ const delayResolvers: Array<() => void> = [];
+ const delay = (_ms: number) =>
+ new Promise<void>((r) => {
+ delayResolvers.push(r);
+ });
+
+ const logger = createFakeLogger();
+ const supervisor = createCollectorSupervisor({
+ ...DEFAULTS,
+ spawn,
+ logger: logger as never,
+ now,
+ delay,
+ });
+ supervisor.start();
+
+ // Simulate rapid crashes (within the restart window)
+ for (let i = 0; i < 5; i++) {
+ children[i]?.resolveExit(1);
+ await Promise.resolve();
+ await Promise.resolve();
+ if (delayResolvers.length > i) {
+ delayResolvers[i]?.();
+ await Promise.resolve();
+ await Promise.resolve();
+ }
+ }
+
+ // Should have spawned 6 times (1 initial + 5 restarts)
+ expect(spawnCount).toBe(6);
+
+ // 6th child also dies — should NOT respawn (cap reached)
+ children[5]?.resolveExit(1);
+ await Promise.resolve();
+ await Promise.resolve();
+
+ // spawnCount should still be 6
+ expect(spawnCount).toBe(6);
+ expect(logger.msgs.some((m) => m.msg === "Collector restart cap reached; giving up")).toBe(
+ true,
+ );
+ });
+
+ it("stop() sends SIGTERM and does not respawn", async () => {
+ const child = createFakeChild();
+ const spawn = () => child.handle;
+
+ const supervisor = createCollectorSupervisor({
+ ...DEFAULTS,
+ spawn,
+ logger: createFakeLogger() as never,
+ });
+ supervisor.start();
+
+ // Resolve exit after SIGTERM (simulating graceful shutdown)
+ const stopPromise = supervisor.stop();
+ child.resolveExit(0);
+ await stopPromise;
+
+ expect(child.signals).toContain("SIGTERM");
+ });
+
+ it("stop() sends SIGKILL when child does not exit in time", async () => {
+ const child = createFakeChild();
+ const spawn = () => child.handle;
+
+ const time = 0;
+ const now = () => time;
+ const delayResolvers: Array<() => void> = [];
+ const delay = (_ms: number) =>
+ new Promise<void>((r) => {
+ delayResolvers.push(r);
+ });
+
+ const supervisor = createCollectorSupervisor({
+ ...DEFAULTS,
+ spawn,
+ logger: createFakeLogger() as never,
+ now,
+ delay,
+ });
+ supervisor.start();
+
+ const stopPromise = supervisor.stop();
+
+ // Don't resolve exit — simulate hung child
+ // Resolve the timeout delay instead
+ expect(delayResolvers.length).toBe(1);
+ delayResolvers[0]?.();
+ await stopPromise;
+
+ expect(child.signals).toContain("SIGTERM");
+ expect(child.signals).toContain("SIGKILL");
+ });
+
+ it("stop() does not respawn after unexpected exit during stop", async () => {
+ const children: FakeChild[] = [];
+ let spawnCount = 0;
+ const spawn = () => {
+ spawnCount++;
+ const child = createFakeChild();
+ children.push(child);
+ return child.handle;
+ };
+
+ const supervisor = createCollectorSupervisor({
+ ...DEFAULTS,
+ spawn,
+ logger: createFakeLogger() as never,
+ });
+ supervisor.start();
+ expect(spawnCount).toBe(1);
+
+ // Child exits during stop — supervisor is already stopping
+ const stopPromise = supervisor.stop();
+ children[0]?.resolveExit(1);
+ await stopPromise;
+
+ // Should NOT have respawned
+ expect(spawnCount).toBe(1);
+ });
+
+ it("spawn throwing does not throw to caller", () => {
+ const spawn = () => {
+ throw new Error("spawn failed");
+ };
+
+ const logger = createFakeLogger();
+ const supervisor = createCollectorSupervisor({
+ ...DEFAULTS,
+ spawn,
+ logger: logger as never,
+ });
+
+ expect(() => supervisor.start()).not.toThrow();
+ expect(logger.msgs.some((m) => m.msg === "Failed to spawn collector")).toBe(true);
+ });
+
+ it("stop() is safe to call when no child was started", async () => {
+ const spawn = () => createFakeChild().handle;
+ const supervisor = createCollectorSupervisor({
+ ...DEFAULTS,
+ spawn,
+ logger: createFakeLogger() as never,
+ });
+
+ await expect(supervisor.stop()).resolves.toBeUndefined();
+ });
});
diff --git a/packages/host-bin/src/collector-supervisor.ts b/packages/host-bin/src/collector-supervisor.ts
index 7b893b9..ff51b86 100644
--- a/packages/host-bin/src/collector-supervisor.ts
+++ b/packages/host-bin/src/collector-supervisor.ts
@@ -1,18 +1,18 @@
import type { Logger } from "@dispatch/kernel";
export interface ChildHandle {
- readonly kill: (signal?: string) => void;
- readonly exited: Promise<number>;
+ readonly kill: (signal?: string) => void;
+ readonly exited: Promise<number>;
}
export interface SupervisorDeps {
- readonly spawn: (cmd: string[]) => ChildHandle;
- readonly journalPath: string;
- readonly dbPath: string;
- readonly interval?: number;
- readonly logger: Logger;
- readonly now?: () => number;
- readonly delay?: (ms: number) => Promise<void>;
+ readonly spawn: (cmd: string[]) => ChildHandle;
+ readonly journalPath: string;
+ readonly dbPath: string;
+ readonly interval?: number;
+ readonly logger: Logger;
+ readonly now?: () => number;
+ readonly delay?: (ms: number) => Promise<void>;
}
const RESTART_WINDOW_MS = 10_000;
@@ -21,118 +21,118 @@ const BACKOFF_BASE_MS = 500;
const STOP_TIMEOUT_MS = 5_000;
export function createCollectorSupervisor(deps: SupervisorDeps): {
- start: () => void;
- stop: () => Promise<void>;
+ start: () => void;
+ stop: () => Promise<void>;
} {
- const {
- spawn,
- journalPath,
- dbPath,
- interval,
- logger,
- now = () => Date.now(),
- delay = (ms) => new Promise((r) => setTimeout(r, ms)),
- } = deps;
-
- let child: ChildHandle | null = null;
- let stopping = false;
- const restartTimestamps: number[] = [];
-
- function buildCmd(): string[] {
- const cmd = [
- "bun",
- "packages/observability-collector/src/main.ts",
- "--journal",
- journalPath,
- "--db",
- dbPath,
- ];
- if (interval !== undefined) {
- cmd.push("--interval", String(interval));
- }
- return cmd;
- }
-
- function pruneOldRestarts(): void {
- const cutoff = now() - RESTART_WINDOW_MS;
- while (restartTimestamps.length > 0) {
- const oldest = restartTimestamps[0];
- if (oldest === undefined || oldest > cutoff) break;
- restartTimestamps.shift();
- }
- }
-
- function shouldRestart(): boolean {
- pruneOldRestarts();
- return restartTimestamps.length < MAX_RESTARTS_IN_WINDOW;
- }
-
- function getBackoffMs(): number {
- return BACKOFF_BASE_MS * 2 ** restartTimestamps.length;
- }
-
- function onChildExit(code: number): void {
- child = null;
- if (stopping) return;
-
- logger.warn("Collector exited unexpectedly", { code } as never);
- if (!shouldRestart()) {
- logger.warn("Collector restart cap reached; giving up", {
- restarts: restartTimestamps.length,
- windowMs: RESTART_WINDOW_MS,
- } as never);
- return;
- }
-
- restartTimestamps.push(now());
- const backoff = getBackoffMs();
- logger.info("Restarting collector after backoff", { backoffMs: backoff } as never);
- delay(backoff)
- .then(() => {
- if (!stopping) spawnChild();
- })
- .catch(() => {});
- }
-
- function spawnChild(): void {
- try {
- const handle = spawn(buildCmd());
- child = handle;
- logger.info("Collector started");
- handle.exited.then(
- (code) => onChildExit(code),
- () => {},
- );
- } catch (err) {
- logger.warn("Failed to spawn collector", { err } as never);
- }
- }
-
- function start(): void {
- spawnChild();
- }
-
- async function stop(): Promise<void> {
- stopping = true;
- if (!child) return;
-
- const handle = child;
- handle.kill("SIGTERM");
-
- let resolved = false;
- const exitedPromise = handle.exited.then(() => {
- resolved = true;
- });
-
- const timeoutPromise = delay(STOP_TIMEOUT_MS).then(() => {
- if (!resolved) {
- handle.kill("SIGKILL");
- return handle.exited;
- }
- });
-
- await Promise.race([exitedPromise, timeoutPromise]);
- }
-
- return { start, stop };
+ const {
+ spawn,
+ journalPath,
+ dbPath,
+ interval,
+ logger,
+ now = () => Date.now(),
+ delay = (ms) => new Promise((r) => setTimeout(r, ms)),
+ } = deps;
+
+ let child: ChildHandle | null = null;
+ let stopping = false;
+ const restartTimestamps: number[] = [];
+
+ function buildCmd(): string[] {
+ const cmd = [
+ "bun",
+ "packages/observability-collector/src/main.ts",
+ "--journal",
+ journalPath,
+ "--db",
+ dbPath,
+ ];
+ if (interval !== undefined) {
+ cmd.push("--interval", String(interval));
+ }
+ return cmd;
+ }
+
+ function pruneOldRestarts(): void {
+ const cutoff = now() - RESTART_WINDOW_MS;
+ while (restartTimestamps.length > 0) {
+ const oldest = restartTimestamps[0];
+ if (oldest === undefined || oldest > cutoff) break;
+ restartTimestamps.shift();
+ }
+ }
+
+ function shouldRestart(): boolean {
+ pruneOldRestarts();
+ return restartTimestamps.length < MAX_RESTARTS_IN_WINDOW;
+ }
+
+ function getBackoffMs(): number {
+ return BACKOFF_BASE_MS * 2 ** restartTimestamps.length;
+ }
+
+ function onChildExit(code: number): void {
+ child = null;
+ if (stopping) return;
+
+ logger.warn("Collector exited unexpectedly", { code } as never);
+ if (!shouldRestart()) {
+ logger.warn("Collector restart cap reached; giving up", {
+ restarts: restartTimestamps.length,
+ windowMs: RESTART_WINDOW_MS,
+ } as never);
+ return;
+ }
+
+ restartTimestamps.push(now());
+ const backoff = getBackoffMs();
+ logger.info("Restarting collector after backoff", { backoffMs: backoff } as never);
+ delay(backoff)
+ .then(() => {
+ if (!stopping) spawnChild();
+ })
+ .catch(() => {});
+ }
+
+ function spawnChild(): void {
+ try {
+ const handle = spawn(buildCmd());
+ child = handle;
+ logger.info("Collector started");
+ handle.exited.then(
+ (code) => onChildExit(code),
+ () => {},
+ );
+ } catch (err) {
+ logger.warn("Failed to spawn collector", { err } as never);
+ }
+ }
+
+ function start(): void {
+ spawnChild();
+ }
+
+ async function stop(): Promise<void> {
+ stopping = true;
+ if (!child) return;
+
+ const handle = child;
+ handle.kill("SIGTERM");
+
+ let resolved = false;
+ const exitedPromise = handle.exited.then(() => {
+ resolved = true;
+ });
+
+ const timeoutPromise = delay(STOP_TIMEOUT_MS).then(() => {
+ if (!resolved) {
+ handle.kill("SIGKILL");
+ return handle.exited;
+ }
+ });
+
+ await Promise.race([exitedPromise, timeoutPromise]);
+ }
+
+ return { start, stop };
}
diff --git a/packages/host-bin/src/config.test.ts b/packages/host-bin/src/config.test.ts
index fc74a79..3a9f463 100644
--- a/packages/host-bin/src/config.test.ts
+++ b/packages/host-bin/src/config.test.ts
@@ -2,96 +2,96 @@ import { describe, expect, it } from "vitest";
import { configMapToAccess, envToConfigMap } from "./config.js";
describe("envToConfigMap", () => {
- it("maps DISPATCH_API_KEY to provider.openai-compat.apiKey", () => {
- const result = envToConfigMap({ DISPATCH_API_KEY: "sk-test-123" });
- expect(result["provider.openai-compat.apiKey"]).toBe("sk-test-123");
- });
+ it("maps DISPATCH_API_KEY to provider.openai-compat.apiKey", () => {
+ const result = envToConfigMap({ DISPATCH_API_KEY: "sk-test-123" });
+ expect(result["provider.openai-compat.apiKey"]).toBe("sk-test-123");
+ });
- it("maps DISPATCH_BASE_URL to provider.openai-compat.baseURL", () => {
- const result = envToConfigMap({ DISPATCH_BASE_URL: "https://custom.api/v1" });
- expect(result["provider.openai-compat.baseURL"]).toBe("https://custom.api/v1");
- });
+ it("maps DISPATCH_BASE_URL to provider.openai-compat.baseURL", () => {
+ const result = envToConfigMap({ DISPATCH_BASE_URL: "https://custom.api/v1" });
+ expect(result["provider.openai-compat.baseURL"]).toBe("https://custom.api/v1");
+ });
- it("maps DISPATCH_MODEL to provider.openai-compat.model", () => {
- const result = envToConfigMap({ DISPATCH_MODEL: "gpt-4" });
- expect(result["provider.openai-compat.model"]).toBe("gpt-4");
- });
+ it("maps DISPATCH_MODEL to provider.openai-compat.model", () => {
+ const result = envToConfigMap({ DISPATCH_MODEL: "gpt-4" });
+ expect(result["provider.openai-compat.model"]).toBe("gpt-4");
+ });
- it("maps all three env vars together", () => {
- const result = envToConfigMap({
- DISPATCH_API_KEY: "key",
- DISPATCH_BASE_URL: "https://api.example.com",
- DISPATCH_MODEL: "my-model",
- });
- expect(result).toEqual({
- "provider.openai-compat.apiKey": "key",
- "provider.openai-compat.baseURL": "https://api.example.com",
- "provider.openai-compat.model": "my-model",
- });
- });
+ it("maps all three env vars together", () => {
+ const result = envToConfigMap({
+ DISPATCH_API_KEY: "key",
+ DISPATCH_BASE_URL: "https://api.example.com",
+ DISPATCH_MODEL: "my-model",
+ });
+ expect(result).toEqual({
+ "provider.openai-compat.apiKey": "key",
+ "provider.openai-compat.baseURL": "https://api.example.com",
+ "provider.openai-compat.model": "my-model",
+ });
+ });
- it("returns empty map when no relevant env vars are set", () => {
- const result = envToConfigMap({ HOME: "/home/user", PATH: "/usr/bin" });
- expect(result).toEqual({});
- });
+ it("returns empty map when no relevant env vars are set", () => {
+ const result = envToConfigMap({ HOME: "/home/user", PATH: "/usr/bin" });
+ expect(result).toEqual({});
+ });
- it("skips undefined env vars", () => {
- const result = envToConfigMap({ DISPATCH_API_KEY: undefined });
- expect(result).toEqual({});
- });
+ it("skips undefined env vars", () => {
+ const result = envToConfigMap({ DISPATCH_API_KEY: undefined });
+ expect(result).toEqual({});
+ });
- it("includes only set vars when some are missing", () => {
- const result = envToConfigMap({
- DISPATCH_API_KEY: "key",
- DISPATCH_MODEL: undefined,
- });
- expect(result).toEqual({
- "provider.openai-compat.apiKey": "key",
- });
- expect(result["provider.openai-compat.baseURL"]).toBeUndefined();
- expect(result["provider.openai-compat.model"]).toBeUndefined();
- });
+ it("includes only set vars when some are missing", () => {
+ const result = envToConfigMap({
+ DISPATCH_API_KEY: "key",
+ DISPATCH_MODEL: undefined,
+ });
+ expect(result).toEqual({
+ "provider.openai-compat.apiKey": "key",
+ });
+ expect(result["provider.openai-compat.baseURL"]).toBeUndefined();
+ expect(result["provider.openai-compat.model"]).toBeUndefined();
+ });
- it("maps SURFACE_WS_PORT to surfaceWsPort", () => {
- const result = envToConfigMap({ SURFACE_WS_PORT: "24206" });
- expect(result.surfaceWsPort).toBe(24206);
- });
+ it("maps SURFACE_WS_PORT to surfaceWsPort", () => {
+ const result = envToConfigMap({ SURFACE_WS_PORT: "24206" });
+ expect(result.surfaceWsPort).toBe(24206);
+ });
- it("ignores a non-numeric SURFACE_WS_PORT", () => {
- const result = envToConfigMap({ SURFACE_WS_PORT: "abc" });
- expect(result.surfaceWsPort).toBeUndefined();
- });
+ it("ignores a non-numeric SURFACE_WS_PORT", () => {
+ const result = envToConfigMap({ SURFACE_WS_PORT: "abc" });
+ expect(result.surfaceWsPort).toBeUndefined();
+ });
- it("ignores a non-positive SURFACE_WS_PORT", () => {
- const result = envToConfigMap({ SURFACE_WS_PORT: "0" });
- expect(result.surfaceWsPort).toBeUndefined();
- });
+ it("ignores a non-positive SURFACE_WS_PORT", () => {
+ const result = envToConfigMap({ SURFACE_WS_PORT: "0" });
+ expect(result.surfaceWsPort).toBeUndefined();
+ });
- it("omits surfaceWsPort when SURFACE_WS_PORT is unset", () => {
- const result = envToConfigMap({});
- expect(result.surfaceWsPort).toBeUndefined();
- });
+ it("omits surfaceWsPort when SURFACE_WS_PORT is unset", () => {
+ const result = envToConfigMap({});
+ expect(result.surfaceWsPort).toBeUndefined();
+ });
});
describe("configMapToAccess", () => {
- it("returns value for existing key", () => {
- const access = configMapToAccess({ "provider.openai-compat.apiKey": "sk-123" });
- expect(access.get("provider.openai-compat.apiKey")).toBe("sk-123");
- });
+ it("returns value for existing key", () => {
+ const access = configMapToAccess({ "provider.openai-compat.apiKey": "sk-123" });
+ expect(access.get("provider.openai-compat.apiKey")).toBe("sk-123");
+ });
- it("returns undefined for missing key", () => {
- const access = configMapToAccess({});
- expect(access.get("nonexistent")).toBeUndefined();
- });
+ it("returns undefined for missing key", () => {
+ const access = configMapToAccess({});
+ expect(access.get("nonexistent")).toBeUndefined();
+ });
- it("returns typed value", () => {
- const access = configMapToAccess({ "some.number": 42 });
- expect(access.get<number>("some.number")).toBe(42);
- });
+ it("returns typed value", () => {
+ const access = configMapToAccess({ "some.number": 42 });
+ expect(access.get<number>("some.number")).toBe(42);
+ });
- it("getAll returns the full map", () => {
- const map = { a: 1, b: "two" };
- const access = configMapToAccess(map);
- expect(access.getAll()).toEqual(map);
- });
+ it("getAll returns the full map", () => {
+ const map = { a: 1, b: "two" };
+ const access = configMapToAccess(map);
+ expect(access.getAll()).toEqual(map);
+ });
});
diff --git a/packages/host-bin/src/config.ts b/packages/host-bin/src/config.ts
index 9a22c00..f69d799 100644
--- a/packages/host-bin/src/config.ts
+++ b/packages/host-bin/src/config.ts
@@ -1,62 +1,62 @@
import type { ConfigAccess } from "@dispatch/kernel";
export function envToConfigMap(
- env: Readonly<Record<string, string | undefined>>,
+ env: Readonly<Record<string, string | undefined>>,
): Record<string, unknown> {
- const map: Record<string, unknown> = {};
-
- const apiKey = env.DISPATCH_API_KEY;
- if (apiKey !== undefined) {
- map["provider.openai-compat.apiKey"] = apiKey;
- }
-
- const baseURL = env.DISPATCH_BASE_URL;
- if (baseURL !== undefined) {
- map["provider.openai-compat.baseURL"] = baseURL;
- }
-
- const model = env.DISPATCH_MODEL;
- if (model !== undefined) {
- map["provider.openai-compat.model"] = model;
- }
-
- // Optional settings consumed by external extensions (e.g. the Claude provider).
- const anthropicModel = env.DISPATCH_ANTHROPIC_MODEL;
- if (anthropicModel !== undefined) {
- map["provider.anthropic.model"] = anthropicModel;
- }
-
- const claudeCredentialKey = env.DISPATCH_CLAUDE_CREDENTIAL_KEY;
- if (claudeCredentialKey !== undefined) {
- map["claude.credentialKey"] = claudeCredentialKey;
- }
-
- const httpPort = env.BACKEND_PORT ?? env.PORT;
- if (httpPort !== undefined) {
- const n = Number(httpPort);
- if (Number.isFinite(n) && n > 0) {
- map.httpPort = n;
- }
- }
-
- const surfaceWsPort = env.SURFACE_WS_PORT;
- if (surfaceWsPort !== undefined) {
- const n = Number(surfaceWsPort);
- if (Number.isFinite(n) && n > 0) {
- map.surfaceWsPort = n;
- }
- }
-
- return map;
+ const map: Record<string, unknown> = {};
+
+ const apiKey = env.DISPATCH_API_KEY;
+ if (apiKey !== undefined) {
+ map["provider.openai-compat.apiKey"] = apiKey;
+ }
+
+ const baseURL = env.DISPATCH_BASE_URL;
+ if (baseURL !== undefined) {
+ map["provider.openai-compat.baseURL"] = baseURL;
+ }
+
+ const model = env.DISPATCH_MODEL;
+ if (model !== undefined) {
+ map["provider.openai-compat.model"] = model;
+ }
+
+ // Optional settings consumed by external extensions (e.g. the Claude provider).
+ const anthropicModel = env.DISPATCH_ANTHROPIC_MODEL;
+ if (anthropicModel !== undefined) {
+ map["provider.anthropic.model"] = anthropicModel;
+ }
+
+ const claudeCredentialKey = env.DISPATCH_CLAUDE_CREDENTIAL_KEY;
+ if (claudeCredentialKey !== undefined) {
+ map["claude.credentialKey"] = claudeCredentialKey;
+ }
+
+ const httpPort = env.BACKEND_PORT ?? env.PORT;
+ if (httpPort !== undefined) {
+ const n = Number(httpPort);
+ if (Number.isFinite(n) && n > 0) {
+ map.httpPort = n;
+ }
+ }
+
+ const surfaceWsPort = env.SURFACE_WS_PORT;
+ if (surfaceWsPort !== undefined) {
+ const n = Number(surfaceWsPort);
+ if (Number.isFinite(n) && n > 0) {
+ map.surfaceWsPort = n;
+ }
+ }
+
+ return map;
}
export function configMapToAccess(map: Readonly<Record<string, unknown>>): ConfigAccess {
- return {
- get<T = unknown>(key: string): T | undefined {
- return map[key] as T | undefined;
- },
- getAll(): Readonly<Record<string, unknown>> {
- return map;
- },
- };
+ return {
+ get<T = unknown>(key: string): T | undefined {
+ return map[key] as T | undefined;
+ },
+ getAll(): Readonly<Record<string, unknown>> {
+ return map;
+ },
+ };
}
diff --git a/packages/host-bin/src/load-external.test.ts b/packages/host-bin/src/load-external.test.ts
index 38e3622..a373e3d 100644
--- a/packages/host-bin/src/load-external.test.ts
+++ b/packages/host-bin/src/load-external.test.ts
@@ -3,45 +3,45 @@ import { describe, expect, it } from "vitest";
import { loadExternalExtensions } from "./load-external.js";
function makeLogger(): Logger {
- const noop = () => {};
- const logger = {
- debug: noop,
- info: noop,
- warn: noop,
- error: noop,
- child: () => logger,
- span: () => {
- throw new Error("not used");
- },
- } as unknown as Logger;
- return logger;
+ const noop = () => {};
+ const logger = {
+ debug: noop,
+ info: noop,
+ warn: noop,
+ error: noop,
+ child: () => logger,
+ span: () => {
+ throw new Error("not used");
+ },
+ } as unknown as Logger;
+ return logger;
}
describe("loadExternalExtensions", () => {
- it("returns an empty array for no specifiers", async () => {
- expect(await loadExternalExtensions([], makeLogger())).toEqual([]);
- });
+ it("returns an empty array for no specifiers", async () => {
+ expect(await loadExternalExtensions([], makeLogger())).toEqual([]);
+ });
- it("loads a real extension module by package name", async () => {
- // auth-apikey is a bundled extension that exports `extension`; we use it as
- // a stand-in for an external module to exercise the dynamic-import path.
- const loaded = await loadExternalExtensions(["@dispatch/auth-apikey"], makeLogger());
- expect(loaded).toHaveLength(1);
- expect(loaded[0]?.manifest.id).toBe("auth-apikey");
- });
+ it("loads a real extension module by package name", async () => {
+ // auth-apikey is a bundled extension that exports `extension`; we use it as
+ // a stand-in for an external module to exercise the dynamic-import path.
+ const loaded = await loadExternalExtensions(["@dispatch/auth-apikey"], makeLogger());
+ expect(loaded).toHaveLength(1);
+ expect(loaded[0]?.manifest.id).toBe("auth-apikey");
+ });
- it("skips a specifier that cannot be imported without throwing", async () => {
- const loaded = await loadExternalExtensions(
- ["./does-not-exist-xyz.js", "@dispatch/auth-apikey"],
- makeLogger(),
- );
- // The bad one is skipped; the good one still loads.
- expect(loaded.map((e) => e.manifest.id)).toEqual(["auth-apikey"]);
- });
+ it("skips a specifier that cannot be imported without throwing", async () => {
+ const loaded = await loadExternalExtensions(
+ ["./does-not-exist-xyz.js", "@dispatch/auth-apikey"],
+ makeLogger(),
+ );
+ // The bad one is skipped; the good one still loads.
+ expect(loaded.map((e) => e.manifest.id)).toEqual(["auth-apikey"]);
+ });
- it("skips a module that exports no valid extension", async () => {
- // `@dispatch/journal-sink` exports factories but no `extension`.
- const loaded = await loadExternalExtensions(["@dispatch/journal-sink"], makeLogger());
- expect(loaded).toEqual([]);
- });
+ it("skips a module that exports no valid extension", async () => {
+ // `@dispatch/journal-sink` exports factories but no `extension`.
+ const loaded = await loadExternalExtensions(["@dispatch/journal-sink"], makeLogger());
+ expect(loaded).toEqual([]);
+ });
});
diff --git a/packages/host-bin/src/load-external.ts b/packages/host-bin/src/load-external.ts
index 34b8bce..522cff8 100644
--- a/packages/host-bin/src/load-external.ts
+++ b/packages/host-bin/src/load-external.ts
@@ -16,32 +16,32 @@ import type { Extension, Logger } from "@dispatch/kernel";
* boot (defend faults, not adversaries; never leave the system broken).
*/
export async function loadExternalExtensions(
- specifiers: readonly string[],
- logger: Logger,
+ specifiers: readonly string[],
+ logger: Logger,
): Promise<Extension[]> {
- const loaded: Extension[] = [];
- for (const spec of specifiers) {
- try {
- const mod = (await import(spec)) as Record<string, unknown>;
- const candidate = mod.extension ?? mod.default ?? mod;
- if (isExtension(candidate)) {
- loaded.push(candidate);
- logger.info(`Loaded external extension "${candidate.manifest.id}" from ${spec}`);
- } else {
- logger.warn(`External module "${spec}" has no valid extension export; skipped`);
- }
- } catch (err) {
- logger.error(`Failed to load external extension "${spec}"; skipped`, { err });
- }
- }
- return loaded;
+ const loaded: Extension[] = [];
+ for (const spec of specifiers) {
+ try {
+ const mod = (await import(spec)) as Record<string, unknown>;
+ const candidate = mod.extension ?? mod.default ?? mod;
+ if (isExtension(candidate)) {
+ loaded.push(candidate);
+ logger.info(`Loaded external extension "${candidate.manifest.id}" from ${spec}`);
+ } else {
+ logger.warn(`External module "${spec}" has no valid extension export; skipped`);
+ }
+ } catch (err) {
+ logger.error(`Failed to load external extension "${spec}"; skipped`, { err });
+ }
+ }
+ return loaded;
}
/** Structural check that a dynamically-imported value is an `Extension`. */
function isExtension(value: unknown): value is Extension {
- if (!value || typeof value !== "object") return false;
- const e = value as { manifest?: unknown; activate?: unknown };
- if (typeof e.activate !== "function") return false;
- const m = e.manifest as { id?: unknown } | undefined;
- return !!m && typeof m.id === "string";
+ if (!value || typeof value !== "object") return false;
+ const e = value as { manifest?: unknown; activate?: unknown };
+ if (typeof e.activate !== "function") return false;
+ const m = e.manifest as { id?: unknown } | undefined;
+ return !!m && typeof m.id === "string";
}
diff --git a/packages/host-bin/src/main.ts b/packages/host-bin/src/main.ts
index 571628f..70d1cb2 100644
--- a/packages/host-bin/src/main.ts
+++ b/packages/host-bin/src/main.ts
@@ -5,27 +5,36 @@ import { extension as cacheWarmingExt } from "@dispatch/cache-warming";
import { extension as conversationStoreExt } from "@dispatch/conversation-store";
import { createCredentialStoreExtension } from "@dispatch/credential-store";
import { createExecBackendExtension } from "@dispatch/exec-backend";
+import { extension as heartbeatExt } from "@dispatch/heartbeat";
import { createJournalSink } from "@dispatch/journal-sink";
import {
- type ConfigAccess,
- createBus,
- createHost,
- createLogger,
- type EventsEmitter,
- type Extension,
- type HostDeps,
- type LogDeps,
- type PermissionGate,
- type ScheduledJob,
- type SecretsAccess,
- type StorageNamespace,
+ type ConfigAccess,
+ createBus,
+ createHost,
+ createLogger,
+ type EventsEmitter,
+ type Extension,
+ type HostDeps,
+ type LogDeps,
+ type PermissionGate,
+ type ScheduledJob,
+ 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";
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,
+ memorySampleAttributes,
+ 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";
@@ -42,194 +51,275 @@ import { extension as toolWriteFileExt } from "@dispatch/tool-write-file";
import { extension as toolYoutubeTranscriptExt } from "@dispatch/tool-youtube-transcript";
import { createTransportHttpExtension } from "@dispatch/transport-http";
import { createTransportWsExtension } from "@dispatch/transport-ws";
+import { extension as visionHandoffExt } from "@dispatch/vision-handoff";
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 {
- get: async () => null,
- set: async () => {},
- delete: async () => {},
- };
+ return {
+ get: async () => null,
+ set: async () => {},
+ delete: async () => {},
+ };
}
function createAllowAllPermissions(): PermissionGate {
- return {
- check: async () => ({ allowed: true }),
- };
+ return {
+ check: async () => ({ allowed: true }),
+ };
}
function createNoopScheduler(): { readonly register: (job: ScheduledJob) => void } {
- return { register: () => {} };
+ return { register: () => {} };
}
function createNoopEvents(): EventsEmitter {
- return { emit: () => {} };
+ return { emit: () => {} };
}
// Core extensions EXCEPT the credential-store, which is assembled in boot() so
// its credential list can include any credentials backed by external providers
// (e.g. a `claude` credential once the external Anthropic provider is loaded).
const CORE_EXTENSIONS: readonly Extension[] = [
- storageSqliteExt,
- conversationStoreExt,
- authApikeyExt,
- providerOpenaiCompatExt,
- providerUmansExt,
- // exec-backend must precede the tool extensions that
- // `dependsOn: ["exec-backend"]` (tool-edit-file/read/shell/write). It
- // provides the ExecBackendResolver the tools resolve through; placing it
- // here keeps the activation DAG honest (it depends only on kernel).
- createExecBackendExtension(),
- toolEditFileExt,
- toolReadFileExt,
- toolShellExt,
- toolWriteFileExt,
- toolWebSearchExt,
- toolYoutubeTranscriptExt,
- throughputStoreExt,
- todoExt,
- messageQueueExt,
- mcpExt,
- sessionOrchestratorExt,
- skillsExt,
- systemPromptExt,
- cacheWarmingExt,
- lspExt,
- // 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
- // exec-backend and the tool extensions (alongside the other standard
- // tool-serving extensions) to keep the DAG honest — and before
- // transport-http, whose routes consume the ComputerService it provides.
- sshExt,
- createTransportHttpExtension(),
- // Surface extensions — dependency order: surface-registry first, then consumers.
- createSurfaceRegistryExtension(),
- createTransportWsExtension(),
- createLoadedExtensionsExtension(),
+ storageSqliteExt,
+ conversationStoreExt,
+ authApikeyExt,
+ providerOpenaiCompatExt,
+ providerUmansExt,
+ providerConcurrencyExt,
+ // exec-backend must precede the tool extensions that
+ // `dependsOn: ["exec-backend"]` (tool-edit-file/read/shell/write). It
+ // provides the ExecBackendResolver the tools resolve through; placing it
+ // here keeps the activation DAG honest (it depends only on kernel).
+ createExecBackendExtension(),
+ toolEditFileExt,
+ toolReadFileExt,
+ toolShellExt,
+ toolWriteFileExt,
+ toolWebSearchExt,
+ toolYoutubeTranscriptExt,
+ throughputStoreExt,
+ todoExt,
+ messageQueueExt,
+ mcpExt,
+ sessionOrchestratorExt,
+ skillsExt,
+ systemPromptExt,
+ cacheWarmingExt,
+ // 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
+ // exec-backend and the tool extensions (alongside the other standard
+ // tool-serving extensions) to keep the DAG honest — and before
+ // transport-http, whose routes consume the ComputerService it provides.
+ sshExt,
+ // heartbeat PROVIDES the HeartbeatService (per-workspace AI loop) the
+ // HTTP routes delegate to. Placed before transport-http (which depends on
+ // it) — mirrors how ssh precedes transport-http for the same reason.
+ heartbeatExt,
+ createTransportHttpExtension(),
+ // Surface extensions — dependency order: surface-registry first, then consumers.
+ createSurfaceRegistryExtension(),
+ createTransportWsExtension(),
+ createLoadedExtensionsExtension(),
];
/** Parse the comma-separated list of external extension module specifiers. */
function parseExternalSpecifiers(env: Readonly<Record<string, string | undefined>>): string[] {
- return (env.DISPATCH_EXTERNAL_EXTENSIONS ?? "")
- .split(",")
- .map((s) => s.trim())
- .filter((s) => s.length > 0);
+ return (env.DISPATCH_EXTERNAL_EXTENSIONS ?? "")
+ .split(",")
+ .map((s) => s.trim())
+ .filter((s) => s.length > 0);
}
async function boot(): Promise<void> {
- const journalPath = process.env.DISPATCH_JOURNAL ?? "./.dispatch/journal/app.ndjson";
- mkdirSync(dirname(journalPath), { recursive: true });
- const logSink = createJournalSink({ path: journalPath });
- const logDeps: LogDeps = { now: () => Date.now(), newId: () => crypto.randomUUID() };
- const logger = createLogger({ extensionId: "host-bin" }, logSink, logDeps);
-
- const traceDbPath = process.env.DISPATCH_TRACE_DB ?? "./.dispatch-data/traces.db";
-
- // Only start the collector supervisor in dev mode (source files available).
- // Compiled binaries don't have the source tree, so the collector can't spawn.
- let supervisor: ReturnType<typeof createCollectorSupervisor> | undefined;
- if (existsSync("packages/observability-collector/src/main.ts")) {
- supervisor = createCollectorSupervisor({
- spawn: (cmd: string[]) => {
- const proc = Bun.spawn(cmd, { stdout: "inherit", stderr: "inherit" });
- const handle: ChildHandle = {
- kill: (signal?: string) => proc.kill(signal as NodeJS.Signals),
- exited: proc.exited,
- };
- return handle;
- },
- journalPath,
- dbPath: traceDbPath,
- logger: logger.child({ extensionId: "collector-supervisor" }),
- });
- supervisor.start();
- }
-
- const dbPath = process.env.DISPATCH_DB ?? "./.dispatch-data/dispatch.db";
- mkdirSync(dirname(dbPath), { recursive: true });
- const sqliteBackend = createSqliteStorage({ path: dbPath });
- const storageFactory = (namespace: string): StorageNamespace => sqliteBackend.storage(namespace);
-
- const configMap = envToConfigMap(process.env as Readonly<Record<string, string | undefined>>);
- const config: ConfigAccess = configMapToAccess(configMap);
-
- const deps: HostDeps = {
- logger,
- config,
- storageFactory,
- secrets: createEmptySecrets(),
- permissions: createAllowAllPermissions(),
- scheduler: createNoopScheduler(),
- bus: createBus(logger),
- events: createNoopEvents(),
- logSink,
- logDeps,
- };
-
- // Load external (out-of-repo) extensions declared via DISPATCH_EXTERNAL_EXTENSIONS.
- const externalSpecifiers = parseExternalSpecifiers(
- process.env as Readonly<Record<string, string | undefined>>,
- );
- const externalExtensions = await loadExternalExtensions(externalSpecifiers, logger);
-
- // Assemble the credential list. MVP keeps the hardcoded `opencode` credential
- // and adds a `claude` credential when an external Anthropic provider is loaded.
- const credentials = [{ name: "opencode", providerId: "openai-compat" }];
-
- // The umans credential is always listed (it's the model-catalog index); the
- // provider itself only registers when UMANS_API_KEY is set, so listCatalog
- // gracefully skips it when the provider is absent.
- if (process.env.UMANS_API_KEY) {
- credentials.push({ name: "umans", providerId: "umans" });
- logger.info(`Registered credential "umans" → umans provider`);
- }
- const hasAnthropic = externalExtensions.some((e) =>
- e.manifest.contributes?.providers?.includes("anthropic"),
- );
- if (hasAnthropic) {
- const claudeName = process.env.DISPATCH_CLAUDE_CREDENTIAL ?? "claude";
- credentials.push({ name: claudeName, providerId: "anthropic" });
- logger.info(`Registered credential "${claudeName}" → anthropic provider`);
- }
-
- const extensions: Extension[] = [
- ...CORE_EXTENSIONS,
- createCredentialStoreExtension({ credentials }),
- ...externalExtensions,
- ];
-
- const host = createHost(extensions, deps);
- await host.activate();
-
- const disabled = host.getDisabled();
- if (disabled.length > 0) {
- for (const d of disabled) {
- logger.warn(`Extension "${d.manifest.id}" disabled: ${d.reason}`);
- }
- }
-
- let shuttingDown = false;
- const shutdown = async () => {
- if (shuttingDown) return;
- shuttingDown = true;
- logger.info("Shutting down — deactivating extensions");
- await host.deactivate();
- logger.info("Draining collector");
- await supervisor?.stop();
- process.exit(0);
- };
- process.on("SIGINT", shutdown);
- process.on("SIGTERM", shutdown);
-
- logger.info("Dispatch booted");
- console.info("Dispatch booted");
+ const journalPath = process.env.DISPATCH_JOURNAL ?? "./.dispatch/journal/app.ndjson";
+ mkdirSync(dirname(journalPath), { recursive: true });
+ const logSink = createJournalSink({ path: journalPath });
+ const logDeps: LogDeps = { now: () => Date.now(), newId: () => crypto.randomUUID() };
+ const logger = createLogger({ extensionId: "host-bin" }, logSink, logDeps);
+
+ const traceDbPath = process.env.DISPATCH_TRACE_DB ?? "./.dispatch-data/traces.db";
+
+ // Only start the collector supervisor in dev mode (source files available).
+ // Compiled binaries don't have the source tree, so the collector can't spawn.
+ let supervisor: ReturnType<typeof createCollectorSupervisor> | undefined;
+ if (existsSync("packages/observability-collector/src/main.ts")) {
+ supervisor = createCollectorSupervisor({
+ spawn: (cmd: string[]) => {
+ const proc = Bun.spawn(cmd, { stdout: "inherit", stderr: "inherit" });
+ const handle: ChildHandle = {
+ kill: (signal?: string) => proc.kill(signal as NodeJS.Signals),
+ exited: proc.exited,
+ };
+ return handle;
+ },
+ journalPath,
+ dbPath: traceDbPath,
+ logger: logger.child({ extensionId: "collector-supervisor" }),
+ });
+ supervisor.start();
+ }
+
+ const dbPath = process.env.DISPATCH_DB ?? "./.dispatch-data/dispatch.db";
+ mkdirSync(dirname(dbPath), { recursive: true });
+ const sqliteBackend = createSqliteStorage({ path: dbPath });
+ const storageFactory = (namespace: string): StorageNamespace => sqliteBackend.storage(namespace);
+
+ const configMap = envToConfigMap(process.env as Readonly<Record<string, string | undefined>>);
+ const config: ConfigAccess = configMapToAccess(configMap);
+
+ const deps: HostDeps = {
+ logger,
+ config,
+ storageFactory,
+ secrets: createEmptySecrets(),
+ permissions: createAllowAllPermissions(),
+ scheduler: createNoopScheduler(),
+ bus: createBus(logger),
+ events: createNoopEvents(),
+ logSink,
+ logDeps,
+ };
+
+ // Load external (out-of-repo) extensions declared via DISPATCH_EXTERNAL_EXTENSIONS.
+ const externalSpecifiers = parseExternalSpecifiers(
+ process.env as Readonly<Record<string, string | undefined>>,
+ );
+ const externalExtensions = await loadExternalExtensions(externalSpecifiers, logger);
+
+ // Assemble the credential list. MVP keeps the hardcoded `opencode` credential
+ // and adds a `claude` credential when an external Anthropic provider is loaded.
+ const credentials = [{ name: "opencode", providerId: "openai-compat" }];
+
+ // The umans credential is always listed (it's the model-catalog index); the
+ // provider itself only registers when UMANS_API_KEY is set, so listCatalog
+ // gracefully skips it when the provider is absent.
+ if (process.env.UMANS_API_KEY) {
+ credentials.push({ name: "umans", providerId: "umans" });
+ logger.info(`Registered credential "umans" → umans provider`);
+ }
+ const hasAnthropic = externalExtensions.some((e) =>
+ e.manifest.contributes?.providers?.includes("anthropic"),
+ );
+ if (hasAnthropic) {
+ const claudeName = process.env.DISPATCH_CLAUDE_CREDENTIAL ?? "claude";
+ credentials.push({ name: claudeName, providerId: "anthropic" });
+ logger.info(`Registered credential "${claudeName}" → anthropic provider`);
+ }
+
+ const extensions: Extension[] = [
+ ...CORE_EXTENSIONS,
+ createCredentialStoreExtension({ credentials }),
+ // vision-handoff activates AFTER credential-store (it resolves the
+ // credential-store service at activate time to find vision-capable models).
+ // Placed here, not in CORE_EXTENSIONS, so the service is available when it
+ // activates. The session-orchestrator resolves its service LAZILY
+ // (per-turn), so activation order between it and session-orchestrator
+ // doesn't matter.
+ visionHandoffExt,
+ ...externalExtensions,
+ ];
+
+ const host = createHost(extensions, deps);
+ await host.activate();
+
+ const disabled = host.getDisabled();
+ if (disabled.length > 0) {
+ for (const d of disabled) {
+ logger.warn(`Extension "${d.manifest.id}" disabled: ${d.reason}`);
+ }
+ }
+
+ // 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;
+ let activeConvCountFn: (() => number) | undefined;
+ try {
+ const orchestrator = host.getHostAPI().getService(sessionOrchestratorHandle);
+ activeConvCountFn = () => orchestrator.getActiveConversationCount();
+ 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");
+ await supervisor?.stop();
+ process.exit(0);
+ };
+ 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");
}
boot().catch((err) => {
- console.error("Fatal boot error:", err);
- process.exit(1);
+ console.error("Fatal boot error:", err);
+ process.exit(1);
});
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..576347b
--- /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 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;
+
+/**
+ * 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 15s. */
+ 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 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.
+ *
+ * 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 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();
+ 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"),
+ };
+}
diff --git a/packages/host-bin/tsconfig.json b/packages/host-bin/tsconfig.json
index e445f13..09b87df 100644
--- a/packages/host-bin/tsconfig.json
+++ b/packages/host-bin/tsconfig.json
@@ -1,65 +1,71 @@
{
- "extends": "../../tsconfig.base.json",
- "compilerOptions": {
- "rootDir": "src",
- "outDir": "dist",
- "composite": true
- },
- "include": ["src/**/*.ts"],
- "references": [
- {
- "path": "../cache-warming"
- },
- {
- "path": "../exec-backend"
- },
- {
- "path": "../kernel"
- },
- {
- "path": "../lsp"
- },
- {
- "path": "../message-queue"
- },
- {
- "path": "../skills"
- },
- {
- "path": "../ssh"
- },
- {
- "path": "../storage-sqlite"
- },
- {
- "path": "../surface-loaded-extensions"
- },
- {
- "path": "../surface-registry"
- },
- {
- "path": "../system-prompt"
- },
- {
- "path": "../throughput-store"
- },
- {
- "path": "../tool-edit-file"
- },
- {
- "path": "../tool-read-file"
- },
- {
- "path": "../tool-shell"
- },
- {
- "path": "../tool-write-file"
- },
- {
- "path": "../transport-http"
- },
- {
- "path": "../transport-ws"
- }
- ]
+ "extends": "../../tsconfig.base.json",
+ "compilerOptions": {
+ "rootDir": "src",
+ "outDir": "dist",
+ "composite": true
+ },
+ "include": ["src/**/*.ts"],
+ "references": [
+ {
+ "path": "../cache-warming"
+ },
+ {
+ "path": "../exec-backend"
+ },
+ {
+ "path": "../kernel"
+ },
+ {
+ "path": "../lsp"
+ },
+ {
+ "path": "../message-queue"
+ },
+ {
+ "path": "../provider-concurrency"
+ },
+ {
+ "path": "../skills"
+ },
+ {
+ "path": "../ssh"
+ },
+ {
+ "path": "../storage-sqlite"
+ },
+ {
+ "path": "../surface-loaded-extensions"
+ },
+ {
+ "path": "../surface-registry"
+ },
+ {
+ "path": "../system-prompt"
+ },
+ {
+ "path": "../throughput-store"
+ },
+ {
+ "path": "../tool-edit-file"
+ },
+ {
+ "path": "../tool-read-file"
+ },
+ {
+ "path": "../tool-shell"
+ },
+ {
+ "path": "../tool-write-file"
+ },
+ {
+ "path": "../transport-http"
+ },
+ {
+ "path": "../transport-ws"
+ },
+ {
+ "path": "../vision-handoff"
+ }
+ ]
}