From b68ce91c3c62e438e4d8ef0eb7c1a9ed49bfc536 Mon Sep 17 00:00:00 2001 From: hristo2612 Date: Sun, 2 Aug 2026 17:21:12 +0300 Subject: [PATCH] refactor(gateway): one owner for connector construction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Connector setup was written three times: a legacy per-type block, an inline 4-case switch for `connectors.instances[]` at boot, and a near-verbatim copy of that switch in reloadConnectorInstances(). The copies had drifted apart. Both config forms now normalize into one list — a top-level connector is just an instance whose id defaults to its type — and a single factory constructs every connector. One wiring function creates, routes, registers, and starts each one. - Boot fire-and-forgets start() for every connector, matching what the comment claimed; the instances path used to await each handshake before HTTP listen. - Reload stops and restarts every connector, not just instance-declared ones; instanceConnectorIds is gone and the { started, stopped, errors } shape stays. - Session context connector names come from the live connector registry, so instance-configured ids are listed and reloads are reflected without any cached copy to refresh (SessionManager no longer takes connectorNames). - template/docs/connectors.md documents instances[]. --- .../src/gateway/__tests__/connectors.test.ts | 127 +++++ packages/jinn/src/gateway/server.ts | 448 +++++------------- .../platform-context-dispatch.test.ts | 8 +- .../__tests__/workflow-attempt-turns.test.ts | 3 - packages/jinn/src/sessions/manager.ts | 8 +- .../__tests__/workflow-recovery.test.ts | 2 +- .../__tests__/workflow-vertical.test.ts | 4 +- packages/jinn/template/docs/connectors.md | 31 ++ 8 files changed, 296 insertions(+), 335 deletions(-) create mode 100644 packages/jinn/src/gateway/__tests__/connectors.test.ts diff --git a/packages/jinn/src/gateway/__tests__/connectors.test.ts b/packages/jinn/src/gateway/__tests__/connectors.test.ts new file mode 100644 index 000000000..95fb70c33 --- /dev/null +++ b/packages/jinn/src/gateway/__tests__/connectors.test.ts @@ -0,0 +1,127 @@ +import { describe, expect, it } from "vitest"; +import { connectorInstancesFromConfig, createConnector } from "../server.js"; +import { SlackConnector } from "../../connectors/slack/index.js"; +import { DiscordConnector } from "../../connectors/discord/index.js"; +import { RemoteDiscordConnector } from "../../connectors/discord/remote.js"; +import { TelegramConnector } from "../../connectors/telegram/index.js"; +import { WhatsAppConnector } from "../../connectors/whatsapp/index.js"; +import { SessionManager } from "../../sessions/manager.js"; +import type { Connector, JinnConfig } from "../../shared/types.js"; + +// eslint-disable-next-line @typescript-eslint/no-explicit-any +const configWith = (connectors: Record, stt?: JinnConfig["stt"]): JinnConfig => + ({ connectors, stt }) as unknown as JinnConfig; + +const mixedConfig = configWith({ + slack: { appToken: "xapp-test", botToken: "xoxb-test", employee: "a-lead" }, + telegram: { botToken: "tg-test" }, + instances: [{ id: "slack-second", type: "slack", employee: "b-lead", appToken: "xapp-2", botToken: "xoxb-2" }], +}); + +describe("connectorInstancesFromConfig", () => { + it("defaults legacy top-level connector ids to their type and appends instances[]", () => { + const normalized = connectorInstancesFromConfig(mixedConfig); + expect(normalized.map((entry) => [entry.id, entry.type])).toEqual([ + ["slack", "slack"], + ["telegram", "telegram"], + ["slack-second", "slack"], + ]); + expect(normalized[0].employee).toBe("a-lead"); + expect(normalized[0].config.id).toBe("slack"); + expect(normalized[2].employee).toBe("b-lead"); + }); + + it("respects the per-type enable guards", () => { + const ids = (config: JinnConfig): string[] => connectorInstancesFromConfig(config).map((entry) => entry.id); + expect(ids(configWith({ slack: { appToken: "xapp-test" } }))).toEqual([]); + expect(ids(configWith({ discord: { proxyVia: "http://127.0.0.1:1" } }))).toEqual(["discord"]); + expect(ids(configWith({ discord: { channelId: "c1" } }))).toEqual([]); + expect(ids(configWith({ telegram: { allowFrom: [1] } }))).toEqual([]); + expect(ids(configWith({ whatsapp: {} }))).toEqual(["whatsapp"]); + }); + + it("forwards global stt settings to telegram connectors", () => { + const stt = { enabled: true, model: "test-model" }; + const normalized = connectorInstancesFromConfig(configWith({ telegram: { botToken: "tg-test" } }, stt)); + expect(normalized[0].config.stt).toEqual(stt); + }); + + it("skips instances without id or type, and duplicate ids", () => { + const normalized = connectorInstancesFromConfig( + configWith({ + slack: { appToken: "xapp-test", botToken: "xoxb-test" }, + instances: [ + { type: "slack", appToken: "a", botToken: "b" }, + { id: "no-type" }, + { id: "slack", type: "slack", appToken: "a", botToken: "b" }, + { id: "telegram-support", type: "telegram", botToken: "tg-2" }, + ], + }), + ); + expect(normalized.map((entry) => entry.id)).toEqual(["slack", "telegram-support"]); + }); +}); + +describe("createConnector", () => { + const build = (type: string, config: Record = {}): Connector => + createConnector({ id: `${type}-1`, type, config: { ...config, id: `${type}-1` } }); + + it("dispatches every supported type to its connector class", () => { + expect(build("slack", { appToken: "xapp-test", botToken: "xoxb-test" })).toBeInstanceOf(SlackConnector); + expect(build("discord", { botToken: "d-test" })).toBeInstanceOf(DiscordConnector); + expect(build("discord", { proxyVia: "http://127.0.0.1:1" })).toBeInstanceOf(RemoteDiscordConnector); + expect(build("telegram", { botToken: "tg-test" })).toBeInstanceOf(TelegramConnector); + expect(build("whatsapp")).toBeInstanceOf(WhatsAppConnector); + }); + + it("rejects an unknown type", () => { + expect(() => build("carrier-pigeon")).toThrow(/Unknown connector type "carrier-pigeon"/); + }); +}); + +describe("connector wiring", () => { + const stubConnector = (start: () => Promise): Connector => + ({ name: "stub", start, stop: async () => {}, sendMessage: async () => undefined, onMessage: () => {} }) as unknown as Connector; + + it("registers a connector before start resolves, so a hung handshake cannot block boot", async () => { + const registry = new Map(); + const hung = stubConnector(() => new Promise(() => {})); + // Mirrors server.ts: register first, then fire-and-forget start(). + registry.set("hung", hung); + const pending = hung.start().catch(() => {}); + expect(registry.has("hung")).toBe(true); + expect(await Promise.race([pending, Promise.resolve("still-booting")])).toBe("still-booting"); + }); + + it("keeps a rejecting start() from throwing into the boot path", async () => { + const failing = stubConnector(async () => { + throw new Error("handshake refused"); + }); + let logged: string | undefined; + expect(() => { + failing.start().catch((err: unknown) => { + logged = err instanceof Error ? err.message : String(err); + }); + }).not.toThrow(); + await new Promise((resolve) => setImmediate(resolve)); + expect(logged).toBe("handshake refused"); + }); + + it("derives session-context connector names from the live registry", () => { + const registry = new Map(); + for (const instance of connectorInstancesFromConfig(mixedConfig)) { + registry.set(instance.id, stubConnector(async () => {})); + } + const manager = new SessionManager(mixedConfig, new Map(), "connector-names-boot"); + manager.setConnectorProvider(() => registry); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const names = (): string[] => (manager as any).connectorNames(); + + expect(names()).toEqual(["slack", "telegram", "slack-second"]); + + // A reload replaces the registry contents; the context list must follow. + registry.delete("telegram"); + registry.set("discord-ops", stubConnector(async () => {})); + expect(names()).toEqual(["slack", "slack-second", "discord-ops"]); + }); +}); diff --git a/packages/jinn/src/gateway/server.ts b/packages/jinn/src/gateway/server.ts index 3bfd8da8b..e699ffb4a 100644 --- a/packages/jinn/src/gateway/server.ts +++ b/packages/jinn/src/gateway/server.ts @@ -6,7 +6,7 @@ import path from "node:path"; import { fileURLToPath } from "node:url"; import { randomUUID } from "node:crypto"; import { WebSocketServer, type WebSocket } from "ws"; -import type { JinnConfig, Connector, Employee, Engine, JsonObject, Session } from "../shared/types.js"; +import type { JinnConfig, Connector, Employee, Engine, JsonObject, Session, SlackConnectorConfig, TelegramConnectorConfig, WhatsAppConnectorConfig } from "../shared/types.js"; import { loadConfig, normalizeClaudeEngineConfig } from "../shared/config.js"; import { getModelRegistry, @@ -350,6 +350,84 @@ export function serveStatic( return true; } +/** A connector declaration with its id resolved — the one runtime shape. */ +export interface NormalizedConnector { + id: string; + type: string; + employee?: string; + /** Config handed to the connector constructor (always carries `id`). */ + config: Record; +} + +/** When a legacy top-level connector counts as configured. */ +// eslint-disable-next-line @typescript-eslint/no-explicit-any +const LEGACY_ENABLED: Record boolean> = { + slack: (config) => Boolean(config.appToken && config.botToken), + discord: (config) => Boolean(config.botToken || config.proxyVia), + telegram: (config) => Boolean(config.botToken), + whatsapp: () => true, +}; + +/** + * Flatten both config forms into one list: the legacy top-level connectors + * (`connectors.slack`, …) become instances whose id defaults to their type, + * followed by the explicitly named `connectors.instances[]`. + */ +export function connectorInstancesFromConfig(config: JinnConfig): NormalizedConnector[] { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const declared = (config.connectors ?? {}) as Record; + const instances: NormalizedConnector[] = []; + const seen = new Set(); + + const add = (id: string, type: string, raw: Record): void => { + if (seen.has(id)) { + logger.warn(`Duplicate connector instance id "${id}", skipping`); + return; + } + seen.add(id); + const connectorConfig: Record = { ...raw, id }; + // Speech-to-text is a global setting the telegram connector reads from its own config. + if (type === "telegram") connectorConfig.stt = config.stt; + instances.push({ id, type, employee: raw.employee as string | undefined, config: connectorConfig }); + }; + + for (const [type, enabled] of Object.entries(LEGACY_ENABLED)) { + const raw = declared[type]; + if (raw && enabled(raw)) add(type, type, raw); + } + + for (const instance of declared.instances ?? []) { + const { id, type, ...rest } = instance; + if (!id || !type) { + logger.warn(`Skipping connector instance without id or type`); + continue; + } + add(id, type, rest); + } + + return instances; +} + +/** The only place a connector is constructed. Throws on an unknown type. */ +export function createConnector(instance: NormalizedConnector): Connector { + const config = instance.config; + switch (instance.type) { + case "slack": + return new SlackConnector(config as unknown as SlackConnectorConfig); + case "discord": + // Remote mode proxies all Discord I/O through the primary instance. + return config.proxyVia + ? new RemoteDiscordConnector({ proxyVia: String(config.proxyVia), channelId: config.channelId as string | undefined }) + : new DiscordConnector(config as unknown as DiscordConnectorConfig); + case "telegram": + return new TelegramConnector(config as unknown as TelegramConnectorConfig); + case "whatsapp": + return new WhatsAppConnector(config as unknown as WhatsAppConnectorConfig); + default: + throw new Error(`Unknown connector type "${instance.type}" for instance "${instance.id}"`); + } +} + export type GatewayCleanup = () => Promise; export async function startGateway( @@ -637,359 +715,87 @@ export async function startGateway( hermes: hermesInteractiveEngine, }; - // Derive connector names from config - const connectorNames: string[] = []; - if (config.connectors?.slack?.appToken && config.connectors?.slack?.botToken) { - connectorNames.push("slack"); - } - if (config.connectors?.discord?.botToken || config.connectors?.discord?.proxyVia) { - connectorNames.push("discord"); - } - if (config.connectors?.telegram?.botToken) { - connectorNames.push("telegram"); - } - if (config.connectors?.whatsapp) { - connectorNames.push("whatsapp"); - } - // Build employee registry let employeeRegistry = scanOrg(); logger.info(`Loaded ${employeeRegistry.size} employee(s) from org directory`); - const sessionManager = new SessionManager(config, engines, connectorNames, bootId, (id) => employeeRegistry.get(id)); + const sessionManager = new SessionManager(config, engines, bootId, (id) => employeeRegistry.get(id)); - // Start connectors + // Start connectors — one normalized list covers both config forms. const connectors: Connector[] = []; const connectorMap = new Map(); - /** IDs of connectors created from config.connectors.instances[] (vs legacy top-level connectors) */ - const instanceConnectorIds = new Set(); /** - * Shared boilerplate for legacy top-level connectors: create, wire onMessage - * routing, register, and fire-and-forget start. Per-connector config guards - * and construction stay explicit at the call sites. + * Create one connector, wire its message routing, and register it. Registration + * happens before start so shutdown can clean up even while a handshake is in + * flight. Returns the start() promise so the caller decides: boot + * fire-and-forgets it (a slow handshake must not delay HTTP listen), reload + * awaits it to report start failures in its response. */ - const initConnector = (opts: { - id: string; - label: string; - create: () => Connector; - /** Read at message time so it tracks the captured config like before. */ - employee: () => string | undefined; - startMsg?: string; - }): void => { - try { - const connector = opts.create(); - connector.onMessage((msg) => { - const routeOpts: RouteOptions = {}; - const employeeName = opts.employee(); - if (employeeName) { - const emp = employeeRegistry.get(employeeName); - if (emp) routeOpts.employee = emp; - } - sessionManager.route(msg, connector, routeOpts).catch((err) => { - logger.error(`${opts.label} route error: ${err instanceof Error ? err.message : err}`); - }); - }); - // Push to registry before starting so shutdown can clean up even if start is in-flight. - connectors.push(connector); - connectorMap.set(opts.id, connector); - // Fire-and-forget: don't block boot — a slow handshake must not delay HTTP listen. - connector.start().catch((err) => { - logger.error(`Failed to start ${opts.label} connector: ${err instanceof Error ? err.message : err}`); + const initConnector = (instance: NormalizedConnector): Promise => { + const connector = createConnector(instance); + connector.onMessage((msg) => { + const routeOpts: RouteOptions = {}; + if (instance.employee) { + const emp = employeeRegistry.get(instance.employee); + if (emp) routeOpts.employee = emp; + } + sessionManager.route(msg, connector, routeOpts).catch((err) => { + logger.error(`${instance.id} route error: ${err instanceof Error ? err.message : err}`); }); - if (opts.startMsg) logger.info(opts.startMsg); - } catch (err) { - logger.error(`Failed to initialize ${opts.label} connector: ${err instanceof Error ? err.message : err}`); - } - }; - - if (config.connectors?.slack?.appToken && config.connectors?.slack?.botToken) { - const slackConfig = config.connectors.slack; - initConnector({ - id: "slack", - label: "Slack", - create: () => - new SlackConnector({ - appToken: slackConfig.appToken, - botToken: slackConfig.botToken, - allowFrom: slackConfig.allowFrom, - ignoreOldMessagesOnBoot: slackConfig.ignoreOldMessagesOnBoot, - }), - employee: () => config.connectors.slack?.employee, }); - } - - if (config.connectors?.discord?.proxyVia) { - // Remote mode: proxy all Discord operations through the primary instance - const discordConfig = config.connectors.discord; - initConnector({ - id: "discord", - label: "remote Discord", - create: () => - new RemoteDiscordConnector({ - proxyVia: discordConfig.proxyVia!, - channelId: discordConfig.channelId, - }), - employee: () => config.connectors.discord?.employee, - startMsg: "Discord remote connector starting", - }); - } else if (config.connectors?.discord?.botToken) { - // Primary mode: direct Discord bot connection - initConnector({ - id: "discord", - label: "Discord", - create: () => new DiscordConnector(config.connectors.discord as DiscordConnectorConfig), - employee: () => config.connectors.discord?.employee, - startMsg: "Discord connector starting", - }); - } - - if (config.connectors?.telegram?.botToken) { - const telegramConfig = config.connectors.telegram; - initConnector({ - id: "telegram", - label: "Telegram", - create: () => - new TelegramConnector({ - botToken: telegramConfig.botToken, - allowFrom: telegramConfig.allowFrom, - ignoreOldMessagesOnBoot: telegramConfig.ignoreOldMessagesOnBoot, - stt: config.stt, - }), - employee: () => config.connectors.telegram?.employee, - }); - } + connectors.push(connector); + connectorMap.set(instance.id, connector); + return connector.start(); + }; - if (config.connectors?.whatsapp) { - initConnector({ - id: "whatsapp", - label: "WhatsApp", - create: () => new WhatsAppConnector(config.connectors.whatsapp ?? {}), - employee: () => config.connectors.whatsapp?.employee, - startMsg: "WhatsApp connector starting (scan QR code if first run)", - }); - } + const describeConnector = (instance: NormalizedConnector): string => + `connector "${instance.id}" (type: ${instance.type}, employee: ${instance.employee || "default"})`; - // Process named connector instances (allows multiple connectors of the same type) - if (config.connectors?.instances) { - for (const instance of config.connectors.instances) { - const { id, type, employee, ...typeConfig } = instance; - if (!id || !type) { - logger.warn(`Skipping connector instance without id or type`); - continue; - } - if (connectorMap.has(id)) { - logger.warn(`Duplicate connector instance id "${id}", skipping`); - continue; - } + // Session context reads connector names off this map, so publish it before the + // first connector can deliver a message. + sessionManager.setConnectorProvider(() => connectorMap); - try { - let connector: Connector; - switch (type) { - case "discord": { - const discordConfig = { ...typeConfig, id } as DiscordConnectorConfig; - const discord = new DiscordConnector(discordConfig); - discord.onMessage((msg) => { - const routeOpts: RouteOptions = {}; - if (employee) { - const emp = employeeRegistry.get(employee); - if (emp) routeOpts.employee = emp; - } - sessionManager.route(msg, discord, routeOpts).catch((err) => { - logger.error(`${id} route error: ${err instanceof Error ? err.message : err}`); - }); - }); - await discord.start(); - connector = discord; - break; - } - case "slack": { - const slackConfig = { ...typeConfig, id } as any; - const slack = new SlackConnector(slackConfig); - slack.onMessage((msg) => { - const routeOpts: RouteOptions = {}; - if (employee) { - const emp = employeeRegistry.get(employee); - if (emp) routeOpts.employee = emp; - } - sessionManager.route(msg, slack, routeOpts).catch((err) => { - logger.error(`${id} route error: ${err instanceof Error ? err.message : err}`); - }); - }); - await slack.start(); - connector = slack; - break; - } - case "whatsapp": { - const whatsapp = new WhatsAppConnector({ ...typeConfig } as any); - whatsapp.onMessage((msg) => { - const routeOpts: RouteOptions = {}; - if (employee) { - const emp = employeeRegistry.get(employee); - if (emp) routeOpts.employee = emp; - } - sessionManager.route(msg, whatsapp, routeOpts).catch((err) => { - logger.error(`${id} route error: ${err instanceof Error ? err.message : err}`); - }); - }); - await whatsapp.start(); - connector = whatsapp; - break; - } - case "telegram": { - const telegramConfig = { ...typeConfig, id, stt: config.stt } as any; - const tg = new TelegramConnector(telegramConfig); - tg.onMessage((msg) => { - const routeOpts: RouteOptions = {}; - if (employee) { - const emp = employeeRegistry.get(employee); - if (emp) routeOpts.employee = emp; - } - sessionManager.route(msg, tg, routeOpts).catch((err) => { - logger.error(`${id} route error: ${err instanceof Error ? err.message : err}`); - }); - }); - await tg.start(); - connector = tg; - break; - } - default: - logger.warn(`Unknown connector type "${type}" for instance "${id}"`); - continue; - } - connectors.push(connector); - connectorMap.set(id, connector); - instanceConnectorIds.add(id); - logger.info(`Connector instance "${id}" (type: ${type}, employee: ${employee || "default"}) started`); - } catch (err) { - logger.error(`Failed to start connector instance "${id}": ${err instanceof Error ? err.message : err}`); - } + for (const instance of connectorInstancesFromConfig(config)) { + try { + // Fire-and-forget: don't block boot — a slow handshake must not delay HTTP listen. + initConnector(instance).catch((err) => { + logger.error(`Failed to start ${describeConnector(instance)}: ${err instanceof Error ? err.message : err}`); + }); + logger.info(`Starting ${describeConnector(instance)}`); + } catch (err) { + logger.error(`Failed to initialize ${describeConnector(instance)}: ${err instanceof Error ? err.message : err}`); } } - sessionManager.setConnectorProvider(() => connectorMap); - - // Reload connector instances from config (stop old instances, start new ones) + /** Stop every running connector and restart from fresh config (POST /api/connectors/reload). */ async function reloadConnectorInstances(): Promise<{ started: string[]; stopped: string[]; errors: string[] }> { - const freshConfig = loadConfig(); const started: string[] = []; const stopped: string[] = []; const errors: string[] = []; - // Find instance-based connectors (keys that came from instances array) - const instanceIds = new Set(); - if (freshConfig.connectors?.instances) { - for (const inst of freshConfig.connectors.instances) { - if (inst.id) instanceIds.add(inst.id); - } - } - - // Stop old instance connectors that are no longer in config or need refresh - for (const [id, connector] of connectorMap.entries()) { - // Skip legacy (top-level) connectors — only reload instance-based ones - if (!instanceConnectorIds.has(id)) continue; + for (const [id, connector] of [...connectorMap.entries()]) { try { await connector.stop(); connectorMap.delete(id); - instanceConnectorIds.delete(id); const idx = connectors.indexOf(connector); if (idx >= 0) connectors.splice(idx, 1); stopped.push(id); - logger.info(`Stopped connector instance "${id}" for reload`); + logger.info(`Stopped connector "${id}" for reload`); } catch (err) { errors.push(`Failed to stop ${id}: ${err instanceof Error ? err.message : err}`); } } - // Start new instances from fresh config - if (freshConfig.connectors?.instances) { - for (const instance of freshConfig.connectors.instances) { - const { id, type, employee, ...typeConfig } = instance; - if (!id || !type) continue; - if (connectorMap.has(id)) continue; - - try { - let connector: Connector; - switch (type) { - case "discord": { - const discordConfig = { ...typeConfig, id } as DiscordConnectorConfig; - const discord = new DiscordConnector(discordConfig); - discord.onMessage((msg) => { - const routeOpts: RouteOptions = {}; - if (employee) { - const emp = employeeRegistry.get(employee); - if (emp) routeOpts.employee = emp; - } - sessionManager.route(msg, discord, routeOpts).catch((err) => { - logger.error(`${id} route error: ${err instanceof Error ? err.message : err}`); - }); - }); - await discord.start(); - connector = discord; - break; - } - case "slack": { - const slackConfig = { ...typeConfig, id } as any; - const slack = new SlackConnector(slackConfig); - slack.onMessage((msg) => { - const routeOpts: RouteOptions = {}; - if (employee) { - const emp = employeeRegistry.get(employee); - if (emp) routeOpts.employee = emp; - } - sessionManager.route(msg, slack, routeOpts).catch((err) => { - logger.error(`${id} route error: ${err instanceof Error ? err.message : err}`); - }); - }); - await slack.start(); - connector = slack; - break; - } - case "whatsapp": { - const whatsapp = new WhatsAppConnector({ ...typeConfig } as any); - whatsapp.onMessage((msg) => { - const routeOpts: RouteOptions = {}; - if (employee) { - const emp = employeeRegistry.get(employee); - if (emp) routeOpts.employee = emp; - } - sessionManager.route(msg, whatsapp, routeOpts).catch((err) => { - logger.error(`${id} route error: ${err instanceof Error ? err.message : err}`); - }); - }); - await whatsapp.start(); - connector = whatsapp; - break; - } - case "telegram": { - const telegramConfig = { ...typeConfig, id, stt: config.stt } as any; - const tg = new TelegramConnector(telegramConfig); - tg.onMessage((msg) => { - const routeOpts: RouteOptions = {}; - if (employee) { - const emp = employeeRegistry.get(employee); - if (emp) routeOpts.employee = emp; - } - sessionManager.route(msg, tg, routeOpts).catch((err) => { - logger.error(`${id} route error: ${err instanceof Error ? err.message : err}`); - }); - }); - await tg.start(); - connector = tg; - break; - } - default: - errors.push(`Unknown connector type "${type}" for instance "${id}"`); - continue; - } - connectors.push(connector); - connectorMap.set(id, connector); - instanceConnectorIds.add(id); - started.push(id); - logger.info(`Connector instance "${id}" (type: ${type}, employee: ${employee || "default"}) started`); - } catch (err) { - errors.push(`Failed to start "${id}": ${err instanceof Error ? err.message : err}`); - logger.error(`Failed to start connector instance "${id}": ${err instanceof Error ? err.message : err}`); - } + for (const instance of connectorInstancesFromConfig(loadConfig())) { + // A connector that refused to stop is still live — leave it alone. + if (connectorMap.has(instance.id)) continue; + try { + await initConnector(instance); + started.push(instance.id); + logger.info(`Started ${describeConnector(instance)}`); + } catch (err) { + errors.push(`Failed to start "${instance.id}": ${err instanceof Error ? err.message : err}`); + logger.error(`Failed to start ${describeConnector(instance)}: ${err instanceof Error ? err.message : err}`); } } diff --git a/packages/jinn/src/sessions/__tests__/platform-context-dispatch.test.ts b/packages/jinn/src/sessions/__tests__/platform-context-dispatch.test.ts index ab2a0ce91..18a706e10 100644 --- a/packages/jinn/src/sessions/__tests__/platform-context-dispatch.test.ts +++ b/packages/jinn/src/sessions/__tests__/platform-context-dispatch.test.ts @@ -180,7 +180,7 @@ describe("SessionManager platform context dispatch", () => { { sessionId: "codex-native", result: "ok" }, ]; const engine = capturingEngine("codex", runs, (runNumber) => results[runNumber - 1]); - const manager = new managerModule.SessionManager(makeConfig(), new Map([["codex", engine]]), [], "boot-a" as any); + const manager = new managerModule.SessionManager(makeConfig(), new Map([["codex", engine]]), "boot-a" as any); const connector = connectorStub(); await manager.route(incoming("initial success"), connector); @@ -212,7 +212,7 @@ describe("SessionManager platform context dispatch", () => { return { sessionId: "codex-native", result: "ok" }; }); const engines = new Map([["codex", engine]]); - const manager = new managerModule.SessionManager(makeConfig(), engines, [], "boot-a" as any); + const manager = new managerModule.SessionManager(makeConfig(), engines, "boot-a" as any); const connector = connectorStub(); await manager.route(incoming("initial success"), connector); @@ -243,7 +243,7 @@ describe("SessionManager platform context dispatch", () => { ["claude", capturingEngine("claude", betaRuns)], ]); let config = makeConfig(); - const manager = new managerModule.SessionManager(config, engines, [], "boot-a" as any); + const manager = new managerModule.SessionManager(config, engines, "boot-a" as any); const connector = connectorStub(); await manager.route(incoming("turn 1"), connector); @@ -272,7 +272,7 @@ describe("SessionManager platform context dispatch", () => { await manager.route(incoming("config stable", "channel-b"), connector); expect(alphaRuns.slice(7, 9).map(headingCount)).toEqual([1, 0]); - const restarted = new managerModule.SessionManager(config, engines, [], "boot-b" as any); + const restarted = new managerModule.SessionManager(config, engines, "boot-b" as any); await restarted.route(incoming("after restart", "channel-b"), connector); await restarted.route(incoming("restart stable", "channel-b"), connector); expect(alphaRuns.slice(9, 11).map(headingCount)).toEqual([1, 0]); diff --git a/packages/jinn/src/sessions/__tests__/workflow-attempt-turns.test.ts b/packages/jinn/src/sessions/__tests__/workflow-attempt-turns.test.ts index 78d445267..6c9bc5142 100644 --- a/packages/jinn/src/sessions/__tests__/workflow-attempt-turns.test.ts +++ b/packages/jinn/src/sessions/__tests__/workflow-attempt-turns.test.ts @@ -79,7 +79,6 @@ function managerWith(runs: EngineRunOpts[]) { return new managerModule.SessionManager( config(), new Map([[engine.name, engine]]), - [], "test-boot", (id) => id === employee.name ? employee : undefined, ); @@ -113,7 +112,6 @@ describe("workflow attempt per-turn completion", () => { const manager = new managerModule.SessionManager( config(), new Map([[engine.name, engine]]), - [], "test-boot", (id) => id === employee.name ? employee : undefined, ); @@ -224,7 +222,6 @@ describe("workflow attempt per-turn completion", () => { const manager = new managerModule.SessionManager( config(), new Map([[engine.name, engine]]), - [], "test-boot", (id) => id === employee.name ? employee : undefined, ); diff --git a/packages/jinn/src/sessions/manager.ts b/packages/jinn/src/sessions/manager.ts index 5b359649f..b74de8583 100644 --- a/packages/jinn/src/sessions/manager.ts +++ b/packages/jinn/src/sessions/manager.ts @@ -145,7 +145,6 @@ export function mergeTransportMeta( export class SessionManager { private config: JinnConfig; private engines: Map; - private connectorNames: string[]; private gatewayBootId: string; private queue = new SessionQueue(); private connectorProvider: () => Map = () => new Map(); @@ -155,13 +154,11 @@ export class SessionManager { constructor( config: JinnConfig, engines: Map, - connectorNames: string[] = [], gatewayBootId = "", private readonly employeeProvider: (id: string) => Employee | undefined = () => undefined, ) { this.config = config; this.engines = engines; - this.connectorNames = connectorNames; this.gatewayBootId = gatewayBootId; this.recoverWorkflowAttemptDispatches(); } @@ -174,6 +171,9 @@ export class SessionManager { this.connectorProvider = provider; } + /** Live connector ids — reflects reloads, with no cached copy to refresh. */ + private connectorNames(): string[] { return [...this.connectorProvider().keys()]; } + setConfig(config: JinnConfig): void { this.config = config; } @@ -431,7 +431,7 @@ export class SessionManager { user: msg.user, employee, engine: session.engine, - connectors: this.connectorNames, + connectors: this.connectorNames(), config: this.config, gatewayBootId: this.gatewayBootId, sessionId: session.id, diff --git a/packages/jinn/src/workflows/__tests__/workflow-recovery.test.ts b/packages/jinn/src/workflows/__tests__/workflow-recovery.test.ts index 214285797..cddcbf2a5 100644 --- a/packages/jinn/src/workflows/__tests__/workflow-recovery.test.ts +++ b/packages/jinn/src/workflows/__tests__/workflow-recovery.test.ts @@ -364,7 +364,7 @@ describe("Workflow retry, cancellation, and restart recovery", () => { const sessionConfig = { gateway: { port: 0, host: "127.0.0.1" }, engines: { default: "test-engine", claude: { bin: "", model: "test" }, codex: { bin: "", model: "test" }, "test-engine": {} }, connectors: {}, logging: { file: false, stdout: false, level: "error" } } as unknown as JinnConfig; - const sessions = new SessionManager(sessionConfig, new Map([[engine.name, engine]]), [], "reconstructed", + const sessions = new SessionManager(sessionConfig, new Map([[engine.name, engine]]), "reconstructed", (id) => id === employee.name ? employee : undefined); service.dispose(); service = new WorkflowService({ repository, executor: new RealExecutor(sessions), diff --git a/packages/jinn/src/workflows/__tests__/workflow-vertical.test.ts b/packages/jinn/src/workflows/__tests__/workflow-vertical.test.ts index afe0f5244..128ee1b8f 100644 --- a/packages/jinn/src/workflows/__tests__/workflow-vertical.test.ts +++ b/packages/jinn/src/workflows/__tests__/workflow-vertical.test.ts @@ -241,7 +241,7 @@ beforeEach(() => { database = openWorkflowDatabase(path.join(root, "workflows.db")); repository = new WorkflowRepository(database); engine = new DeferredEngine(); - manager = new SessionManager(config, new Map([["claude", engine], ["codex", engine]]), [], "vertical-boot", (id) => id === employee.name ? employee : undefined); + manager = new SessionManager(config, new Map([["claude", engine], ["codex", engine]]), "vertical-boot", (id) => id === employee.name ? employee : undefined); changes = []; service = createService(); }); @@ -296,7 +296,7 @@ describe("first Workflow vertical", () => { engine.resolve({ sessionId: "native-recovered", result: "Recovered.\n```jinn-output\n{\"result\":\"ok\"}\n```", durationMs: 1 }); await vi.waitFor(() => expect(getSession(repository.getRun(authored.id, lost.id)!.attempts[0]!.sessionId!)?.attemptOutcome).toBe("succeeded")); - manager = new SessionManager(config, new Map([["claude", engine], ["codex", engine]]), [], "reconstructed", (id) => id === employee.name ? employee : undefined); + manager = new SessionManager(config, new Map([["claude", engine], ["codex", engine]]), "reconstructed", (id) => id === employee.name ? employee : undefined); service = createService(); expect(await service.recover(new Date().toISOString())).toEqual({ resumedRuns: 1, resumedWaits: 0 }); expect(await service.recover(new Date().toISOString())).toEqual({ resumedRuns: 0, resumedWaits: 0 }); diff --git a/packages/jinn/template/docs/connectors.md b/packages/jinn/template/docs/connectors.md index a6c2eccd6..806933971 100644 --- a/packages/jinn/template/docs/connectors.md +++ b/packages/jinn/template/docs/connectors.md @@ -63,6 +63,37 @@ Reactions provide visual feedback during processing: - `@mention`: messages mentioning a specific employee name route to that employee - Thread continuity: replies in a thread continue with the same employee +## Named Instances + +`connectors.instances[]` declares connectors explicitly, so you can run several of the +same type — each with its own credentials and its own employee. + +```yaml +connectors: + instances: + - id: slack-support # unique connector id + type: slack # slack | discord | telegram | whatsapp + employee: support-lead # optional — who handles messages from this connector + appToken: xapp-... # remaining keys are the type's own config + botToken: xoxb-... + - id: telegram-ops + type: telegram + botToken: ... +``` + +Both config forms produce the same thing at runtime: a top-level connector +(`connectors.slack`, `connectors.discord`, `connectors.telegram`, `connectors.whatsapp`) +is simply an instance whose `id` defaults to its type. So `connectors.slack` and an +instance with `id: slack` are the same connector — the duplicate is skipped, as is any +entry missing `id` or `type`. + +Connector ids are what the rest of the gateway addresses: + +- `POST /api/connectors//send` and the `send_connector_message` company tool +- the `## Available connectors` list in every session's context +- `POST /api/connectors/reload`, which stops every running connector and restarts it + from the current `config.yaml` — regardless of which form declared it + ## Future Connectors The connector interface is designed for additional platforms: