feat(companion): add runtime event subscription hooks

This commit is contained in:
William Valentin
2026-02-16 18:43:05 -08:00
parent ec2811eaef
commit 96b11bd60f
5 changed files with 88 additions and 1 deletions
+1
View File
@@ -11,6 +11,7 @@ export { CompanionHeartbeatLoop } from './heartbeatLoop.js';
export type {
CompanionRuntimeClientOptions,
CompanionEventHandler,
RegisterNodeInput,
ListNodesInput,
SetNodeStatusInput,
+55
View File
@@ -96,6 +96,61 @@ afterAll(async () => {
});
describe('CompanionRuntimeClient', () => {
it('dispatches gateway events to subscribed handlers and supports unsubscribe', () => {
const client = new CompanionRuntimeClient({
url: 'ws://127.0.0.1:1',
});
const handler = vi.fn();
const unsubscribe = client.subscribeEvents(handler);
(client as unknown as { handleMessage: (raw: string) => void }).handleMessage(
JSON.stringify({
id: 42,
event: 'agent.stream',
data: { token: 'hello' },
}),
);
expect(handler).toHaveBeenCalledWith('agent.stream', { token: 'hello' });
unsubscribe();
(client as unknown as { handleMessage: (raw: string) => void }).handleMessage(
JSON.stringify({
id: 43,
event: 'agent.stream',
data: { token: 'world' },
}),
);
expect(handler).toHaveBeenCalledTimes(1);
});
it('isolates subscriber callback failures', () => {
const client = new CompanionRuntimeClient({
url: 'ws://127.0.0.1:1',
});
const badHandler = vi.fn(() => {
throw new Error('subscriber failed');
});
const goodHandler = vi.fn();
client.subscribeEvents(badHandler);
client.subscribeEvents(goodHandler);
expect(() => {
(client as unknown as { handleMessage: (raw: string) => void }).handleMessage(
JSON.stringify({
id: 44,
event: 'agent.stream',
data: { token: 'safe' },
}),
);
}).not.toThrow();
expect(badHandler).toHaveBeenCalledOnce();
expect(goodHandler).toHaveBeenCalledWith('agent.stream', { token: 'safe' });
});
it('connects and performs node registration + capability discovery', async () => {
if (!LISTEN_ALLOWED) {
return;
+17
View File
@@ -41,6 +41,8 @@ export interface CompanionRuntimeClientOptions {
websocketFactory?: (url: string) => WebSocket;
}
export type CompanionEventHandler = (event: string, data: unknown) => void;
export interface RegisterNodeInput {
nodeId: string;
role: string;
@@ -257,6 +259,7 @@ export class CompanionRuntimeClient {
private connectPromise: Promise<void> | null = null;
private nextId = 1;
private pending = new Map<number, PendingRequest>();
private readonly eventHandlers = new Set<CompanionEventHandler>();
constructor(options: CompanionRuntimeClientOptions) {
this.url = options.url;
@@ -342,6 +345,13 @@ export class CompanionRuntimeClient {
ws.close(code, reason);
}
subscribeEvents(handler: CompanionEventHandler): () => void {
this.eventHandlers.add(handler);
return () => {
this.eventHandlers.delete(handler);
};
}
async call<T>(method: string, params?: Record<string, unknown>): Promise<T> {
if (!this.connected) {
if (!this.autoConnect) {
@@ -498,6 +508,13 @@ export class CompanionRuntimeClient {
}
if ('event' in parsed) {
for (const handler of this.eventHandlers) {
try {
handler(parsed.event, parsed.data);
} catch {
// Event subscribers are userland callbacks; isolate failures.
}
}
return;
}