lvmh/plugin/smoke.ts

/**
 * Dev-only smoke test for lvmh-agent.ts. Not part of the shipped extension.
 *
 * Run:  node plugin/smoke.ts          (Node >= 23.6; older: --experimental-strip-types)
 *
 * Scenarios (full matrix in selftest.md):
 *   1. env unset          -> factory inert, nothing registered
 *   2. live daemon        -> bearer header, hello snapshot, mirrored events,
 *                            delta-only updates, message mapping, 2000-char
 *                            truncation, usage mapping, monotonic seq
 *   3. prompt / abort     -> pi.sendUserMessage(steer) / pi.abort; foreign
 *                            session prompts ignored
 *   4. drop + reconnect   -> replay of persisted events after welcome.lastSeq,
 *                            seq continuity, buffer_overflow notice after
 *                            queue/replay overflow
 *   5. dead host          -> loads, retries, never throws, event loop responsive
 *   6. session_shutdown   -> socket closed, no reconnect afterwards
 */

import {
	check as rawCheck,
	sleep,
	waitFor,
	startMiniDaemon,
} from "./mini-daemon.ts";

const SESSION_ID = "sess-1";

let failures = 0;

function check(name: string, ok: boolean, detail = ""): void {
	rawCheck(name, ok, detail);
	if (!ok) failures++;
}

interface FakePi {
	handlers: Map<string, (event: unknown, ctx: unknown) => unknown>;
	sentMessages: Array<{ message: string; options: unknown }>;
	aborted: number;
	on(event: string, handler: (event: unknown, ctx: unknown) => unknown): void;
	sendUserMessage(message: string, options?: unknown): void;
	abort(): void;
}

function makeFakePi(): FakePi {
	return {
		handlers: new Map(),
		sentMessages: [],
		aborted: 0,
		on(event, handler) {
			this.handlers.set(event, handler);
		},
		sendUserMessage(message, options) {
			this.sentMessages.push({ message, options });
		},
		abort() {
			this.aborted++;
		},
	};
}

function makeFakeCtx(): unknown {
	return {
		cwd: "/work/repo",
		model: { id: "glm-5.3", provider: "zai-renaud" },
		sessionManager: {
			getSessionId: () => SESSION_ID,
			getSessionName: () => undefined,
			getCwd: () => "/work/repo",
			getHeader: () => ({
				timestamp: "2024-12-03T14:00:00.000Z",
				id: SESSION_ID,
			}),
		},
	};
}

async function loadExtension(): Promise<(pi: unknown) => void> {
	const mod = (await import("./lvmh-agent.ts")) as {
		default: (pi: unknown) => void;
	};
	return mod.default;
}

async function main(): Promise<void> {
	// --- Scenario 1: env unset -> inert --------------------------------
	delete process.env.LVMH_URL;
	delete process.env.LVMH_TOKEN;
	delete process.env.LVMH_AGENT;
	delete process.env.LVMH_REPO;
	const factory = await loadExtension();
	const inertPi = makeFakePi();
	inertPi.on = ((event: string) => {
		throw new Error(`inert extension registered handler ${event}`);
	}) as FakePi["on"];
	factory(inertPi); // must not touch pi at all
	check("1 inert: no side effects on pi", true);

	// --- Scenario 2: live daemon ---------------------------------------
	let welcomeLastSeq = 0;
	const daemon = await startMiniDaemon(() => welcomeLastSeq);
	process.env.LVMH_URL = daemon.url;
	process.env.LVMH_TOKEN = "smoke-token";

	const pi = makeFakePi();
	factory(pi);
	const handlerNames = [
		"session_start",
		"session_shutdown",
		"session_info_changed",
		"model_select",
		"message_start",
		"message_update",
		"message_end",
		"tool_execution_start",
		"tool_execution_update",
		"tool_execution_end",
		"agent_start",
		"agent_end",
		"agent_settled",
	];
	check(
		"2 all 13 handlers registered",
		handlerNames.every((n) => pi.handlers.has(n)),
	);

	let lastReturn: unknown = "sentinel";
	const fire = (name: string, event: unknown): void => {
		const h = pi.handlers.get(name);
		if (h === undefined) throw new Error(`missing handler ${name}`);
		lastReturn = h(event, makeFakeCtx());
		if (lastReturn instanceof Promise) lastReturn.catch(() => undefined);
	};

	fire("session_start", { reason: "startup" });
	const helloSeen = await waitFor(
		() => daemon.frames.some((f) => f.type === "hello"),
		5000,
	);
	check("2 connects and sends hello", helloSeen);
	check(
		"2 bearer auth on upgrade",
		daemon.authHeaders.at(-1) === "Bearer smoke-token",
		daemon.authHeaders.at(-1),
	);

	const hello = daemon.frames.find((f) => f.type === "hello");
	const hs = (hello?.session ?? {}) as Record<string, unknown>;
	check(
		"2 hello snapshot",
		hs.id === SESSION_ID &&
			hs.name === null &&
			hs.cwd === "/work/repo" &&
			hs.model === "glm-5.3" &&
			hs.provider === "zai-renaud" &&
			hs.agent === false &&
			hs.repo === null &&
			typeof hs.startedAt === "number",
		JSON.stringify(hs),
	);

	fire("message_start", {
		message: { role: "assistant", content: [], timestamp: 1000 },
	});
	fire("message_update", {
		message: {
			role: "assistant",
			content: [{ type: "text", text: "Hel" }],
			timestamp: 1000,
		},
	});
	fire("message_update", {
		message: {
			role: "assistant",
			content: [{ type: "text", text: "Hello world" }],
			timestamp: 1000,
		},
	});
	fire("message_end", {
		message: {
			role: "assistant",
			content: [
				{ type: "thinking", thinking: "hmm" },
				{ type: "text", text: "Hello world" },
				{
					type: "toolCall",
					id: "tc1",
					name: "bash",
					arguments: { command: "ls" },
				},
			],
			timestamp: 1000,
		},
	});
	await waitFor(
		() => daemon.frames.some((f) => f.type === "message_end"),
		2000,
	);

	const n0 = daemon.frames.findIndex((f) => f.type === "hello");
	const mirrored = daemon.frames.slice(n0 + 1).map((f) => f.type);
	check(
		"2 mirror order",
		JSON.stringify(mirrored) ===
			JSON.stringify([
				"message_start",
				"message_update",
				"message_update",
				"message_end",
			]),
		JSON.stringify(mirrored),
	);
	const deltas = daemon.frames
		.filter((f) => f.type === "message_update")
		.map((f) => f.delta);
	check(
		"2 delta-only updates",
		JSON.stringify(deltas) === JSON.stringify(["Hel", "lo world"]),
		JSON.stringify(deltas),
	);

	const msg = daemon.frames.find((f) => f.type === "message_end")
		?.message as Record<string, unknown>;
	const tc = (
		msg?.toolCalls as Array<Record<string, unknown>> | undefined
	)?.[0];
	check(
		"2 message_end mapping",
		msg?.role === "assistant" &&
			msg?.text === "Hello world" &&
			msg?.thinking === "hmm" &&
			typeof msg?.id === "string" &&
			tc?.id === "tc1" &&
			tc?.name === "bash" &&
			tc?.argsJson === '{"command":"ls"}' &&
			msg?.toolCallId === null,
		JSON.stringify(msg),
	);

	const big = "x".repeat(3000);
	fire("agent_start", {});
	fire("tool_execution_start", {
		toolCallId: "tc1",
		toolName: "bash",
		args: { command: "ls" },
	});
	fire("tool_execution_update", {
		toolCallId: "tc1",
		toolName: "bash",
		partialResult: { content: [{ type: "text", text: big }] },
	});
	fire("tool_execution_end", {
		toolCallId: "tc1",
		toolName: "bash",
		isError: true,
		result: { content: [{ type: "text", text: big }] },
	});
	fire("agent_end", {
		messages: [
			{
				role: "assistant",
				usage: { input: 10, output: 5, cost: { total: 0.25 } },
			},
		],
	});
	fire("agent_settled", {});
	await waitFor(
		() => daemon.frames.some((f) => f.type === "agent_settled"),
		2000,
	);

	const partial = daemon.frames.find((f) => f.type === "tool_execution_update");
	const end = daemon.frames.find((f) => f.type === "tool_execution_end");
	const partialLen = (partial?.partial as string | undefined)?.length;
	const previewLen = (end?.resultPreview as string | undefined)?.length;
	check("2 partial truncated to 2000", partialLen === 2000, String(partialLen));
	check(
		"2 resultPreview truncated + isError",
		previewLen === 2000 && end?.isError === true,
	);
	const usage = daemon.frames.find((f) => f.type === "agent_end")
		?.usage as Record<string, number>;
	check(
		"2 agent_end usage",
		usage?.inputTokens === 10 &&
			usage?.outputTokens === 5 &&
			usage?.totalCost === 0.25,
		JSON.stringify(usage),
	);

	const seqs = daemon.frames.filter((f) => f.seq > 0).map((f) => f.seq);
	check(
		"2 seq strictly monotonic",
		seqs.length > 0 && seqs.every((s, i) => i === 0 || s > seqs[i - 1]),
		JSON.stringify(seqs),
	);
	check(
		"2 handler return values are undefined (never a promise)",
		lastReturn === undefined,
	);

	// --- Scenario 3: prompt / abort ------------------------------------
	daemon.pushAll(
		JSON.stringify({
			v: 1,
			type: "prompt",
			sessionId: "other-session",
			seq: 0,
			ts: Date.now(),
			promptId: "p0",
			message: "wrong session",
		}),
	);
	daemon.pushAll(
		JSON.stringify({
			v: 1,
			type: "prompt",
			sessionId: SESSION_ID,
			seq: 0,
			ts: Date.now(),
			promptId: "p1",
			message: "run the tests",
		}),
	);
	await waitFor(() => pi.sentMessages.length > 0, 2000);
	check(
		"3 prompt delivered as steer, foreign session ignored",
		pi.sentMessages.length === 1 &&
			pi.sentMessages[0].message === "run the tests" &&
			JSON.stringify(pi.sentMessages[0].options) === '{"deliverAs":"steer"}',
		JSON.stringify(pi.sentMessages),
	);
	daemon.pushAll(
		JSON.stringify({
			v: 1,
			type: "abort",
			sessionId: SESSION_ID,
			seq: 0,
			ts: Date.now(),
		}),
	);
	await waitFor(() => pi.aborted > 0, 2000);
	check("3 abort calls pi.abort", pi.aborted === 1);

	// --- Scenario 4: drop + reconnect, replay, overflow ----------------
	const seqBeforeDrop = Math.max(...daemon.frames.map((f) => f.seq));
	welcomeLastSeq = seqBeforeDrop; // daemon has everything up to the drop
	daemon.dropConnections();
	await sleep(150); // let the client notice
	fire("message_end", {
		message: { role: "user", content: "offline-1", timestamp: 2000 },
	});
	fire("message_end", {
		message: { role: "user", content: "offline-2", timestamp: 2001 },
	});
	const reconnected = await waitFor(
		() => daemon.frames.filter((f) => f.type === "hello").length >= 2,
		10000,
	);
	check("4 reconnects after drop", reconnected);
	const replayedTexts = daemon.frames
		.filter((f) => f.type === "message_end" && f.seq > seqBeforeDrop)
		.map((f) => (f.message as { text?: string }).text);
	check(
		"4 buffered events replayed after welcome.lastSeq",
		replayedTexts.includes("offline-1") && replayedTexts.includes("offline-2"),
		JSON.stringify(replayedTexts),
	);
	const allSeqs = daemon.frames.filter((f) => f.seq > 0).map((f) => f.seq);
	const dupCount = allSeqs.length - new Set(allSeqs).size;
	check(
		"4 no duplicate seq delivery",
		dupCount === 0,
		`duplicates: ${dupCount}`,
	);
	const seqsAfter = daemon.frames
		.filter((f) => f.seq > seqBeforeDrop)
		.map((f) => f.seq);
	check(
		"4 seq continues monotonically across reconnect",
		seqsAfter.length >= 2 &&
			seqsAfter.every((s, i) => i === 0 || s > seqsAfter[i - 1]),
		JSON.stringify(seqsAfter),
	);

	// overflow both bounded buffers, then reconnect once more
	const seqBeforeOverflow = Math.max(...daemon.frames.map((f) => f.seq));
	welcomeLastSeq = seqBeforeOverflow;
	daemon.dropConnections();
	await sleep(150);
	for (let i = 0; i < 10050; i++) {
		fire("agent_settled", {}); // persisted kind: fills replay buffer + send queue
	}
	const reconnected2 = await waitFor(
		() => daemon.frames.filter((f) => f.type === "hello").length >= 3,
		15000,
	);
	check("4 reconnects after overflow window", reconnected2);
	const overflowNotice = await waitFor(
		() => daemon.frames.some((f) => f.type === "buffer_overflow"),
		5000,
	);
	const overflow = daemon.frames.find((f) => f.type === "buffer_overflow");
	check(
		"4 buffer_overflow notice after drops",
		overflowNotice &&
			typeof (overflow?.dropped as number) === "number" &&
			(overflow?.dropped as number) > 0,
		JSON.stringify(overflow ?? null),
	);

	// --- Scenario 9: daemon hangs mid-handshake (no welcome) ----------
	// Fast watchdog knobs; read lazily by the plugin so late env wins.
	process.env.LVMH_WELCOME_TIMEOUT_MS = "400";
	process.env.LVMH_STALL_CHECK_MS = "120";
	process.env.LVMH_STALL_BYTES = "1000000";
	const hungDaemon = await startMiniDaemon(() => 0, { suppressWelcome: true });
	const prevUrl = process.env.LVMH_URL;
	process.env.LVMH_URL = hungDaemon.url;
	const pi3 = makeFakePi();
	factory(pi3);
	pi3.handlers.get("session_start")?.({ reason: "startup" }, makeFakeCtx());
	const hungHellos = await waitFor(
		() => hungDaemon.frames.filter((f) => f.type === "hello").length >= 2,
		12000,
	);
	check(
		"9 welcome timeout: abandons hung socket and retries",
		hungHellos,
		String(hungDaemon.frames.length),
	);
	pi3.handlers.get("session_shutdown")?.({ reason: "quit" }, makeFakeCtx());
	hungDaemon.close();
	process.env.LVMH_URL = prevUrl;

	// --- Scenario 10: daemon dies without TCP close (wedged socket) ----
	// wedge() pauses daemon-side reads: TCP stays "open", undici keeps
	// buffering sends. The plugin must detect bufferedAmount not draining
	// and reconnect, or mirroring would be dead forever + memory unbounded.
	// Fresh plugin instance: its stall interval inherits the fast knobs set
	// in scenario 9 (interval period is fixed at first connect).
	const stallDaemon = await startMiniDaemon(() => 0);
	const urlBeforeStall: string | undefined = process.env.LVMH_URL;
	process.env.LVMH_URL = stallDaemon.url;
	const pi4 = makeFakePi();
	factory(pi4);
	const fire4 = (name: string, event: unknown): void => {
		const h = pi4.handlers.get(name);
		if (h === undefined) throw new Error(`missing handler ${name}`);
		const r = h(event, makeFakeCtx());
		if (r instanceof Promise) r.catch(() => undefined);
	};
	fire4("session_start", { reason: "startup" });
	await waitFor(() => stallDaemon.frames.some((f) => f.type === "hello"), 5000);
	stallDaemon.wedge();
	const bigText: string = "w".repeat(1024 * 1024);
	for (let i = 0; i < 60; i++) {
		fire4("message_end", {
			message: { role: "user", content: bigText, timestamp: 3000 + i },
		});
	}
	const wedgedRecovered = await waitFor(
		() => stallDaemon.frames.filter((f) => f.type === "hello").length >= 2,
		20000,
	);
	check(
		"10 stall watchdog: wedged socket detected, reconnected",
		wedgedRecovered,
		String(stallDaemon.frames.length),
	);
	stallDaemon.unwedge();
	fire4("message_end", {
		message: { role: "user", content: "after-stall", timestamp: 4000 },
	});
	const afterStallDelivered = await waitFor(
		() =>
			stallDaemon.frames.some(
				(f) =>
					f.type === "message_end" &&
					(f.message as { text?: string })?.text === "after-stall",
			),
		15000,
	);
	check("10 mirroring works after stall recovery", afterStallDelivered);
	fire4("session_shutdown", { reason: "quit" });
	stallDaemon.close();
	process.env.LVMH_URL = urlBeforeStall;
	delete process.env.LVMH_WELCOME_TIMEOUT_MS;
	delete process.env.LVMH_STALL_CHECK_MS;
	delete process.env.LVMH_STALL_BYTES;

	// --- Scenario 6: session_shutdown ----------------------------------
	const helloCountAtShutdown = daemon.frames.filter(
		(f) => f.type === "hello",
	).length;
	fire("session_shutdown", { reason: "quit" });
	const closed = await waitFor(() => daemon.connections() === 0, 3000);
	check("6 shutdown closes connection", closed);
	await sleep(2600); // longer than one backoff attempt (~1-2s)
	const helloCountAfter = daemon.frames.filter(
		(f) => f.type === "hello",
	).length;
	check(
		"6 no reconnect after shutdown",
		helloCountAfter === helloCountAtShutdown,
	);

	daemon.close();

	// --- Scenario 5: dead host -----------------------------------------
	const deadDaemon = await startMiniDaemon(() => 0);
	const deadUrl = deadDaemon.url;
	deadDaemon.close(); // port now closed
	await sleep(100);
	process.env.LVMH_URL = deadUrl;
	const pi2 = makeFakePi();
	factory(pi2);
	pi2.handlers.get("session_start")?.({ reason: "startup" }, makeFakeCtx());
	await sleep(400);
	const t0 = Date.now();
	await sleep(100);
	const loopResponsive = Date.now() - t0 < 500;
	check("5 dead host: pi loads, event loop responsive", loopResponsive);
	let threw = false;
	try {
		pi2.handlers.get("message_end")?.(
			{ message: { role: "user", content: "hi", timestamp: 1 } },
			makeFakeCtx(),
		);
	} catch {
		threw = true;
	}
	check("5 dead host: handlers never throw", !threw);

	delete process.env.LVMH_URL;
	delete process.env.LVMH_TOKEN;
	console.log(
		failures === 0 ? "\nALL CHECKS PASSED" : `\n${failures} CHECK(S) FAILED`,
	);
	process.exit(failures === 0 ? 0 : 1);
}

main().catch((err: unknown) => {
	console.error("smoke harness crashed:", err);
	process.exit(1);
});