File size: 5,166 Bytes
4e23b01 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | /**
* The transport-agnostic klient factory. Every transport entry point
* (`@moonshot-ai/klient/ipc|memory`) builds a `KlientChannel` and hands
* it here; the returned `Klient` is identical in shape and behavior no matter
* which transport carried the bytes.
*/
import type { KlientChannel, ScopeRef } from './channel.js';
import { globalContract, isStreamingContract } from '#/contract/index';
import { globalEvents, type KlientEventPayloads } from '#/contract/global/events';
import { sessionEvents, type SessionEventPayloads } from '#/contract/session/events';
import { agentEvents, type AgentEventPayloads } from '#/contract/agent/events';
import type { EventRegistration, StreamingProcedureContract } from '#/contract/types';
import { EventHub, type KlientEvents } from './events/hub.js';
import { createGlobalFacade, type GlobalFacade, type ScopedCaller, type ScopedStreamCaller } from './facade/global.js';
import { createSessionFacade, type SessionFacade } from './facade/session.js';
import { createAgentFacade, type AgentFacade } from './facade/agent.js';
import { parseChunk, parseInput, parseOutput } from './validation.js';
export interface KlientOptions {
/**
* Validate wire inputs/outputs and event payloads against the contract.
* Default `true`. Disable only on measured hot paths — validation is cheap
* (sub-µs for typical payloads) and is the drift tripwire.
*/
readonly validate?: boolean;
}
export interface SessionHandle extends SessionFacade {
readonly events: KlientEvents<SessionEventPayloads>;
agent(agentId: string): AgentHandle;
}
export interface AgentHandle extends AgentFacade {
readonly events: KlientEvents<AgentEventPayloads>;
}
export interface Klient {
readonly global: GlobalFacade;
readonly events: KlientEvents;
session(sessionId: string): SessionHandle;
close(): Promise<void>;
}
export function createKlientFromChannel(
channel: KlientChannel,
options: KlientOptions = {},
): Klient {
const validate = options.validate ?? true;
const call: ScopedCaller = async (scope, service, method, args, options) => {
const procedure = globalContract[service]?.[method];
if (procedure === undefined) {
// A facade method without a contract entry is a klient bug, not a wire error.
throw new Error(`no contract registered for ${service}.${method}`);
}
if (isStreamingContract(procedure)) {
throw new Error(`${service}.${method} is a streaming procedure — use callStream instead`);
}
const name = `${service}.${method}`;
const wireArgs = validate ? parseInput(name, procedure, args) : args;
const data = await channel.call(scope, service, method, wireArgs, options);
return validate ? parseOutput(name, procedure, data) : data;
};
const callStream: ScopedStreamCaller = (scope, service, method, args) => {
const procedure = globalContract[service]?.[method];
if (procedure === undefined) {
throw new Error(`no contract registered for ${service}.${method}`);
}
if (!isStreamingContract(procedure)) {
throw new Error(`${service}.${method} is not a streaming procedure — use call instead`);
}
const name = `${service}.${method}`;
const wireArgs = validate ? parseInput(name, procedure, args) : args;
const source = channel.stream(scope, service, method, wireArgs);
if (!validate) return source;
// Wrap the iterable to validate each chunk.
const contract = procedure as StreamingProcedureContract;
return {
[Symbol.asyncIterator]() {
const iter = source[Symbol.asyncIterator]();
return {
async next() {
const result = await iter.next();
if (result.done) return { done: true as const, value: undefined };
return { done: false, value: parseChunk(name, contract, result.value) };
},
async return(value?: unknown) {
await iter.return?.(value);
return { done: true as const, value: undefined };
},
};
},
};
};
const hubs = new Set<{ close(): void }>();
const makeHub = <TPayloadMap extends object>(
scope: ScopeRef,
registrations: Record<string, EventRegistration>,
): KlientEvents<TPayloadMap> => {
const hub = new EventHub<TPayloadMap>(channel, validate, scope, registrations);
hubs.add(hub);
return hub;
};
return {
global: createGlobalFacade(call, callStream),
events: makeHub<KlientEventPayloads>({}, globalEvents),
session(sessionId: string): SessionHandle {
const scope: ScopeRef = { sessionId };
return {
...createSessionFacade(call, sessionId),
events: makeHub<SessionEventPayloads>(scope, sessionEvents),
agent(agentId: string): AgentHandle {
const agentScope: ScopeRef = { sessionId, agentId };
return {
...createAgentFacade(call, agentScope),
events: makeHub<AgentEventPayloads>(agentScope, agentEvents),
};
},
};
},
close: () => {
for (const hub of hubs) {
hub.close();
}
hubs.clear();
return channel.close();
},
};
}
|