feat(kernel): event/hook bus (events, filters, services) — pure dispatch + stateful shell

1 files changed, 365 insertions, 0 deletions

diff --git a/packages/kernel/src/bus/bus.test.ts b/packages/kernel/src/bus/bus.test.ts
new file mode 100644
index 0000000..7366b50
--- /dev/null
+++ b/packages/kernel/src/bus/bus.test.ts
@@ -0,0 +1,365 @@
+import { beforeEach, describe, expect, it } from "vitest";
+import type { Logger } from "../contracts/extension.js";
+import { defineEventHook, defineFilter, defineService } from "../contracts/hooks.js";
+import { type Bus, createBus } from "./bus.js";
+import { applyFilterChain, dispatchEventSync, sortFilters } from "./pure.js";
+
+interface FakeLogger extends Logger {
+	readonly errors: Array<{ message: string; args: unknown[] }>;
+}
+
+function createFakeLogger(): FakeLogger {
+	const errors: Array<{ message: string; args: unknown[] }> = [];
+	return {
+		errors,
+		debug: () => {},
+		info: () => {},
+		warn: () => {},
+		error: (message: string, ...args: unknown[]) => {
+			errors.push({ message, args });
+		},
+	};
+}
+
+describe("event hooks", () => {
+	let logger: FakeLogger;
+	let bus: Bus;
+
+	beforeEach(() => {
+		logger = createFakeLogger();
+		bus = createBus(logger);
+	});
+
+	it("fires all registered listeners", () => {
+		const hook = defineEventHook<{ value: number }>("test/event");
+		const received: number[] = [];
+
+		bus.on(hook, (payload) => {
+			received.push(payload.value);
+		});
+		bus.on(hook, (payload) => {
+			received.push(payload.value * 10);
+		});
+
+		bus.emit(hook, { value: 3 });
+
+		expect(received).toEqual([3, 30]);
+	});
+
+	it("isolates a throwing listener (others still run, error logged)", () => {
+		const hook = defineEventHook<string>("test/isolate");
+		const received: string[] = [];
+
+		bus.on(hook, () => {
+			throw new Error("boom");
+		});
+		bus.on(hook, (payload) => {
+			received.push(payload);
+		});
+
+		bus.emit(hook, "hello");
+
+		expect(received).toEqual(["hello"]);
+		expect(logger.errors).toHaveLength(1);
+		expect(logger.errors[0]?.message).toContain("test/isolate");
+	});
+
+	it("isolates an async handler rejection", async () => {
+		const hook = defineEventHook<string>("test/async-reject");
+		const received: string[] = [];
+
+		bus.on(hook, async () => {
+			throw new Error("async boom");
+		});
+		bus.on(hook, async (payload) => {
+			received.push(payload);
+		});
+
+		bus.emit(hook, "data");
+
+		await new Promise((resolve) => {
+			setTimeout(resolve, 10);
+		});
+
+		expect(received).toEqual(["data"]);
+		expect(logger.errors).toHaveLength(1);
+		expect(logger.errors[0]?.message).toContain("test/async-reject");
+	});
+
+	it("unsubscribe removes the handler", () => {
+		const hook = defineEventHook<void>("test/unsub");
+		let count = 0;
+
+		const unsub = bus.on(hook, () => {
+			count++;
+		});
+
+		bus.emit(hook, undefined);
+		expect(count).toBe(1);
+
+		unsub();
+		bus.emit(hook, undefined);
+		expect(count).toBe(1);
+	});
+
+	it("emit with no handlers is a no-op", () => {
+		const hook = defineEventHook<string>("test/empty");
+		expect(() => bus.emit(hook, "nothing")).not.toThrow();
+	});
+
+	it("emitAsync awaits all handlers", async () => {
+		const hook = defineEventHook<number>("test/async");
+		const received: number[] = [];
+
+		bus.on(hook, async (payload) => {
+			await new Promise((resolve) => {
+				setTimeout(resolve, 5);
+			});
+			received.push(payload);
+		});
+		bus.on(hook, async (payload) => {
+			received.push(payload * 2);
+		});
+
+		await bus.emitAsync(hook, 5);
+
+		expect(received).toEqual([10, 5]);
+	});
+
+	it("emitAsync respects timeout", async () => {
+		const hook = defineEventHook<void>("test/timeout");
+		let completed = false;
+
+		bus.on(hook, async () => {
+			await new Promise((resolve) => {
+				setTimeout(resolve, 100);
+			});
+			completed = true;
+		});
+
+		await bus.emitAsync(hook, undefined, 10);
+
+		expect(completed).toBe(false);
+	});
+});
+
+describe("filter hooks", () => {
+	let logger: FakeLogger;
+	let bus: Bus;
+
+	beforeEach(() => {
+		logger = createFakeLogger();
+		bus = createBus(logger);
+	});
+
+	it("chains filters in registration order", async () => {
+		const hook = defineFilter<string>("test/chain");
+
+		bus.addFilter(hook, (value) => `${value}-a`);
+		bus.addFilter(hook, (value) => `${value}-b`);
+
+		const result = await bus.applyFilters(hook, "start");
+		expect(result).toBe("start-a-b");
+	});
+
+	it("respects priority ordering (lower runs first)", async () => {
+		const hook = defineFilter<string>("test/priority");
+
+		bus.addFilter(hook, (value) => `${value}-second`);
+		bus.addFilter(hook, (value) => `${value}-first`, { priority: -1 });
+
+		const result = await bus.applyFilters(hook, "start");
+		expect(result).toBe("start-first-second");
+	});
+
+	it("fail-open passes value through on throw", async () => {
+		const hook = defineFilter<number>("test/fail-open");
+
+		bus.addFilter(hook, (value) => value + 1);
+		bus.addFilter(hook, () => {
+			throw new Error("filter boom");
+		});
+		bus.addFilter(hook, (value) => value * 2);
+
+		const result = await bus.applyFilters(hook, 5);
+		expect(result).toBe(12);
+		expect(logger.errors).toHaveLength(1);
+		expect(logger.errors[0]?.message).toContain("test/fail-open");
+	});
+
+	it("fail-closed propagates the error", async () => {
+		const hook = defineFilter<number>("test/fail-closed");
+
+		bus.addFilter(hook, () => {
+			throw new Error("closed boom");
+		});
+
+		await expect(bus.applyFilters(hook, 5, { failClosed: true })).rejects.toThrow("closed boom");
+	});
+
+	it("applyFilters with no filters returns value unchanged", async () => {
+		const hook = defineFilter<string>("test/no-filters");
+		const result = await bus.applyFilters(hook, "unchanged");
+		expect(result).toBe("unchanged");
+	});
+
+	it("unsubscribe removes a filter from the chain", async () => {
+		const hook = defineFilter<string>("test/filter-unsub");
+
+		const unsub = bus.addFilter(hook, (value) => `${value}-removed`);
+		bus.addFilter(hook, (value) => `${value}-kept`);
+
+		unsub();
+
+		const result = await bus.applyFilters(hook, "start");
+		expect(result).toBe("start-kept");
+	});
+});
+
+describe("services", () => {
+	let logger: FakeLogger;
+	let bus: Bus;
+
+	beforeEach(() => {
+		logger = createFakeLogger();
+		bus = createBus(logger);
+	});
+
+	it("provide and get round-trips", () => {
+		const handle = defineService<{ greet: (name: string) => string }>("test/service");
+		const impl = { greet: (name: string) => `hello ${name}` };
+
+		bus.provideService(handle, impl);
+		const retrieved = bus.getService(handle);
+
+		expect(retrieved.greet("world")).toBe("hello world");
+	});
+
+	it("getService on missing service throws", () => {
+		const handle = defineService<string>("test/missing");
+		expect(() => bus.getService(handle)).toThrow("test/missing");
+	});
+
+	it("double-provide throws", () => {
+		const handle = defineService<number>("test/double");
+
+		bus.provideService(handle, 1);
+		expect(() => bus.provideService(handle, 2)).toThrow("test/double");
+	});
+});
+
+describe("pure functions", () => {
+	describe("dispatchEventSync", () => {
+		it("calls all handlers with the payload", () => {
+			const logger = createFakeLogger();
+			const received: number[] = [];
+
+			dispatchEventSync(
+				[
+					(payload) => {
+						received.push(payload);
+					},
+					(payload) => {
+						received.push(payload * 2);
+					},
+				],
+				5,
+				logger,
+				"test",
+			);
+
+			expect(received).toEqual([5, 10]);
+		});
+
+		it("catches sync throws and logs them", () => {
+			const logger = createFakeLogger();
+			const received: number[] = [];
+
+			dispatchEventSync(
+				[
+					() => {
+						throw new Error("sync boom");
+					},
+					(payload) => {
+						received.push(payload);
+					},
+				],
+				42,
+				logger,
+				"test/sync",
+			);
+
+			expect(received).toEqual([42]);
+			expect(logger.errors).toHaveLength(1);
+		});
+	});
+
+	describe("sortFilters", () => {
+		it("sorts by priority ascending, then by order ascending", () => {
+			const entries = [
+				{ fn: async (v: number) => v, priority: 10, order: 0 },
+				{ fn: async (v: number) => v, priority: -1, order: 1 },
+				{ fn: async (v: number) => v, priority: 10, order: 2 },
+				{ fn: async (v: number) => v, priority: 0, order: 3 },
+			];
+
+			const sorted = sortFilters(entries);
+			expect(sorted.map((e) => e.order)).toEqual([1, 3, 0, 2]);
+		});
+
+		it("preserves registration order when priorities are equal", () => {
+			const entries = [
+				{ fn: async (v: string) => v, priority: 0, order: 0 },
+				{ fn: async (v: string) => v, priority: 0, order: 1 },
+				{ fn: async (v: string) => v, priority: 0, order: 2 },
+			];
+
+			const sorted = sortFilters(entries);
+			expect(sorted.map((e) => e.order)).toEqual([0, 1, 2]);
+		});
+	});
+
+	describe("applyFilterChain", () => {
+		it("applies filters in order", async () => {
+			const logger = createFakeLogger();
+			const result = await applyFilterChain([(v) => v + 1, (v) => v * 3], 2, logger, "test", false);
+			expect(result).toBe(9);
+		});
+
+		it("fail-open skips the throwing filter", async () => {
+			const logger = createFakeLogger();
+			const result = await applyFilterChain(
+				[
+					(v) => v + 10,
+					() => {
+						throw new Error("skip me");
+					},
+					(v) => v + 1,
+				],
+				0,
+				logger,
+				"test",
+				false,
+			);
+			expect(result).toBe(11);
+			expect(logger.errors).toHaveLength(1);
+		});
+
+		it("fail-closed throws on error", async () => {
+			const logger = createFakeLogger();
+			await expect(
+				applyFilterChain(
+					[
+						() => {
+							throw new Error("closed");
+						},
+					],
+					0,
+					logger,
+					"test",
+					true,
+				),
+			).rejects.toThrow("closed");
+		});
+	});
+});