diff --git a/CHANGELOG.md b/CHANGELOG.md index 602c43f..2b7e86a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ ## Unreleased +- Gateway (#144): sender allowlists and the gateway toolset are one + `GatewayPolicy`, built from config at startup (`src/gateway/access.ts`). + Platform adapters declare `capabilities` (`{ kind: "text", max_reply_chars }`); + the runner logs them. **Breaking for custom adapters:** `PlatformAdapter` + requires `capabilities`, and `create_idle_adapter` takes it as a third + argument. - Gateway (#144): conversations queue on the shared `SessionManager` instead of their own promise chains. Beyond `max_conversations`, the least recently used conversation is evicted (it was the oldest inserted). Only Telegram's `/start` diff --git a/docs/architecture/extending.md b/docs/architecture/extending.md index eb82bf6..0b07753 100644 --- a/docs/architecture/extending.md +++ b/docs/architecture/extending.md @@ -156,16 +156,23 @@ A platform adapter is an implementation of the `PlatformAdapter` contract ```ts export interface PlatformAdapter { readonly name: string; + readonly capabilities: AdapterCapabilities; // { kind: "text", max_reply_chars?: number } start(): Promise; stop(): Promise; } ``` +`capabilities` declares what the adapter can deliver. Every platform today is +`kind: "text"` (one plain reply per message); set `max_reply_chars` to the +platform's message cap and split replies to it. The runner logs it at start. + Inbound flow: the adapter normalizes a platform event into `(platform, chat_id, user_id, text)` and calls `run_inbound_message` with the -shared `InboundHandler`; the handler is the bus's `handle`, which resolves to -the reply text; the adapter then sends the reply through the platform's send -API. The bus owns history, serialization, and error sanitization - adapters +shared `InboundHandler`; the handler is the bus's `reply`, and +`run_inbound_message` reduces its result to the reply text; the adapter then sends the reply through the platform's send +API. The bus owns history, serialization, and error sanitization, and checks +each sender against the `GatewayPolicy` built from config at startup +(`src/gateway/access.ts`: allowlists plus the gateway toolset) - adapters stay thin. Recipe (mirroring [`telegram.ts`](../../src/gateway/telegram.ts), the @@ -189,7 +196,7 @@ await send_reply(reply); 3. **Handle missing credentials with the idle-adapter pattern.** When the platform's token env var is absent, return - `create_idle_adapter("platform", "ENV_VAR not set")` instead of throwing. + `create_idle_adapter("platform", "ENV_VAR not set", CAPABILITIES)` instead of throwing. The adapter logs once why it is idle and no-ops on start/stop, so the gateway keeps serving the platforms that *do* have credentials - `run_gateway` only needs one valid platform. @@ -198,7 +205,7 @@ await send_reply(reply); structural `RawSocket` type works on both Bun and Node runtimes), `sanitize_agent_error` for reply strings on failure, and the whitespace splitter `split_text` from `src/gateway/format.ts` for size-capped - transports (telegram caps at 4096, discord at 2000, twitch at 512). + transports (telegram caps at 4096, discord at 2000, twitch at 450 to stay under its 500-char limit). **Testing.** `test/gateway.test.ts` covers the webhook adapter over a real `node:http` server (a `on_listening` test hook reports the bound port), the diff --git a/src/gateway/access.ts b/src/gateway/access.ts index 3533ccd..b852c3f 100644 --- a/src/gateway/access.ts +++ b/src/gateway/access.ts @@ -1,5 +1,7 @@ /** - * Gateway access: public-platform allowlists and the safe default toolset. + * Gateway access: public-platform allowlists and the safe default toolset, + * gathered into one GatewayPolicy built at startup (decision 0001: Profiles + * and Policy evolve from this file). */ import type { AgentConfig } from "../agent/config.js"; import { logger } from "../util/log.js"; @@ -42,6 +44,21 @@ export function is_gateway_sender_allowed( return user_ok && chat_ok; } +/** What the gateway lets in and what its agent may use; built once from config. */ +export interface GatewayPolicy { + /** Tools the shared gateway agent registers. */ + readonly tools_enabled: "all" | readonly string[]; + /** True when the sender may talk to the agent; logs a denial. */ + allows(platform: string, chat_id: string, user_id: string): boolean; +} + +export function create_gateway_policy(config: AgentConfig): GatewayPolicy { + return { + tools_enabled: gateway_tools_enabled(config), + allows: (platform, chat_id, user_id) => check_gateway_sender(config, platform, chat_id, user_id), + }; +} + /** Logs and returns false when the sender is not on the allowlist. */ export function check_gateway_sender( config: AgentConfig, diff --git a/src/gateway/bus.ts b/src/gateway/bus.ts index 61e6573..9447650 100644 --- a/src/gateway/bus.ts +++ b/src/gateway/bus.ts @@ -12,7 +12,7 @@ import { history_after_run_error } from "../agent/loop.js"; import type { Message } from "../providers/types.js"; import { create_session_manager, type SessionManager } from "../session/manager.js"; import { logger } from "../util/log.js"; -import { check_gateway_sender } from "./access.js"; +import { create_gateway_policy, type GatewayPolicy } from "./access.js"; import { sanitize_agent_error, type GatewayReply } from "./types.js"; export interface GatewayBusOptions { @@ -22,6 +22,8 @@ export interface GatewayBusOptions { export interface BusParams { config: AgentConfig; + /** Who may talk to the agent; defaults to `create_gateway_policy(config)`. */ + policy?: GatewayPolicy; agent_factory: () => Agent; /** Subscribe to agent tool events for debug logging (creates the agent). */ wire_tool_logging?: boolean; @@ -31,7 +33,7 @@ const DEFAULT_HISTORY_CAP = 40; const DEFAULT_MAX_CONVERSATIONS = 200; export class GatewayBus { - private readonly config: AgentConfig; + private readonly policy: GatewayPolicy; private readonly agent_factory: () => Agent; private agent: Agent | undefined; private readonly histories: Map = new Map(); @@ -41,7 +43,7 @@ export class GatewayBus { private stop_logging: (() => void) | undefined; constructor(params: BusParams, options?: GatewayBusOptions) { - this.config = params.config; + this.policy = params.policy ?? create_gateway_policy(params.config); this.agent_factory = params.agent_factory; this.history_cap = options?.history_cap ?? DEFAULT_HISTORY_CAP; this.max_conversations = options?.max_conversations ?? DEFAULT_MAX_CONVERSATIONS; @@ -58,7 +60,7 @@ export class GatewayBus { /** Like `handle`, with the run's usage and whether it failed; undefined when the sender is denied. */ async reply(platform: string, chat_id: string, user_id: string, text: string): Promise { - if (check_gateway_sender(this.config, platform, chat_id, user_id) === false) { + if (this.policy.allows(platform, chat_id, user_id) === false) { return undefined; } const key = conversation_key(platform, chat_id); diff --git a/src/gateway/discord.ts b/src/gateway/discord.ts index 2e84949..d0dcb92 100644 --- a/src/gateway/discord.ts +++ b/src/gateway/discord.ts @@ -7,13 +7,15 @@ import { logger } from "../util/log.js"; import { sleep } from "../util/sleep.js"; import { platform_token_env, read_platform_token } from "./token_env.js"; import type { AdapterParams, PlatformAdapter, RawSocket } from "./types.js"; -import { create_idle_adapter, open_socket, run_inbound_message } from "./types.js"; +import { create_idle_adapter, open_socket, run_inbound_message, type AdapterCapabilities } from "./types.js"; const DISCORD_API = "https://discord.com/api/v10"; const GATEWAY_URL = "wss://gateway.discord.gg/?v=10&encoding=json"; /** Guild messages + message content + direct messages. */ const INTENTS = 512 | 32768 | 4096; export const DISCORD_BACKOFF_MS = [5000, 10000, 20000, 30000] as const; +export const DISCORD_MAX_MESSAGE_CHARS = 2000; +export const DISCORD_CAPABILITIES: AdapterCapabilities = { kind: "text", max_reply_chars: DISCORD_MAX_MESSAGE_CHARS }; /** Auth / sharding / API-version failures — reconnecting cannot recover. */ const DISCORD_FATAL_CLOSE = new Set([4004, 4010, 4011, 4012, 4013, 4014]); /** Live heartbeat timers keyed by socket, cleared when the session ends. */ @@ -41,13 +43,14 @@ interface DiscordSessionState { export function create_discord_adapter(params: AdapterParams): PlatformAdapter { const token = read_platform_token(params.config, "discord"); if (token === undefined) { - return create_idle_adapter("discord", `${platform_token_env(params.config, "discord")} not set`); + return create_idle_adapter("discord", `${platform_token_env(params.config, "discord")} not set`, DISCORD_CAPABILITIES); } let running = false; let socket: RawSocket | undefined; let stop_controller = new AbortController(); return { name: "discord", + capabilities: DISCORD_CAPABILITIES, start: async () => { running = true; stop_controller = new AbortController(); @@ -226,7 +229,7 @@ async function rest_send_message(params: AdapterParams, channel_id: string, text if (token === undefined || channel_id === "") { return; } - for (const chunk of split_chunks(text, 2000)) { + for (const chunk of split_chunks(text, DISCORD_MAX_MESSAGE_CHARS)) { const response = await fetch(`${DISCORD_API}/channels/${channel_id}/messages`, { method: "POST", headers: { authorization: `Bot ${token}`, "content-type": "application/json" }, diff --git a/src/gateway/runner.ts b/src/gateway/runner.ts index 298abe5..8cb3653 100644 --- a/src/gateway/runner.ts +++ b/src/gateway/runner.ts @@ -5,7 +5,7 @@ import { create_agent_with_plugins, type Agent } from "../agent/agent.js"; import type { AgentConfig } from "../agent/config.js"; import { logger } from "../util/log.js"; -import { gateway_tools_enabled } from "./access.js"; +import { create_gateway_policy } from "./access.js"; import { GatewayBus } from "./bus.js"; import { create_discord_adapter } from "./discord.js"; import { create_telegram_adapter } from "./telegram.js"; @@ -15,9 +15,10 @@ import { create_webhook_adapter } from "./webhook.js"; /** Preload plugins, then hand that same agent to the bus factory. */ export async function create_gateway_bus(config: AgentConfig): Promise<{ agent: Agent; bus: GatewayBus }> { - const agent_config = { ...config, tools_enabled: gateway_tools_enabled(config) }; + const policy = create_gateway_policy(config); + const agent_config = { ...config, tools_enabled: policy.tools_enabled }; const agent = await create_agent_with_plugins(agent_config); - const bus = new GatewayBus({ config, agent_factory: () => agent, wire_tool_logging: true }); + const bus = new GatewayBus({ config, policy, agent_factory: () => agent, wire_tool_logging: true }); return { agent, bus }; } @@ -102,7 +103,8 @@ async function start_all_adapters(adapters: readonly PlatformAdapter[]): Promise for (const adapter of adapters) { try { await adapter.start(); - logger.info(`gateway adapter started: ${adapter.name}`); + const cap = adapter.capabilities.max_reply_chars; + logger.info(`gateway adapter started: ${adapter.name} (${adapter.capabilities.kind}${cap === undefined ? "" : `, replies split at ${cap} chars`})`); } catch (error) { logger.error(`gateway adapter failed to start: ${adapter.name}`, error); } diff --git a/src/gateway/telegram.ts b/src/gateway/telegram.ts index e7ffe9a..28465fc 100644 --- a/src/gateway/telegram.ts +++ b/src/gateway/telegram.ts @@ -7,9 +7,10 @@ import { sleep } from "../util/sleep.js"; import { logger } from "../util/log.js"; import { platform_token_env, read_platform_token } from "./token_env.js"; import type { AdapterParams, PlatformAdapter } from "./types.js"; -import { create_idle_adapter, run_inbound_message } from "./types.js"; +import { create_idle_adapter, run_inbound_message, type AdapterCapabilities } from "./types.js"; export const TELEGRAM_MAX_MESSAGE_CHARS = 4096; +export const TELEGRAM_CAPABILITIES: AdapterCapabilities = { kind: "text", max_reply_chars: TELEGRAM_MAX_MESSAGE_CHARS }; interface TelegramUpdate { update_id?: number; @@ -25,11 +26,12 @@ export const TELEGRAM_BACKOFF_MS = [2000, 4000, 8000, 16000, 30000] as const; export function create_telegram_adapter(params: AdapterParams): PlatformAdapter { const token = read_platform_token(params.config, "telegram"); if (token === undefined) { - return create_idle_adapter("telegram", `${platform_token_env(params.config, "telegram")} not set`); + return create_idle_adapter("telegram", `${platform_token_env(params.config, "telegram")} not set`, TELEGRAM_CAPABILITIES); } let running = false; return { name: "telegram", + capabilities: TELEGRAM_CAPABILITIES, start: async () => { running = true; void poll_loop(params, token, () => running); diff --git a/src/gateway/twitch.ts b/src/gateway/twitch.ts index d4ceb51..38501a9 100644 --- a/src/gateway/twitch.ts +++ b/src/gateway/twitch.ts @@ -8,11 +8,12 @@ import { logger } from "../util/log.js"; import { sleep } from "../util/sleep.js"; import { platform_token_env, read_platform_token } from "./token_env.js"; import type { AdapterParams, PlatformAdapter, RawSocket } from "./types.js"; -import { create_idle_adapter, open_socket, run_inbound_message } from "./types.js"; +import { create_idle_adapter, open_socket, run_inbound_message, type AdapterCapabilities } from "./types.js"; export const TWITCH_IRC_URL = "wss://irc-ws.chat.twitch.tv:443"; /** Content budget under Twitch's 500-char message / 512-byte IRC line caps. */ export const TWITCH_MESSAGE_CAP = 450; +export const TWITCH_CAPABILITIES: AdapterCapabilities = { kind: "text", max_reply_chars: TWITCH_MESSAGE_CAP }; export const TWITCH_BACKOFF_MS = [5000, 10000, 20000, 30000] as const; const TWITCH_CHUNK_GAP_MS = 1600; @@ -34,13 +35,14 @@ export function create_twitch_adapter(params: AdapterParams): PlatformAdapter { if (twitch === undefined) { const token_env = platform_token_env(params.config, "twitch"); const reason = token_env === "LICH_TWITCH_OAUTH_TOKEN" ? "LICH_TWITCH_OAUTH_TOKEN / NICK not set" : `${token_env} / LICH_TWITCH_NICK not set`; - return create_idle_adapter("twitch", reason); + return create_idle_adapter("twitch", reason, TWITCH_CAPABILITIES); } let running = false; let socket: RawSocket | undefined; let stop_controller = new AbortController(); return { name: "twitch", + capabilities: TWITCH_CAPABILITIES, start: async () => { running = true; stop_controller = new AbortController(); diff --git a/src/gateway/types.ts b/src/gateway/types.ts index f2b20cd..d8bfae2 100644 --- a/src/gateway/types.ts +++ b/src/gateway/types.ts @@ -23,9 +23,20 @@ export interface ReplySink { send(text: string): Promise; } +/** + * What an adapter can deliver (decision 0001: adapters declare capabilities). + * Every platform adapter today is "text": a plain reply per message, split + * into chunks of at most `max_reply_chars` when the platform caps length. + */ +export interface AdapterCapabilities { + readonly kind: "text"; + readonly max_reply_chars?: number; +} + /** Lifecycle contract implemented by every platform adapter. */ export interface PlatformAdapter { readonly name: string; + readonly capabilities: AdapterCapabilities; start(): Promise; stop(): Promise; } @@ -81,10 +92,11 @@ export function sanitize_agent_error(error: unknown): string { } /** Adapter that logs why it is idle once and otherwise does nothing. */ -export function create_idle_adapter(name: string, reason: string): PlatformAdapter { +export function create_idle_adapter(name: string, reason: string, capabilities: AdapterCapabilities): PlatformAdapter { logger.warn(`gateway ${name} adapter idle: ${reason}`); return { name, + capabilities, start: async () => undefined, stop: async () => undefined, }; diff --git a/src/gateway/webhook.ts b/src/gateway/webhook.ts index fafb775..b1ba21d 100644 --- a/src/gateway/webhook.ts +++ b/src/gateway/webhook.ts @@ -35,6 +35,8 @@ export function create_webhook_adapter(params: WebhookAdapterParams): PlatformAd return { name: "webhook", + // One JSON reply per request; no length cap. + capabilities: { kind: "text" }, start: async () => { assert_bind_allowed(host, token); server = createServer((request, response) => { diff --git a/test/gateway.test.ts b/test/gateway.test.ts index c56c261..79d3166 100644 --- a/test/gateway.test.ts +++ b/test/gateway.test.ts @@ -4,15 +4,20 @@ * No real network to telegram/discord/twitch — webhook binds an ephemeral * port (0) on loopback only. */ -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { mkdirSync, mkdtempSync, rmSync } from "node:fs"; import { join } from "node:path"; import { parse_agent_config } from "../src/agent/config.js"; import { + create_gateway_policy, DEFAULT_GATEWAY_TOOLS_ENABLED, gateway_tools_enabled, is_gateway_sender_allowed, + type GatewayPolicy, } from "../src/gateway/access.js"; +import { create_discord_adapter } from "../src/gateway/discord.js"; +import { create_twitch_adapter } from "../src/gateway/twitch.js"; +import { logger } from "../src/util/log.js"; import { assert_bind_allowed, create_webhook_adapter, MAX_WEBHOOK_BODY_BYTES } from "../src/gateway/webhook.js"; import { format_agent_reply, split_text } from "../src/gateway/format.js"; import { create_telegram_adapter } from "../src/gateway/telegram.js"; @@ -607,6 +612,60 @@ describe("gateway access", () => { rmSync(work_dir, { recursive: true, force: true }); } }); + + it("builds one GatewayPolicy from config: toolset plus the sender allowlists", () => { + const work_dir = temp_work_dir(); + try { + const policy = create_gateway_policy(config_for(work_dir, { allowed_users: { discord: ["u1"] } })); + expect(policy.tools_enabled).toEqual([...DEFAULT_GATEWAY_TOOLS_ENABLED]); + expect(policy.allows("webhook", "c", "anyone")).toBe(true); + expect(policy.allows("discord", "c", "u1")).toBe(true); + expect(policy.allows("discord", "c", "u2")).toBe(false); + expect(policy.allows("telegram", "c", "u1")).toBe(false); + } finally { + rmSync(work_dir, { recursive: true, force: true }); + } + }); + + it("lets the bus use an injected policy instead of the config's", async () => { + const work_dir = temp_work_dir(); + try { + const records: RunRecord[] = []; + const deny_all: GatewayPolicy = { tools_enabled: [], allows: () => false }; + const bus = new GatewayBus({ config: config_for(work_dir), policy: deny_all, agent_factory: () => recording_agent(records) }); + expect(await bus.handle("webhook", "c1", "u1", "hello")).toBeUndefined(); + expect(records).toEqual([]); + } finally { + rmSync(work_dir, { recursive: true, force: true }); + } + }); +}); + +describe("adapter capabilities", () => { + it("declares each platform's reply cap, also on idle adapters", () => { + const saved = { ...process.env }; + for (const key of ["LICH_TELEGRAM_BOT_TOKEN", "LICH_DISCORD_BOT_TOKEN", "LICH_TWITCH_OAUTH_TOKEN", "LICH_TWITCH_NICK"]) { + delete process.env[key]; + } + const params = { + config: parse_agent_config({ providers: [{ kind: "openai_compat", name: "m", model: "x" }] }), + handle_message: async () => "", + get_agent: (): Agent => { + throw new Error("not used"); + }, + reply_router: () => undefined, + }; + vi.spyOn(logger, "warn").mockImplementation(() => undefined); + try { + expect(create_telegram_adapter(params).capabilities).toEqual({ kind: "text", max_reply_chars: 4096 }); + expect(create_discord_adapter(params).capabilities).toEqual({ kind: "text", max_reply_chars: 2000 }); + expect(create_twitch_adapter(params).capabilities).toEqual({ kind: "text", max_reply_chars: TWITCH_MESSAGE_CAP }); + expect(create_webhook_adapter({ ...params, port: 0, host: "127.0.0.1" }).capabilities).toEqual({ kind: "text" }); + } finally { + vi.restoreAllMocks(); + process.env = saved; + } + }); }); describe("webhook adapter", () => { diff --git a/test/gateway_shutdown.test.ts b/test/gateway_shutdown.test.ts index 2802e43..a6a835f 100644 --- a/test/gateway_shutdown.test.ts +++ b/test/gateway_shutdown.test.ts @@ -22,11 +22,13 @@ describe("shutdown_gateway", () => { const { agent, bus } = tracked(order); const slow: PlatformAdapter = { name: "slow", + capabilities: { kind: "text" }, start: async () => undefined, stop: () => new Promise((resolve) => setTimeout(() => resolve(order.push("slow.stop") as unknown as void), 20)), }; const broken: PlatformAdapter = { name: "broken", + capabilities: { kind: "text" }, start: async () => undefined, stop: async () => { throw new Error("socket gone"); @@ -41,7 +43,7 @@ describe("shutdown_gateway", () => { it("stops waiting after the timeout when an adapter hangs", async () => { const order: string[] = []; const { agent, bus } = tracked(order); - const hung: PlatformAdapter = { name: "hung", start: async () => undefined, stop: () => new Promise(() => undefined) }; + const hung: PlatformAdapter = { name: "hung", capabilities: { kind: "text" }, start: async () => undefined, stop: () => new Promise(() => undefined) }; const warn = vi.spyOn(logger, "warn").mockImplementation(() => undefined); await shutdown_gateway(agent, bus, [hung], 20); expect(order).toEqual(["bus.stop", "agent.close"]);