diff --git a/.gitignore b/.gitignore index 8e0a234..84b1838 100644 --- a/.gitignore +++ b/.gitignore @@ -51,4 +51,8 @@ config/bots/DrElaraVoss.yaml config/bots/Archimedes.yaml tools/strangesonnet45/config/bots/strangesonnet45.yaml config/bots/strangesonnet45.yaml -tools/strangesonnet45/* \ No newline at end of file +tools/strangesonnet45/* +# portal local test instance + scratch +.k27-tmp/ +portal-creds.json +verify.mjs diff --git a/package-lock.json b/package-lock.json index 5b2f55e..e746e40 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9,7 +9,9 @@ "version": "0.1.0", "license": "MIT", "dependencies": { - "@animalabs/membrane": "^0.5.47", + "@animalabs/membrane": "^0.5.67", + "@animalabs/portal-client": "^0.2.1", + "@animalabs/portal-protocol": "^0.2.0", "@anthropic-ai/sdk": "^0.32.1", "@aws-sdk/client-bedrock-runtime": "^3.709.0", "@google/generative-ai": "^0.21.0", @@ -41,9 +43,9 @@ } }, "node_modules/@animalabs/membrane": { - "version": "0.5.47", - "resolved": "https://registry.npmjs.org/@animalabs/membrane/-/membrane-0.5.47.tgz", - "integrity": "sha512-KCoBF2D3hIJ6YHgQmCrlmOqTNENdGV2Ip2DiDcSImyDQ+3HfUbgyEI06EMbmL3BEHN20pSVcqYAzyPoDKeTWOQ==", + "version": "0.5.68", + "resolved": "https://registry.npmjs.org/@animalabs/membrane/-/membrane-0.5.68.tgz", + "integrity": "sha512-MfSERy0ECxdBZ4NelHdZbqsP/ORWJsfeAI6vuyy6gP77aGcdJL8qf4XrUkOox5yZa5MqB50y0K58Ffv79jCnbg==", "license": "MIT", "dependencies": { "@anthropic-ai/sdk": "^0.52.0" @@ -61,6 +63,26 @@ "anthropic-ai-sdk": "bin/cli" } }, + "node_modules/@animalabs/portal-client": { + "version": "0.2.1", + "resolved": "https://registry.npmjs.org/@animalabs/portal-client/-/portal-client-0.2.1.tgz", + "integrity": "sha512-9plhxES8mwV6RtM5SS3B7kvDLHwWGNT4nBr2XLJSX4cO2JX6ZXg3GE8MARonRTteWTAL1vQNplV9SQ+1NVyn5g==", + "dependencies": { + "@animalabs/portal-protocol": "^0.2.0", + "ws": "^8.18.0" + }, + "engines": { + "node": ">=20.0.0" + } + }, + "node_modules/@animalabs/portal-protocol": { + "version": "0.2.0", + "resolved": "https://registry.npmjs.org/@animalabs/portal-protocol/-/portal-protocol-0.2.0.tgz", + "integrity": "sha512-TbuwvdUmDwWyp3oxojSM81O+l8F3IbjjAcGLKmcgx1CY3/8x0j1ypNYaK0iUQ9oinJQuRPARsVom0s4k4G9T7g==", + "engines": { + "node": ">=20.0.0" + } + }, "node_modules/@anthropic-ai/sdk": { "version": "0.32.1", "resolved": "https://registry.npmjs.org/@anthropic-ai/sdk/-/sdk-0.32.1.tgz", diff --git a/package.json b/package.json index a426118..e26bf97 100644 --- a/package.json +++ b/package.json @@ -23,7 +23,9 @@ "author": "", "license": "MIT", "dependencies": { - "@animalabs/membrane": "^0.5.47", + "@animalabs/membrane": "^0.5.67", + "@animalabs/portal-client": "^0.2.1", + "@animalabs/portal-protocol": "^0.2.0", "@anthropic-ai/sdk": "^0.32.1", "@aws-sdk/client-bedrock-runtime": "^3.709.0", "@google/generative-ai": "^0.21.0", diff --git a/src/agent/loop.ts b/src/agent/loop.ts index 7b6e978..3cc47f0 100644 --- a/src/agent/loop.ts +++ b/src/agent/loop.ts @@ -6,7 +6,7 @@ import { EventQueue } from './event-queue.js' import { DeferredQueue, isTransientError } from './deferred-queue.js' import { ChannelStateManager } from './state-manager.js' -import { DiscordConnector, type PinnedSteer } from '../discord/connector.js' +import type { IConnector, PinnedSteer } from '../connector/types.js' import { ConfigSystem } from '../config/system.js' import { ContextBuilder, BuildContextParams } from '../context/builder.js' import { ToolSystem } from '../tools/system.js' @@ -97,7 +97,7 @@ export class AgentLoop { constructor( private botId: string, private queue: EventQueue, - private connector: DiscordConnector, + private connector: IConnector, private stateManager: ChannelStateManager, private configSystem: ConfigSystem, private contextBuilder: ContextBuilder, @@ -793,7 +793,12 @@ export class AgentLoop { if ((event.data as any)._isMCommand) { reason = 'm_command' - } else if (message.reference?.messageId && this.botMessageIds.has(message.reference.messageId)) { + } else if ( + (message.reference?.messageId && this.botMessageIds.has(message.reference.messageId)) || + // Portal: the relay's 'reply' address reason is authoritative across restarts + // (botMessageIds only knows this session's sends). + message._address?.reasons?.includes('reply') + ) { reason = 'reply' } else if (this.botUserId && message.mentions?.has(this.botUserId)) { reason = 'mention' @@ -874,7 +879,10 @@ export class AgentLoop { userId: triggeringUser.id, serverId: guildId, channelId: channelId, - botId: this.botUserId || '', + // Portal personas have no Discord user id; key cost by the EMS bot name + // (this.botId) so admin cost commands (which resolve a portal role to the + // EMS name) target the same key. Account bots stay keyed by Discord user id. + botId: this.connector.isPortal() ? this.botId : (this.botUserId || ''), messageId: triggeringMessageId || '', triggerType: triggerReason as SomaTriggerType, userRoles: triggeringUser.roles || [], @@ -1401,6 +1409,33 @@ export class AgentLoop { // A reply with "ping on reply" enabled includes the bot in message.mentions and is // already handled by the mention check above. A reply with the ping toggled off must // not wake the bot — fall through to the remaining triggers (random / name / etc.). + // + // EXCEPTION — portal backend: webhook personas can't carry a native reply/ping, so a + // reply-ping never lands in message.mentions. The relay instead surfaces it as the + // 'reply' address reason (computed per-recipient in portal-relay reasonsFor()). It + // cannot encode the ping toggle for personas, so any reply to one of our messages + // counts — this is the portal equivalent of the account-bot reply-ping above, and + // restores pre-hard-rule behavior for portal only. Account/discord.js events carry + // no _address, so they remain governed by the hard rule. + const portalReply = this.connector.isPortal() && message._address?.reasons?.includes('reply') + if (portalReply) { + const chainDepth = await this.connector.getBotReplyChainDepth(channelId, message) + + if (!loadConfig()) return false + + if (chainDepth >= config.max_bot_reply_chain_depth) { + logger.info({ + messageId: message.id, + chainDepth, + limit: config.max_bot_reply_chain_depth + }, 'Bot reply chain depth limit reached, blocking portal reply activation') + await this.connector.addReaction(channelId, message.id, config.bot_reply_chain_depth_emote) + continue + } + + logger.debug({ messageId: message.id, chainDepth }, 'Activated by portal reply') + return true + } // 4. Random chance activation if (!loadConfig()) return false diff --git a/src/api/server.ts b/src/api/server.ts index 97f0e68..fd0c46f 100644 --- a/src/api/server.ts +++ b/src/api/server.ts @@ -4,7 +4,7 @@ */ import express, { Request, Response, NextFunction } from 'express' -import { DiscordConnector } from '../discord/connector.js' +import type { IConnector } from '../connector/types.js' import { ConfigSystem } from '../config/system.js' import { ContextBuilder } from '../context/builder.js' import { logger } from '../utils/logger.js' @@ -68,7 +68,7 @@ export class ApiServer { constructor( private config: ApiConfig, - private connector: DiscordConnector, + private connector: IConnector, private configSystem?: ConfigSystem, private contextBuilder?: ContextBuilder, private botName?: string @@ -259,6 +259,10 @@ export class ApiServer { } const client = (this.connector as any).client + if (!client) { + res.status(501).json({ error: 'Not Implemented', message: 'channel listing requires the discord backend' }) + return + } const guilds = guildIdFilter ? [client.guilds.cache.get(guildIdFilter)].filter(Boolean) : Array.from(client.guilds.cache.values()) @@ -578,7 +582,8 @@ export class ApiServer { private async getUserInfo(userId: string, guildId?: string): Promise { const client = (this.connector as any).client - + if (!client) throw new Error('user lookup requires the discord backend (portal mode)') + // Fetch user from Discord let user try { @@ -629,7 +634,8 @@ export class ApiServer { private async getUserAvatar(userId: string, size: number = 128): Promise { const client = (this.connector as any).client - + if (!client) throw new Error('avatar lookup requires the discord backend (portal mode)') + try { const user = await client.users.fetch(userId) if (!user) { diff --git a/src/connector/portal/adapters.test.ts b/src/connector/portal/adapters.test.ts new file mode 100644 index 0000000..1384cd8 --- /dev/null +++ b/src/connector/portal/adapters.test.ts @@ -0,0 +1,190 @@ +import { describe, it, expect } from 'vitest' +import type { PortalMessage, PortalEvent } from '@animalabs/portal-protocol' +import { inboundFromPortal, queueEventFromPortal, authorOfMessage, containerOf, type AdapterCtx } from './adapters.js' + +const BOT = 'persona_bot' + +function pm(over: Partial = {}): PortalMessage { + return { + id: 'rm_chan_100', + nativeId: '100', + channelId: 'chan', + guildId: 'guild', + author: { kind: 'user', userId: 'u1', username: 'alice', displayName: 'Alice', bot: false }, + content: 'hello', + cleanContent: 'hello', + attachments: [], + mentions: { personas: [], roles: [], users: [], everyone: false }, + reactions: [], + createdAt: '2026-06-14T00:00:00.000Z', + ...over, + } +} + +const ctx: AdapterCtx = { + botPersonaId: BOT, + authorOf: (id) => (id === 'rm_chan_100' ? 'u1' : undefined), + guildOf: () => 'guild', +} + +describe('authorOfMessage', () => { + it('maps persona/user/system kinds', () => { + expect(authorOfMessage(pm({ author: { kind: 'persona', personaId: 'p9', displayName: 'Mythos', avatarUrl: '' } }))).toEqual({ + id: 'p9', + username: 'Mythos', + bot: true, + globalName: 'Mythos', + }) + expect(authorOfMessage(pm()).id).toBe('u1') + expect(authorOfMessage(pm()).bot).toBe(false) + expect(authorOfMessage(pm({ author: { kind: 'system' } }))).toEqual({ id: 'system', username: 'system', bot: true }) + }) +}) + +describe('containerOf', () => { + it('prefers threadId', () => { + expect(containerOf({ channelId: 'c', threadId: 't' })).toBe('t') + expect(containerOf({ channelId: 'c' })).toBe('c') + }) +}) + +describe('inboundFromPortal', () => { + it('builds a discord.js-ish object', () => { + const m = inboundFromPortal(pm(), BOT) + expect(m.id).toBe('rm_chan_100') + expect(m.channelId).toBe('chan') + expect(m.author.id).toBe('u1') + expect(m.system).toBe(false) + expect(m.reference).toBeUndefined() + }) + + it('mentions.has() is true for a persona role-mention (→ bot mentioned)', () => { + const m = inboundFromPortal(pm({ mentions: { personas: [BOT], roles: [], users: [], everyone: false } }), BOT) + expect(m.mentions.has(BOT)).toBe(true) + expect(m.mentions.has('someone-else')).toBe(false) + }) + + it('mentions.has() is true for a direct user mention', () => { + const m = inboundFromPortal(pm({ mentions: { personas: [], roles: [], users: ['u42'], everyone: false } }), BOT) + expect(m.mentions.has('u42')).toBe(true) + }) + + it('reply maps replyToId → reference.messageId', () => { + const m = inboundFromPortal(pm({ replyToId: 'rm_chan_50' }), BOT) + expect(m.reference?.messageId).toBe('rm_chan_50') + }) + + it('threaded message uses threadId as channelId', () => { + const m = inboundFromPortal(pm({ threadId: 'thread1' }), BOT) + expect(m.channelId).toBe('thread1') + }) + + it('reactions.cache.some sees the eye emoji by name', () => { + const m = inboundFromPortal(pm({ reactions: [{ emoji: '👁‍🗨', count: 1, kind: 'native', by: [] }] }), BOT) + const hit = m.reactions!.cache.some((r: any) => r.emoji?.name === '👁‍🗨') + expect(hit).toBe(true) + }) + + it('system author sets system flag', () => { + const m = inboundFromPortal(pm({ author: { kind: 'system' } }), BOT) + expect(m.system).toBe(true) + }) + + it('event.data is mutable (loop sets _isMCommand)', () => { + const m = inboundFromPortal(pm(), BOT) as any + m._isMCommand = true + expect(m._isMCommand).toBe(true) + }) +}) + +describe('queueEventFromPortal', () => { + it('message_create → message event', () => { + const e: PortalEvent = { type: 'message_create', message: pm(), addressedToMe: false, reasons: [] } + const ev = queueEventFromPortal(e, ctx)! + expect(ev.type).toBe('message') + expect(ev.channelId).toBe('chan') + expect(ev.guildId).toBe('guild') + expect(ev.data.id).toBe('rm_chan_100') + expect(ev.timestamp instanceof Date).toBe(true) + }) + + it('message_update → edit event with {old:undefined,new}', () => { + const e: PortalEvent = { type: 'message_update', message: pm({ content: 'edited' }), addressedToMe: false, reasons: [] } + const ev = queueEventFromPortal(e, ctx)! + expect(ev.type).toBe('edit') + expect(ev.data.old).toBeUndefined() + expect(ev.data.new.content).toBe('edited') + }) + + it('message_delete → delete event, resolves author from ctx', () => { + const e: PortalEvent = { type: 'message_delete', channelId: 'chan', messageId: 'rm_chan_100' } + const ev = queueEventFromPortal(e, ctx)! + expect(ev.type).toBe('delete') + expect(ev.data.id).toBe('rm_chan_100') + expect(ev.data.author.id).toBe('u1') + }) + + it('message_delete → author "unknown" when unresolved', () => { + const e: PortalEvent = { type: 'message_delete', channelId: 'chan', messageId: 'rm_chan_999' } + const ev = queueEventFromPortal(e, ctx)! + expect(ev.data.author.id).toBe('unknown') + }) + + it('reaction_add → reaction event with reactor + author', () => { + const e: PortalEvent = { + type: 'reaction_add', + channelId: 'chan', + messageId: 'rm_chan_100', + reaction: { emoji: '✅', count: 1, kind: 'native', by: [{ kind: 'user', id: 'u7', name: 'bob' }] }, + } + const ev = queueEventFromPortal(e, ctx)! + expect(ev.type).toBe('reaction') + expect(ev.data.messageId).toBe('rm_chan_100') + expect(ev.data.emoji).toBe('✅') + expect(ev.data.userId).toBe('u7') + expect(ev.data.messageAuthorId).toBe('u1') + }) + + it('reaction_add with no known reactor → null (unattributable)', () => { + const e: PortalEvent = { + type: 'reaction_add', + channelId: 'chan', + messageId: 'rm_chan_100', + reaction: { emoji: '✅', count: 1, kind: 'native', by: [] }, + } + expect(queueEventFromPortal(e, ctx)).toBeNull() + }) + + it('reaction_remove / typing / structural events → null', () => { + expect(queueEventFromPortal({ type: 'reaction_remove', channelId: 'c', messageId: 'm', emoji: 'x', actor: { kind: 'user', id: 'u', name: 'n' } }, ctx)).toBeNull() + expect(queueEventFromPortal({ type: 'typing', channelId: 'c', author: { kind: 'system' } }, ctx)).toBeNull() + expect(queueEventFromPortal({ type: 'pins_update', channelId: 'c' }, ctx)).toBeNull() + }) + + it('thread message_create uses threadId as event channelId', () => { + const e: PortalEvent = { type: 'message_create', message: pm({ threadId: 'thr' }), addressedToMe: false, reasons: [] } + const ev = queueEventFromPortal(e, ctx)! + expect(ev.channelId).toBe('thr') + }) + + it('propagates relay address reasons onto inbound._address (reply-ping signal)', () => { + // A reply-ping to a webhook persona never lands in mentions, so the loop's + // portal reply-activation depends on this 'reply' reason being carried through. + const e: PortalEvent = { + type: 'message_create', + message: pm({ replyToId: 'rm_chan_50' }), + addressedToMe: true, + reasons: ['reply'], + } + const ev = queueEventFromPortal(e, ctx)! + expect(ev.data._address).toEqual({ addressedToMe: true, reasons: ['reply'] }) + expect(ev.data._address.reasons.includes('reply')).toBe(true) + }) + + it('ambient (unaddressed) message carries empty reasons', () => { + const e: PortalEvent = { type: 'message_create', message: pm(), addressedToMe: false, reasons: ['subscription'] } + const ev = queueEventFromPortal(e, ctx)! + expect(ev.data._address.addressedToMe).toBe(false) + expect(ev.data._address.reasons.includes('reply')).toBe(false) + }) +}) diff --git a/src/connector/portal/adapters.ts b/src/connector/portal/adapters.ts new file mode 100644 index 0000000..ea8bb70 --- /dev/null +++ b/src/connector/portal/adapters.ts @@ -0,0 +1,150 @@ +/** + * Pure adapters: PortalMessage / PortalEvent → chatperx's internal shapes. + * + * These are the unit-test target. No I/O, no portal-client — just shape + * translation. The agent loop reads discord.js-`Message`-compatible objects off + * `event.data` (see connector/types.ts `InboundMessage`), so we synthesize that + * duck type faithfully. + */ +import type { PortalMessage, PortalEvent } from '@animalabs/portal-protocol' +import type { Event } from '../../types.js' +import type { InboundMessage } from '../types.js' + +/** Context the event adapter needs that isn't on the event itself. */ +export interface AdapterCtx { + botPersonaId: string + /** relayId → author id (persona/user) of a previously-seen message. */ + authorOf(messageId: string): string | undefined + /** channelId (or threadId) → guild id, from the client cache. */ + guildOf(channelId: string): string +} + +/** The immediate channel a message lives in (thread when threaded). */ +export function containerOf(pm: { channelId: string; threadId?: string }): string { + return pm.threadId ?? pm.channelId +} + +/** Resolve a PortalMessage author to a discord.js-ish `{id,username,bot}`. */ +export function authorOfMessage(pm: PortalMessage): { + id: string + username: string + bot: boolean + globalName?: string +} { + const a = pm.author + if (a.kind === 'persona') return { id: a.personaId, username: a.displayName, bot: true, globalName: a.displayName } + if (a.kind === 'user') return { id: a.userId, username: a.username, bot: a.bot, globalName: a.displayName } + return { id: 'system', username: 'system', bot: true } +} + +/** + * PortalMessage → the discord.js-`Message`-compatible object the loop reads off + * `event.data`. `mentions.has(id)` translates a persona role-mention into a + * "bot mentioned" signal (botUserId === personaId). `address` carries the relay's + * per-recipient addressing reasons (role_mention / reply / subscription); a + * reply-ping can only be detected via `reasons`, since webhook personas never + * appear in `mentions`. + */ +export function inboundFromPortal( + pm: PortalMessage, + _botPersonaId: string, + address?: { addressedToMe: boolean; reasons: string[] }, +): InboundMessage { + const author = authorOfMessage(pm) + const mentionedIds = new Set([...pm.mentions.personas, ...pm.mentions.users]) + const users = new Map(pm.mentions.users.map((id) => [id, { id }])) + return { + id: pm.id, + channelId: containerOf(pm), + guildId: pm.guildId ?? '', + content: pm.content, + author, + system: pm.author.kind === 'system', + reference: pm.replyToId ? { messageId: pm.replyToId } : undefined, + _address: address, + mentions: { + has: (id: string) => mentionedIds.has(id), + users, + }, + // Back the show-reaction override with the message's real reactions. The + // loop calls `r.emoji?.name` — Portal carries the unicode/`name:id` string, + // so wrap each as `{ emoji: { name } }`. + reactions: { + cache: { + some: (fn: (r: unknown) => boolean) => pm.reactions.some((r) => fn({ emoji: { name: r.emoji } })), + }, + }, + // member.roles unavailable inbound (P1); role-gated user features degrade. + member: undefined, + _portal: pm, + } +} + +/** A PortalEvent → an EventQueue `Event`, or null when it shouldn't enqueue. */ +export function queueEventFromPortal(e: PortalEvent, ctx: AdapterCtx): Event | null { + switch (e.type) { + case 'message_create': { + const inbound = inboundFromPortal(e.message, ctx.botPersonaId, { + addressedToMe: e.addressedToMe, + reasons: e.reasons, + }) + return { + type: 'message', + channelId: inbound.channelId, + guildId: inbound.guildId, + data: inbound, + timestamp: new Date(e.message.createdAt), + receivedAt: Date.now(), + } + } + case 'message_update': { + const inbound = inboundFromPortal(e.message, ctx.botPersonaId, { + addressedToMe: e.addressedToMe, + reasons: e.reasons, + }) + return { + type: 'edit', + channelId: inbound.channelId, + guildId: inbound.guildId, + data: { old: undefined, new: inbound }, + timestamp: new Date(e.message.editedAt ?? e.message.createdAt), + } + } + case 'message_delete': { + const channelId = e.threadId ?? e.channelId + return { + type: 'delete', + channelId, + guildId: ctx.guildOf(channelId), + data: { id: e.messageId, author: { id: ctx.authorOf(e.messageId) ?? 'unknown' } }, + timestamp: new Date(), + } + } + case 'reaction_add': { + const channelId = e.threadId ?? e.channelId + const reactorId = e.reaction.by[0]?.id + // Can't attribute a reaction with no known reactor — drop it. Emitting + // userId:undefined would bypass the loop's skip-own-reaction check + // (userId === botUserId is false for undefined), letting the bot react to + // its own reaction / process an unattributable trigger. + if (!reactorId) return null + return { + type: 'reaction', + channelId, + guildId: ctx.guildOf(channelId), + data: { + messageId: e.messageId, + emoji: e.reaction.emoji, + userId: reactorId, + messageAuthorId: ctx.authorOf(e.messageId), + }, + timestamp: new Date(), + receivedAt: Date.now(), + } + } + // reaction_remove is observability-only in the discord backend (no enqueue); + // typing/pins/structural/persona events are handled by the cache, not the loop. + default: + return null + } +} diff --git a/src/connector/portal/context.test.ts b/src/connector/portal/context.test.ts new file mode 100644 index 0000000..765312c --- /dev/null +++ b/src/connector/portal/context.test.ts @@ -0,0 +1,47 @@ +import { describe, it, expect } from 'vitest' +import type { PortalMessage } from '@animalabs/portal-protocol' +import { portalMessageToDiscordMessage } from './context.js' + +function pm(over: Partial = {}): PortalMessage { + return { + id: 'rm_chan_100', + nativeId: '100', + channelId: 'chan', + guildId: 'guild', + author: { kind: 'user', userId: 'u1', username: 'alice', displayName: 'Alice', bot: false }, + content: 'hi', + cleanContent: 'hi', + attachments: [], + mentions: { personas: [], roles: [], users: [], everyone: false }, + reactions: [], + createdAt: '2026-06-14T00:00:00.000Z', + ...over, + } +} +const noMap = new Map() + +describe('portalMessageToDiscordMessage — strip portal- role prefix', () => { + it('strips the portal- prefix from a resolved role mention', () => { + const d = portalMessageToDiscordMessage(pm({ cleanContent: '@portal-glm52 hey' }), noMap, 'bot') + expect(d.content).toBe('@glm52 hey') + }) + + it('strips every portal- mention and preserves the rest', () => { + const d = portalMessageToDiscordMessage( + pm({ cleanContent: 'ping @portal-nemotron-3-ultra and @portal-Fugu, cc @alice' }), + noMap, + 'bot', + ) + expect(d.content).toBe('ping @nemotron-3-ultra and @Fugu, cc @alice') + }) + + it('leaves ordinary (non-portal) mentions untouched', () => { + const d = portalMessageToDiscordMessage(pm({ cleanContent: 'hey @alice and @bob' }), noMap, 'bot') + expect(d.content).toBe('hey @alice and @bob') + }) + + it('is a no-op on raw <@&id> content (no cleanContent)', () => { + const d = portalMessageToDiscordMessage(pm({ cleanContent: '', content: '<@&123> hi' }), noMap, 'bot') + expect(d.content).toBe('<@&123> hi') + }) +}) diff --git a/src/connector/portal/context.ts b/src/connector/portal/context.ts new file mode 100644 index 0000000..ec7caf8 --- /dev/null +++ b/src/connector/portal/context.ts @@ -0,0 +1,212 @@ +/** + * Portal fetchContext assembly + PortalMessage → DiscordMessage conversion. + * + * Mirrors DiscordConnector.fetchContext's stages — (1) .history-resolved fetch, + * (2) thread-parent assembly, (3) cache-anchor stability, (4) attachments — but + * on PortalMessage, using `nativeId` (snowflake) for all ordering/anchor math + * while message identity stays the relay id. + */ +import type { PortalMessage } from '@animalabs/portal-protocol' +import type { DiscordContext, DiscordMessage } from '../../types.js' +import type { FetchContextParams } from '../types.js' +import { logger } from '../../utils/logger.js' +import { authorOfMessage, containerOf } from './adapters.js' +import { fetchPortalHistory, snowflakeFromId, type PortalFetchDeps } from './history.js' +import type { AttachmentCache } from '../util/attachment-cache.js' + +const DISCORD_EPOCH = 1420070400000 +function snowflakeToTimestamp(snowflake: string): number { + try { + return Number(BigInt(snowflake) >> 22n) + DISCORD_EPOCH + } catch { + return 0 + } +} +function snowflakeLT(a: string, b: string): boolean { + try { + return BigInt(a) < BigInt(b) + } catch { + return a < b + } +} + +/** The relay addresses personas via pooled roles named `portal-`, so a + * resolved role mention in `cleanContent` reads `@portal-glm52`. Strip that + * internal prefix so the model sees `@glm52` — the addressing plumbing is noise + * to it. Text-only rewrite of the readable rendering; a no-op on raw `<@&id>`. */ +function stripPortalRolePrefix(text: string): string { + return text.replaceAll('@portal-', '@') +} + +/** Full PortalMessage → DiscordMessage. Uses portal's resolved `cleanContent`; + * adds the `` prefix for non-bot authors (matching convertMessage). */ +export function portalMessageToDiscordMessage( + pm: PortalMessage, + messageMap: Map, + _botPersonaId: string, +): DiscordMessage { + const a = authorOfMessage(pm) + let content = stripPortalRolePrefix(pm.cleanContent || pm.content) + + if (pm.replyToId && !a.bot) { + const replied = messageMap.get(pm.replyToId) + const name = replied ? authorOfMessage(replied).username : 'someone' + content = ` ${content}` + } + + return { + id: pm.id, + channelId: containerOf(pm), + guildId: pm.guildId ?? '', + author: { id: a.id, username: a.username, displayName: a.globalName ?? a.username, bot: a.bot }, + content, + timestamp: new Date(pm.createdAt), + attachments: pm.attachments.map((att) => ({ + id: att.id, + url: att.url, + filename: att.name, + contentType: att.contentType ?? undefined, + size: att.size, + })), + reactions: pm.reactions.map((r) => ({ emoji: r.emoji, count: r.count })), + mentions: pm.mentions.users, + referencedMessage: pm.replyToId, + } +} + +export interface ContextDeps extends PortalFetchDeps { + attachments: AttachmentCache + /** Channel meta from the client cache. */ + channelMeta(channelId: string): { guildId: string; isThread: boolean; parentId?: string; native?: string } | null +} + +const MAX_ANCHOR_GAP_MS = 24 * 60 * 60 * 1000 +const MAX_ANCHOR_EXTEND = 500 + +/** Assemble a DiscordContext from portal history + attachments. */ +export async function buildPortalContext(params: FetchContextParams, deps: ContextDeps): Promise { + const { channelId, depth, targetMessageId, firstMessageId, authorized_roles, maxImages, ignoreHistory } = params + const meta = deps.channelMeta(channelId) + + // ── Stage 1: channel messages with .history ── + const result = await fetchPortalHistory( + channelId, + targetMessageId ? snowflakeFromId(targetMessageId) : undefined, + undefined, + depth, + authorized_roles ?? [], + ignoreHistory ?? false, + deps, + ) + let messages = result.messages + const historyDidClear = result.didClear + let historyOriginChannelId = result.originChannelId + + // ── Stage 2: thread-parent assembly ── + if (meta?.isThread && !historyDidClear && meta.parentId) { + const starterSnowflake = meta.native // thread id === starter message snowflake + const parentResult = await fetchPortalHistory( + meta.parentId, + starterSnowflake, + undefined, + Math.max(0, depth - messages.length), + authorized_roles ?? [], + ignoreHistory ?? false, + deps, + ) + const parentMessages = parentResult.messages + if (starterSnowflake && !parentMessages.some((m) => m.nativeId === starterSnowflake)) { + const starter = await deps.fetchSingle(meta.parentId, starterSnowflake) + if (starter) parentMessages.push(starter) + } + messages = [...parentMessages, ...messages] + if (parentResult.originChannelId && !historyOriginChannelId) historyOriginChannelId = parentResult.originChannelId + } + + // ── Stage 3: cache-anchor stability (snowflake-based) ── + let cacheAnchorTrimmed = false + let anchor = firstMessageId + if (anchor && !historyDidClear) { + const anchorSnow = snowflakeFromId(anchor) + let firstIndex = messages.findIndex((m) => m.id === anchor || m.nativeId === anchorSnow) + const oldest = messages[0] + + // Temporal sanity: skip extension if the anchor is far older than the window. + if (firstIndex < 0 && oldest) { + const gap = snowflakeToTimestamp(oldest.nativeId) - snowflakeToTimestamp(anchorSnow) + if (gap > MAX_ANCHOR_GAP_MS) { + logger.warn({ anchor, gapHours: Math.round(gap / 3.6e6) }, 'portal: cache anchor too old — skipping extension') + anchor = undefined + } + } + + if (firstIndex < 0 && oldest && anchor) { + const extendChannel = meta?.isThread && meta.parentId && oldest.nativeId === meta.native ? meta.parentId : channelId + let extended = 0 + let before = oldest.nativeId + while (extended < MAX_ANCHOR_EXTEND) { + const batch = await deps.fetchBatch(extendChannel, { before, limit: 100 }) + if (batch.length === 0) break + const sorted = [...batch].sort((a, b) => (snowflakeLT(a.nativeId, b.nativeId) ? -1 : 1)) + messages = [...sorted, ...messages] + extended += sorted.length + firstIndex = messages.findIndex((m) => m.id === anchor || m.nativeId === anchorSnow) + if (firstIndex >= 0) break + before = sorted[0]!.nativeId + } + } + + const historyWasUsed = !!historyOriginChannelId + if (firstIndex > 0 && !historyWasUsed) { + messages = messages.slice(firstIndex) + cacheAnchorTrimmed = true + } + } + + // ── Convert + filter ── + const messageMap = new Map(messages.map((m) => [m.id, m])) + const discordMessages = messages + .map((m) => portalMessageToDiscordMessage(m, messageMap, deps.botPersonaId)) + .filter((m) => m.content || m.attachments.length > 0) + + // ── Stage 4: attachments (newest-first so the image cap keeps recent ones) ── + const images = [] + const documents = [] + let newDownloads = 0 + const capReached = () => maxImages !== undefined && images.length >= maxImages + for (let i = messages.length - 1; i >= 0; i--) { + for (const att of messages[i]!.attachments) { + if (att.contentType?.startsWith('image/')) { + if (capReached()) continue + const had = deps.attachments.has(att.url) + const cached = await deps.attachments.cacheImage(att.url, att.contentType) + if (cached) { + images.push(cached) + if (!had) newDownloads++ + } + } else if (deps.attachments.isTextAttachment({ url: att.url, name: att.name, contentType: att.contentType, size: att.size })) { + const doc = await deps.attachments.fetchTextAttachment( + { url: att.url, name: att.name, contentType: att.contentType, size: att.size }, + messages[i]!.id, + ) + if (doc) documents.push(doc) + } + } + } + if (newDownloads > 0) deps.attachments.saveUrlMap() + + const inheritanceInfo: DiscordContext['inheritanceInfo'] = {} + if (meta?.isThread && meta.parentId) inheritanceInfo.parentChannelId = meta.parentId + if (historyOriginChannelId) inheritanceInfo.historyOriginChannelId = historyOriginChannelId + if (historyDidClear) inheritanceInfo.historyDidClear = true + if (cacheAnchorTrimmed) inheritanceInfo.cacheAnchorTrimmed = true + + return { + messages: discordMessages, + pinnedConfigs: params.pinnedConfigs ?? [], + images, + documents, + guildId: meta?.guildId ?? '', + inheritanceInfo: Object.keys(inheritanceInfo).length > 0 ? inheritanceInfo : undefined, + } +} diff --git a/src/connector/portal/history.test.ts b/src/connector/portal/history.test.ts new file mode 100644 index 0000000..fd09031 --- /dev/null +++ b/src/connector/portal/history.test.ts @@ -0,0 +1,178 @@ +import { describe, it, expect } from 'vitest' +import type { PortalMessage } from '@animalabs/portal-protocol' +import { fetchPortalHistory, snowflakeFromId, type PortalFetchDeps, type PortalChannelRef } from './history.js' + +/** Build a PortalMessage with snowflake `nativeId` and relay id `rm__`. */ +function msg(chan: string, sf: string, content = 'hi', over: Partial = {}): PortalMessage { + return { + id: `rm_${chan}_${sf}`, + nativeId: sf, + channelId: chan, + guildId: 'g', + author: { kind: 'user', userId: 'u', username: 'alice', displayName: 'Alice', bot: false }, + content, + cleanContent: content, + attachments: [], + mentions: { personas: [], roles: [], users: [], everyone: false }, + reactions: [], + createdAt: '2026-06-14T00:00:00.000Z', + ...over, + } +} + +/** In-memory deps: channels → messages (any order; sorted internally). */ +function makeDeps( + channels: Record, + opts: { natives?: Record; botPersonaId?: string } = {}, +): PortalFetchDeps { + const sf = (m: PortalMessage) => BigInt(m.nativeId) + return { + botPersonaId: opts.botPersonaId ?? 'bot', + async fetchBatch(channelId, { before, limit }) { + const all = [...(channels[channelId] ?? [])].sort((a, b) => (sf(a) > sf(b) ? -1 : 1)) // newest-first + const filtered = before ? all.filter((m) => sf(m) < BigInt(before)) : all + return filtered.slice(0, limit) + }, + async fetchSingle(channelId, snowflake) { + return (channels[channelId] ?? []).find((m) => m.nativeId === snowflake) ?? null + }, + async resolveChannelBySnowflake(snowflake): Promise { + const natives = opts.natives ?? {} + const chan = Object.keys(natives).find((c) => natives[c] === snowflake) + return chan ? { id: chan, isTextBased: true } : null + }, + async authorRoles() { + return null + }, + } +} + +const ids = (r: { messages: PortalMessage[] }) => r.messages.map((m) => m.nativeId) + +describe('snowflakeFromId', () => { + it('extracts snowflake from a relay id and passes through raw snowflakes', () => { + expect(snowflakeFromId('rm_chan_12345')).toBe('12345') + expect(snowflakeFromId('98765')).toBe('98765') + expect(snowflakeFromId('rm_thread99_777')).toBe('777') + }) +}) + +describe('fetchPortalHistory', () => { + it('returns chronological order (oldest first)', async () => { + const deps = makeDeps({ c: [msg('c', '100'), msg('c', '200'), msg('c', '300')] }) + const r = await fetchPortalHistory('c', undefined, undefined, 50, [], false, deps) + expect(ids(r)).toEqual(['100', '200', '300']) + expect(r.didClear).toBe(false) + }) + + it('respects the maxMessages cap (keeps newest)', async () => { + const deps = makeDeps({ c: ['100', '200', '300', '400'].map((s) => msg('c', s)) }) + const r = await fetchPortalHistory('c', undefined, undefined, 2, [], false, deps) + expect(ids(r)).toEqual(['300', '400']) + }) + + it('drops relay system messages', async () => { + const deps = makeDeps({ + c: [msg('c', '100'), msg('c', '200', '', { author: { kind: 'system' } }), msg('c', '300')], + }) + const r = await fetchPortalHistory('c', undefined, undefined, 50, [], false, deps) + expect(ids(r)).toEqual(['100', '300']) + }) + + it('.history (bare) clears — returns only newer messages', async () => { + const deps = makeDeps({ + c: [msg('c', '100'), msg('c', '200', '.history'), msg('c', '300'), msg('c', '400')], + }) + const r = await fetchPortalHistory('c', undefined, undefined, 50, [], false, deps) + expect(ids(r)).toEqual(['300', '400']) + expect(r.didClear).toBe(true) + }) + + it('.history range pulls a linked range + newer messages (same channel)', async () => { + // URLs carry the Discord channel SNOWFLAKE (native), resolved back to 'c'. + const last = 'https://discord.com/channels/1/5000/250' + const first = 'https://discord.com/channels/1/5000/150' + const deps = makeDeps( + { + c: [ + msg('c', '100'), + msg('c', '150'), + msg('c', '200'), + msg('c', '250'), + msg('c', '300', `.history\n---\nfirst: ${first}\nlast: ${last}`), + msg('c', '400'), + ], + }, + { natives: { c: '5000' } }, + ) + const r = await fetchPortalHistory('c', undefined, undefined, 50, [], false, deps) + // range [150..250] then messages newer than the .history (400) + expect(ids(r)).toEqual(['150', '200', '250', '400']) + expect(r.didClear).toBe(true) + expect(r.originChannelId).toBe('c') + }) + + it('.history range resolves a cross-channel link by native snowflake', async () => { + const last = 'https://discord.com/channels/1/9999/250' + const deps = makeDeps( + { + cur: [msg('cur', '300', `.history\n---\nlast: ${last}`), msg('cur', '400')], + other: [msg('other', '200'), msg('other', '250'), msg('other', '300')], + }, + { natives: { other: '9999' } }, + ) + const r = await fetchPortalHistory('cur', undefined, undefined, 50, [], false, deps) + // everything up to & including 250 in `other`, then newer-than-.history in cur (400) + expect(ids(r)).toEqual(['200', '250', '400']) + expect(r.originChannelId).toBe('cur') + }) + + it('.history range treats an unresolvable channel as clear', async () => { + const last = 'https://discord.com/channels/1/9999/250' + const deps = makeDeps({ cur: [msg('cur', '300', `.history\n---\nlast: ${last}`), msg('cur', '400')] }) + const r = await fetchPortalHistory('cur', undefined, undefined, 50, [], false, deps) + expect(ids(r)).toEqual(['400']) + expect(r.didClear).toBe(true) + }) + + it('skips a .history targeted at a different persona', async () => { + const deps = makeDeps( + { c: [msg('c', '100'), msg('c', '200', '.history', { mentions: { personas: ['other-bot'], roles: [], users: [], everyone: false } }), msg('c', '300')] }, + { botPersonaId: 'bot' }, + ) + const r = await fetchPortalHistory('c', undefined, undefined, 50, [], false, deps) + // targeting another persona → the .history is ignored, all messages kept + expect(ids(r)).toEqual(['100', '300']) + expect(r.didClear).toBe(false) + }) + + it('applies a .history targeted at us (our persona mention)', async () => { + const deps = makeDeps( + { c: [msg('c', '100'), msg('c', '200', '.history', { mentions: { personas: ['bot'], roles: [], users: [], everyone: false } }), msg('c', '300')] }, + { botPersonaId: 'bot' }, + ) + const r = await fetchPortalHistory('c', undefined, undefined, 50, [], false, deps) + expect(ids(r)).toEqual(['300']) + expect(r.didClear).toBe(true) + }) + + it('skips .history when authorizedRoles set but roles unverifiable (P2 deps return null)', async () => { + const deps = makeDeps({ c: [msg('c', '100'), msg('c', '200', '.history'), msg('c', '300')] }) + const r = await fetchPortalHistory('c', undefined, undefined, 50, ['admin'], false, deps) + // unauthorized/unverifiable → .history ignored, all kept + expect(ids(r)).toEqual(['100', '300']) + }) + + it('ignoreHistory bypasses .history processing entirely', async () => { + const deps = makeDeps({ c: [msg('c', '100'), msg('c', '200', '.history'), msg('c', '300')] }) + const r = await fetchPortalHistory('c', undefined, undefined, 50, [], true, deps) + expect(ids(r)).toEqual(['100', '200', '300']) + }) + + it('startFrom includes the boundary message and fetches backward', async () => { + const deps = makeDeps({ c: ['100', '200', '300', '400'].map((s) => msg('c', s)) }) + const r = await fetchPortalHistory('c', '300', undefined, 50, [], false, deps) + // from 300 backward (inclusive): 100,200,300 — not 400 + expect(ids(r)).toEqual(['100', '200', '300']) + }) +}) diff --git a/src/connector/portal/history.ts b/src/connector/portal/history.ts new file mode 100644 index 0000000..b96980a --- /dev/null +++ b/src/connector/portal/history.ts @@ -0,0 +1,193 @@ +/** + * Portal `.history` engine — a port of context-fetch.ts's fetchChannelMessages, + * operating on PortalMessage instead of discord.js Message. + * + * The critical adaptation: discord.js uses `msg.id` for BOTH identity and + * ordering (it's the snowflake). In portal those split — identity is the relay + * id (`rm__`), ordering/boundaries are the Discord + * snowflake (`pm.nativeId`). So all cursor/boundary math here uses the snowflake + * (BigInt-compared), while message identity stays the relay id downstream. + * + * The pure parsers (parseHistoryCommand / extract*FromUrl) are reused verbatim + * from context-fetch.ts. + */ +import type { PortalMessage } from '@animalabs/portal-protocol' +import { parseHistoryCommand, extractMessageIdFromUrl, extractChannelIdFromUrl } from '../../discord/context-fetch.js' +import { logger } from '../../utils/logger.js' + +const MAX_RECURSION_DEPTH = 5 + +export interface PortalChannelRef { + id: string + isTextBased: boolean +} + +export interface PortalFetchDeps { + /** Fetch a batch strictly before `before` (relay id or snowflake), any order. */ + fetchBatch(channelId: string, opts: { before?: string; limit: number }): Promise + /** Fetch a single message whose snowflake === `snowflake`, or null. */ + fetchSingle(channelId: string, snowflake: string): Promise + /** Resolve a Discord channel snowflake (from a .history URL) to a portal channel. */ + resolveChannelBySnowflake(snowflake: string): Promise + /** Author role NAMES of `pm` for .history auth; null when unverifiable. */ + authorRoles(pm: PortalMessage): Promise + botPersonaId: string +} + +export interface PortalHistoryResult { + /** Chronological (oldest first). */ + messages: PortalMessage[] + didClear: boolean + originChannelId: string | null +} + +/** Recover the Discord snowflake from a relay id (`rm__`) + * or pass through a raw snowflake. */ +export function snowflakeFromId(id: string): string { + if (id.startsWith('rm_')) { + const us = id.lastIndexOf('_') + if (us > 2) return id.slice(us + 1) + } + return id +} + +function snowflakeLE(a: string, b: string): boolean { + try { + return BigInt(a) <= BigInt(b) + } catch { + return a <= b + } +} + +/** Keep only real user/persona messages (drop relay system notices). */ +function isKeepable(pm: PortalMessage): boolean { + return pm.author.kind !== 'system' +} + +function isHistoryCommand(content: string): boolean { + return !!content && content.startsWith('.history') && /^\.history(?:\s|$)/.test(content) +} + +/** + * Fetch messages from a portal channel with .history resolution. + * + * @param channelId portal channel id (thread id ok — relay resolves it) + * @param startFrom snowflake to fetch backward from (included); undefined = latest + * @param stopAt snowflake boundary (.history range `first:`); included + */ +export async function fetchPortalHistory( + channelId: string, + startFrom: string | undefined, + stopAt: string | undefined, + maxMessages: number, + authorizedRoles: string[], + ignoreHistory: boolean, + deps: PortalFetchDeps, + _depth = 0, +): Promise { + if (_depth > MAX_RECURSION_DEPTH) { + logger.warn({ channelId, depth: _depth }, 'portal .history: max recursion depth — returning empty') + return { messages: [], didClear: true, originChannelId: null } + } + + const collected: PortalMessage[] = [] + let cursor = startFrom + + // Include the boundary/trigger message itself (fetchBatch `before` is exclusive). + if (startFrom) { + const trigger = await deps.fetchSingle(channelId, startFrom) + if (trigger) collected.push(trigger) + } + + let done = false + while (collected.length < maxMessages && !done) { + const batchLimit = Math.min(100, maxMessages - collected.length) + const batch = await deps.fetchBatch(channelId, { before: cursor, limit: batchLimit }) + if (batch.length === 0) break + + // Newest-first by snowflake. + batch.sort((a, b) => (snowflakeLE(a.nativeId, b.nativeId) ? 1 : -1)) + + for (const msg of batch) { + if (stopAt && snowflakeLE(msg.nativeId, stopAt)) { + if (msg.nativeId === stopAt) collected.push(msg) + done = true + break + } + + if (!ignoreHistory && isHistoryCommand(msg.content)) { + const res = await processPortalHistoryCommand(msg, channelId, collected, maxMessages, authorizedRoles, ignoreHistory, deps, _depth) + if (res) return res + continue // didn't apply (wrong target / unauthorized / malformed) — skip it + } + + if (!isKeepable(msg)) continue + collected.push(msg) + } + + const oldest = batch[batch.length - 1] + if (!oldest) break + cursor = oldest.nativeId + } + + return { messages: collected.reverse(), didClear: false, originChannelId: null } +} + +async function processPortalHistoryCommand( + msg: PortalMessage, + currentChannelId: string, + collectedNewer: PortalMessage[], + maxMessages: number, + authorizedRoles: string[], + ignoreHistory: boolean, + deps: PortalFetchDeps, + depth: number, +): Promise { + // ── Authorization (by role name) ── + if (authorizedRoles.length > 0) { + const roles = await deps.authorRoles(msg) + if (!roles || !authorizedRoles.some((r) => roles.includes(r))) return null + } + + // ── Bot targeting: persona role-mention. If targeted and not us, skip. ── + if (msg.mentions.personas.length > 0 && !msg.mentions.personas.includes(deps.botPersonaId)) { + return null + } + + const parsed = parseHistoryCommand(msg.content) + if (parsed === false) return null // malformed + + if (parsed === null) { + // Clear: only the messages newer than this command. + return { messages: collectedNewer.slice().reverse(), didClear: true, originChannelId: null } + } + + // ── Range ── + const targetSnowflake = extractChannelIdFromUrl(parsed.last) + const targetChannel = targetSnowflake ? await deps.resolveChannelBySnowflake(targetSnowflake) : { id: currentChannelId, isTextBased: true } + + if (!targetChannel || !targetChannel.isTextBased) { + logger.warn({ messageId: msg.id, targetSnowflake }, 'portal .history range: target channel inaccessible — treating as clear') + return { messages: collectedNewer.slice().reverse(), didClear: true, originChannelId: null } + } + + const histLast = extractMessageIdFromUrl(parsed.last) ?? undefined + const histFirst = parsed.first ? (extractMessageIdFromUrl(parsed.first) ?? undefined) : undefined + + const rangeResult = await fetchPortalHistory( + targetChannel.id, + histLast, + histFirst, + maxMessages - collectedNewer.length, + authorizedRoles, + ignoreHistory, + deps, + depth + 1, + ) + + return { + messages: [...rangeResult.messages, ...collectedNewer.slice().reverse()], + didClear: true, + originChannelId: currentChannelId, + } +} diff --git a/src/connector/portal/portal-connector.ts b/src/connector/portal/portal-connector.ts new file mode 100644 index 0000000..32b16f1 --- /dev/null +++ b/src/connector/portal/portal-connector.ts @@ -0,0 +1,482 @@ +/** + * PortalConnector — IConnector backed by the shared portal-relay over a + * websocket (portal-client), instead of an own discord.js gateway. + * + * Phase 1 scope: connect as a persona, feed inbound events to the EventQueue, + * send/edit/delete/react/typing. fetchContext is minimal (live history only), + * pins + role auth + attachment upload land in later phases. Degradations are + * marked `P1 degrade` / `P-N` below. + */ +import { PortalClient, fileFromBytes } from '@animalabs/portal-client' +import type { PortalMessage, PortalEvent } from '@animalabs/portal-protocol' +import { EventQueue } from '../../agent/event-queue.js' +import type { DiscordContext } from '../../types.js' +import { logger } from '../../utils/logger.js' +import type { IConnector, FetchContextParams, TrackedPin, PinnedSteer } from '../types.js' +import { splitContent } from '../util/split.js' +import { queueEventFromPortal, authorOfMessage, type AdapterCtx } from './adapters.js' +import { AttachmentCache } from '../util/attachment-cache.js' +import { buildPortalContext, type ContextDeps } from './context.js' +import type { PortalChannelRef } from './history.js' +import { extractConfigs, extractSteers, extractSleeps } from '../util/pin-extract.js' +import type { BotIdentity } from '../../agent/pin-target.js' + +const ROLE_CACHE_TTL_MS = 5 * 60 * 1000 + +/** PortalMessage → TrackedPin (raw content; snowflake id for stable sort). */ +function portalMessageToTrackedPin(pm: PortalMessage): TrackedPin { + const a = authorOfMessage(pm) + return { + id: pm.nativeId, + content: pm.content, + authorId: a.id, + authorBot: a.bot, + // The relay resolves role mentions to persona ids; carry both so pinned + // .config/.sleep/.steer can be matched by persona/role mention. + mentionedPersonaIds: pm.mentions?.personas ? [...pm.mentions.personas] : undefined, + mentionedRoleIds: pm.mentions?.roles ? [...pm.mentions.roles] : undefined, + } +} + +/** chatperx emits `<@username>`; the relay's outgoing-mention resolver matches + * bare `@handle`. Strip the brackets (but leave raw numeric `<@id>` alone — the + * relay can't resolve those and Discord would render the snowflake). */ +function normalizeOutgoingMentions(content: string): string { + return content.replace(/<@!?([A-Za-z][\w.]*)>/g, '@$1') +} + +export interface PortalConnectorOptions { + url: string + token: string + personaId: string + cacheDir: string + maxBackoffMs?: number +} + +/** Portal channel types that carry messages. */ +function isTextBasedType(type?: string): boolean { + return type === 'text' || type === 'thread' +} + +const AUTHOR_CAP = 5000 +const TYPING_REFRESH_MS = 8000 + +export class PortalConnector implements IConnector { + private client: PortalClient + private personaId: string + private displayName = '' + private attachments: AttachmentCache + private typingIntervals = new Map>() + /** relayId → author id, so deletes/reactions can resolve their author. */ + private authorById = new Map() + /** channelId → tracked pins (bootstrapped via list_pins, refreshed on pins_update). */ + private pinnedByChannel = new Map() + /** guildId → (roleId → name) with a fetch timestamp, for name-based auth. */ + private roleNameCache = new Map; at: number }>() + + constructor( + private queue: EventQueue, + opts: PortalConnectorOptions, + ) { + this.personaId = opts.personaId + this.attachments = new AttachmentCache(opts.cacheDir) + this.client = new PortalClient({ + url: opts.url, + token: opts.token, + personaId: opts.personaId, + maxBackoffMs: opts.maxBackoffMs, + }) + } + + // ── Lifecycle ── + + async start(): Promise { + this.client.on('event', (e) => this.onEvent(e)) + this.client.on('close', (info) => + logger.warn({ code: info.code, willReconnect: info.willReconnect }, 'portal connection closed'), + ) + this.client.on('error', (err) => logger.error({ err: err.message }, 'portal client error')) + await this.client.connect() + this.displayName = this.client.cache.persona?.displayName ?? '' + logger.info({ personaId: this.personaId, displayName: this.displayName }, 'portal connector ready') + } + + async close(): Promise { + for (const t of this.typingIntervals.values()) clearInterval(t) + this.typingIntervals.clear() + this.client.close() + } + + // ── Inbound ── + + private onEvent(e: PortalEvent): void { + if (e.type === 'message_create' || e.type === 'message_update') this.recordAuthor(e.message) + // Keep the pin cache warm so the hot-path getCachedPinned* getters stay fresh. + if (e.type === 'pins_update') void this.bootstrapPins(e.channelId) + const ev = queueEventFromPortal(e, this.adapterCtx()) + if (ev) this.queue.push(ev) + } + + private adapterCtx(): AdapterCtx { + return { + botPersonaId: this.personaId, + authorOf: (id) => this.authorById.get(id), + guildOf: (channelId) => this.client.cache.getChannel(channelId)?.guildId ?? '', + } + } + + private recordAuthor(pm: PortalMessage): void { + this.authorById.set(pm.id, authorOfMessage(pm).id) + if (this.authorById.size > AUTHOR_CAP) { + const first = this.authorById.keys().next().value + if (first !== undefined) this.authorById.delete(first) + } + } + + // ── Identity ── + + getBotUserId(): string | undefined { + return this.personaId + } + isPortal(): boolean { + return true + } + getBotUsername(): string | undefined { + return this.displayName || undefined + } + getBotDiscordIdentity(): { userId?: string; username?: string; globalName?: string } { + return { userId: this.personaId, username: this.displayName, globalName: this.displayName } + } + generateInviteUrl(): string | undefined { + return undefined // the relay owns the gateway + invite + } + + // ── Send ── + + async sendMessage(channelId: string, content: string, replyToMessageId?: string): Promise { + const ids: string[] = [] + const chunks = splitContent(normalizeOutgoingMentions(content)) + for (let i = 0; i < chunks.length; i++) { + const chunk = chunks[i]! + // The relay resolves a thread id passed as channelId into parent+thread. + // Only the first chunk replies to the trigger (matches DiscordConnector); + // forwarding replyToId on every chunk piles replies on the original message. + const { messageId } = await this.client.sendMessage({ + channelId, + content: chunk, + replyToId: i === 0 ? replyToMessageId : undefined, + }) + ids.push(messageId) + } + return ids + } + + async sendMessageWithAttachment( + channelId: string, + content: string, + attachment: { name: string; content: string }, + replyToMessageId?: string, + ): Promise { + const file = fileFromBytes(attachment.name, Buffer.from(attachment.content, 'utf-8')) + const { messageId } = await this.client.sendMessage({ + channelId, + content: content || undefined, + files: [file], + replyToId: replyToMessageId, + }) + return [messageId] + } + + async sendImageAttachment( + channelId: string, + imageBase64: string, + mediaType = 'image/png', + caption?: string, + replyToMessageId?: string, + ): Promise { + const ext = mediaType.split('/')[1] || 'png' + const file = fileFromBytes(`image.${ext}`, Buffer.from(imageBase64, 'base64'), { contentType: mediaType }) + const { messageId } = await this.client.sendMessage({ + channelId, + content: caption || undefined, + files: [file], + replyToId: replyToMessageId, + }) + return [messageId] + } + + async sendFileAttachment( + channelId: string, + fileBuffer: Buffer, + filename: string, + contentType: string, + caption?: string, + replyToMessageId?: string, + ): Promise { + const file = fileFromBytes(filename, fileBuffer, { contentType }) + const { messageId } = await this.client.sendMessage({ + channelId, + content: caption || undefined, + files: [file], + replyToId: replyToMessageId, + }) + return [messageId] + } + + async sendWebhook(channelId: string, content: string, username: string): Promise { + // A persona is itself one webhook identity; inline the tool name for attribution. + await this.sendMessage(channelId, `**${username}:** ${content}`) + } + + // ── Edit / react / pin / delete ── + + async editMessage(_channelId: string, messageId: string, newContent: string): Promise { + await this.client.editMessage(messageId, newContent) + } + + async findAndEditBotMessage( + channelId: string, + contentPrefix: string, + newContent: string, + maxMessages = 20, + ): Promise { + const { messages } = await this.client.fetchHistory({ channelId, limit: maxMessages }) + const target = messages.find( + (m) => m.author.kind === 'persona' && m.author.personaId === this.personaId && m.content.startsWith(contentPrefix), + ) + if (!target) return false + await this.client.editMessage(target.id, newContent) + return true + } + + async deleteMessage(_channelId: string, messageId: string): Promise { + await this.client.deleteMessage(messageId) + } + + async addReaction(_channelId: string, messageId: string, emoji: string): Promise { + // Personas can't add native Discord reactions; portal records a pseudo + // reaction (structured, not shown to humans). Visible feedback is P4. + await this.client.react(messageId, emoji, false) + } + + async reactToLatestMessage(channelId: string, emoji: string): Promise { + const { messages } = await this.client.fetchHistory({ channelId, limit: 1 }) + const latest = messages[0] + if (latest) await this.client.react(latest.id, emoji, false) + } + + async pinMessage(channelId: string, messageId: string): Promise { + // Portal has no pin-mutation RPC (read-only list_pins). No-op + warn. + logger.warn({ channelId, messageId }, 'portal: pinMessage unsupported (no pin-mutation RPC)') + } + + // ── Typing ── + + async startTyping(channelId: string): Promise { + if (this.typingIntervals.has(channelId)) return + const fire = () => this.client.call('set_typing', { channelId }).catch(() => {}) + fire() + this.typingIntervals.set(channelId, setInterval(fire, TYPING_REFRESH_MS)) + } + + async stopTyping(channelId: string): Promise { + const t = this.typingIntervals.get(channelId) + if (t) { + clearInterval(t) + this.typingIntervals.delete(channelId) + } + } + + // ── Context / history ── + + async fetchContext(params: FetchContextParams): Promise { + return buildPortalContext(params, this.contextDeps()) + } + + /** Build the injected deps for the .history engine + context assembly. */ + private contextDeps(): ContextDeps { + return { + botPersonaId: this.personaId, + attachments: this.attachments, + fetchBatch: async (channelId, opts) => { + const { messages } = await this.client.fetchHistory({ channelId, before: opts.before, limit: opts.limit }) + return messages + }, + fetchSingle: (channelId, snowflake) => this.fetchSingleBySnowflake(channelId, snowflake), + resolveChannelBySnowflake: (snowflake) => this.resolveChannelBySnowflake(snowflake), + authorRoles: (pm) => this.authorRolesOf(pm), + channelMeta: (channelId) => { + const ch = this.client.cache.getChannel(channelId) + if (!ch) return null + return { guildId: ch.guildId ?? '', isThread: ch.type === 'thread', parentId: ch.parentId, native: ch.native } + }, + } + } + + /** Fetch the single message whose snowflake === `snowflake`, or null. Portal + * `fetch_history` `before` is exclusive, so ask for the page before snowflake+1. */ + private async fetchSingleBySnowflake(channelId: string, snowflake: string): Promise { + let before: string | undefined + try { + before = (BigInt(snowflake) + 1n).toString() + } catch { + before = undefined + } + const { messages } = await this.client.fetchHistory({ channelId, before, limit: 1 }) + const m = messages[0] + return m && m.nativeId === snowflake ? m : null + } + + /** Resolve a Discord channel snowflake (from a .history URL) to a portal + * channel ref, populating the cache via list_channels on a miss. */ + private async resolveChannelBySnowflake(snowflake: string): Promise { + let ch = this.client.cache.allChannels().find((c) => c.native === snowflake) + if (!ch) { + for (const g of this.client.cache.allGuilds()) { + await this.client.call('list_channels', { guildId: g.id }).catch(() => undefined) + } + ch = this.client.cache.allChannels().find((c) => c.native === snowflake) + } + return ch ? { id: ch.id, isTextBased: isTextBasedType(ch.type) } : null + } + + async getChannelMeta( + channelId: string, + ): Promise<{ name?: string; isThread: boolean; parentChannelId?: string }> { + const ch = this.client.cache.getChannel(channelId) + return { + name: ch?.name ?? undefined, + isThread: ch?.type === 'thread', + parentChannelId: ch?.parentId, + } + } + + async getParentChannelId(channelId: string): Promise { + return this.client.cache.getChannel(channelId)?.parentId + } + + async getMessageBefore( + channelId: string, + beforeMessageId: string, + ): Promise<{ id: string; author: { id: string; bot: boolean } } | null> { + const { messages } = await this.client.fetchHistory({ channelId, before: beforeMessageId, limit: 1 }) + const m = messages[0] + if (!m) return null + const a = authorOfMessage(m) + return { id: m.id, author: { id: a.id, bot: a.bot } } + } + + async getBotReplyChainDepth( + channelId: string, + message: { id: string; author?: { id?: string; bot?: boolean }; reference?: { messageId?: string } }, + ): Promise { + // Walk the reply chain over a recent window, counting consecutive bot msgs. + const { messages } = await this.client.fetchHistory({ channelId, limit: 100 }) + const byId = new Map(messages.map((m) => [m.id, m])) + let depth = 0 + let cursor = message.reference?.messageId + while (cursor) { + const m = byId.get(cursor) + if (!m) break + const a = authorOfMessage(m) + if (!a.bot) break + depth++ + cursor = m.replyToId + } + return depth + } + + // ── Pins ── + + /** Bootstrap (or refresh) a channel's pin cache via list_pins. */ + private async bootstrapPins(channelId: string): Promise { + try { + const { messages } = await this.client.call('list_pins', { channelId }) + this.pinnedByChannel.set(channelId, messages.map(portalMessageToTrackedPin)) + } catch (err) { + logger.warn({ err: (err as Error).message, channelId }, 'portal: list_pins failed') + if (!this.pinnedByChannel.has(channelId)) this.pinnedByChannel.set(channelId, []) + } + } + + private async ensurePins(channelId: string): Promise { + let pins = this.pinnedByChannel.get(channelId) + if (!pins) { + await this.bootstrapPins(channelId) + pins = this.pinnedByChannel.get(channelId) + } + return pins ?? [] + } + + /** Identity the relay-routed persona resolves pin targets against: its persona + * id (relay-resolved role mentions) and display name. botId / config name are + * matched downstream in config parsing. */ + private pinMatchContext(): { identity: BotIdentity } { + return { + identity: { + botId: '', + discordUsername: this.getBotUsername() ?? undefined, + discordUserId: this.getBotUserId() ?? undefined, + }, + } + } + + /** Portal personas have no own guild roles; targeting is via persona mention. */ + getOwnRoleIds(_channelId: string): string[] | null { + return null + } + + async fetchPinnedConfigs(channelId: string): Promise { + return extractConfigs(await this.ensurePins(channelId), this.pinMatchContext()) + } + async fetchPinnedSteerMessages(channelId: string): Promise { + return extractSteers(await this.ensurePins(channelId)) + } + async fetchPinnedSleeps(channelId: string): Promise { + return extractSleeps(await this.ensurePins(channelId)) + } + getCachedPinnedConfigs(channelId: string): string[] | null { + const pins = this.pinnedByChannel.get(channelId) + return pins ? extractConfigs(pins, this.pinMatchContext()) : null + } + getCachedPinnedSteers(channelId: string): PinnedSteer[] | null { + const pins = this.pinnedByChannel.get(channelId) + return pins ? extractSteers(pins) : null + } + getCachedPinnedSleeps(channelId: string): TrackedPin[] | null { + const pins = this.pinnedByChannel.get(channelId) + return pins ? extractSleeps(pins) : null + } + + // ── Auth (role names via list_members + list_roles) ── + + /** guildId → (roleId → name), cached with a short TTL. */ + private async rolesFor(guildId: string): Promise> { + const cached = this.roleNameCache.get(guildId) + if (cached && Date.now() - cached.at < ROLE_CACHE_TTL_MS) return cached.map + const { roles } = await this.client.call('list_roles', { guildId }) + const map = new Map(roles.map((r) => [r.id, r.name])) + this.roleNameCache.set(guildId, { map, at: Date.now() }) + return map + } + + async fetchMemberRoles(userId: string, guildId: string): Promise { + try { + const { members, membersAvailable } = await this.client.call('list_members', { guildId, limit: 1000 }) + if (!membersAvailable) return null + const member = members.find((m) => m.userId === userId) + if (!member) return null + const names = await this.rolesFor(guildId) + return member.roles.map((id) => names.get(id)).filter((n): n is string => !!n) + } catch (err) { + logger.warn({ err: (err as Error).message, userId, guildId }, 'portal: fetchMemberRoles failed') + return null + } + } + + /** Role names of a message's author (for .history auth); null if not a guild user. */ + private async authorRolesOf(pm: PortalMessage): Promise { + if (!pm.guildId || pm.author.kind !== 'user') return null + return this.fetchMemberRoles(pm.author.userId, pm.guildId) + } +} diff --git a/src/connector/types.ts b/src/connector/types.ts new file mode 100644 index 0000000..8d4f868 --- /dev/null +++ b/src/connector/types.ts @@ -0,0 +1,173 @@ +/** + * Backend-agnostic connector contract. + * + * Both `DiscordConnector` (own discord.js gateway) and `PortalConnector` (routed + * through the shared portal-relay) implement `IConnector`. The agent loop, API + * server, and sleep logic depend only on this interface, so the backend is + * selectable at runtime (see `CONNECTOR_BACKEND` in main.ts). + * + * `ConnectorOptions`, `TrackedPin`, and `FetchContextParams` live here (rather + * than in discord/connector.ts) so the portal backend can reference them without + * importing discord.js. connector.ts re-exports them for back-compat. + */ +import type { DiscordContext } from '../types.js' + +export interface ConnectorOptions { + token: string + cacheDir: string + maxBackoffMs: number +} + +/** + * A pinned message tracked via gateway/relay events. The minimal set of fields + * needed to resolve .config / .steer / .sleep without re-hitting a pins endpoint. + */ +export interface TrackedPin { + id: string + content: string + authorId: string + authorBot: boolean + /** Relay-resolved persona ids addressed by this pin (portal backend only). */ + mentionedPersonaIds?: string[] + /** Discord/relay role ids mentioned in this pin (for `<@&roleId>` targeting). */ + mentionedRoleIds?: string[] +} + +/** A pinned `.steer` message with its resolved mention context. */ +export interface PinnedSteer { + content: string + authorId: string + mentionedPersonaIds?: string[] + mentionedRoleIds?: string[] +} + +export interface FetchContextParams { + channelId: string + depth: number // Max messages + targetMessageId?: string // Fetch backward from this message ID (API range queries) + firstMessageId?: string // Stop when this message is encountered + authorized_roles?: string[] + pinnedConfigs?: string[] // Pre-fetched pinned configs (skips fetchPinned call) + maxImages?: number // Cap image fetching (default: unlimited) + ignoreHistory?: boolean // Skip .history command processing (raw fetch) +} + +/** + * The discord.js-`Message`-compatible shape the agent loop reads off the + * EventQueue's `event.data`. The discord.js backend pushes real `Message` + * objects (which structurally satisfy this); the portal backend synthesizes it + * from a `PortalMessage` (see connector/portal/adapters.ts). `event.data` is + * typed `any`, so this type is documentation + the adapter's construction target. + */ +export interface InboundMessage { + id: string + channelId: string + guildId: string + content: string + author: { id: string; username: string; bot: boolean; globalName?: string } + system?: boolean + pinned?: boolean + reference?: { messageId?: string } + mentions: { has(id: string): boolean; users?: Map } + reactions?: { cache: { some(fn: (r: unknown) => boolean): boolean } } + member?: { roles: { cache: Map } } + /** + * Per-recipient addressing annotation from the portal relay (portal backend + * only; absent for the discord.js backend). Lets the loop distinguish a direct + * address (`role_mention` / `reply`) from ambient `subscription` traffic when + * the persona has no native Discord identity to appear in `mentions`. In + * particular, a reply-ping to a webhook persona never lands in `mentions` + * (webhooks can't carry a native reply), so `reasons` carries the only signal. + */ + _address?: { addressedToMe: boolean; reasons: string[] } + /** Escape hatch: the source PortalMessage, for adapters that need raw fields. */ + _portal?: unknown +} + +export interface IConnector { + // ── Lifecycle ── + start(): Promise + close(): Promise + + // ── Identity ── + getBotUserId(): string | undefined + /** True for the portal-relay backend (no own Discord account). */ + isPortal(): boolean + getBotUsername(): string | undefined + getBotDiscordIdentity(): { userId?: string; username?: string; globalName?: string } + generateInviteUrl(options?: { + permissions?: bigint | string[] + guildId?: string + disableGuildSelect?: boolean + }): string | undefined + + // ── Context / history ── + fetchContext(params: FetchContextParams): Promise + getChannelMeta(channelId: string): Promise<{ name?: string; isThread: boolean; parentChannelId?: string }> + getParentChannelId(channelId: string): Promise + /** Narrowed from discord.js `Message` — callers only read `.id`/`.author`. */ + getMessageBefore( + channelId: string, + beforeMessageId: string, + ): Promise<{ id: string; author: { id: string; bot: boolean } } | null> + /** Narrowed param — reimplemented over reply chains. */ + getBotReplyChainDepth( + channelId: string, + message: { id: string; author?: { id?: string; bot?: boolean }; reference?: { messageId?: string } }, + ): Promise + + // ── Pins (async fetch + sync cache) ── + fetchPinnedConfigs(channelId: string): Promise + fetchPinnedSteerMessages(channelId: string): Promise + fetchPinnedSleeps(channelId: string): Promise + getCachedPinnedConfigs(channelId: string): string[] | null + getCachedPinnedSteers(channelId: string): PinnedSteer[] | null + getCachedPinnedSleeps(channelId: string): TrackedPin[] | null + /** The bot's own role ids in a channel's guild (account bots; null if uncached/portal). */ + getOwnRoleIds(channelId: string): string[] | null + + // ── Send ── + sendMessage(channelId: string, content: string, replyToMessageId?: string): Promise + sendMessageWithAttachment( + channelId: string, + content: string, + attachment: { name: string; content: string }, + replyToMessageId?: string, + ): Promise + sendImageAttachment( + channelId: string, + imageBase64: string, + mediaType?: string, + caption?: string, + replyToMessageId?: string, + ): Promise + sendFileAttachment( + channelId: string, + fileBuffer: Buffer, + filename: string, + contentType: string, + caption?: string, + replyToMessageId?: string, + ): Promise + sendWebhook(channelId: string, content: string, username: string): Promise + + // ── Edit / react / pin / delete ── + editMessage(channelId: string, messageId: string, newContent: string): Promise + findAndEditBotMessage( + channelId: string, + contentPrefix: string, + newContent: string, + maxMessages?: number, + ): Promise + deleteMessage(channelId: string, messageId: string): Promise + addReaction(channelId: string, messageId: string, emoji: string): Promise + reactToLatestMessage(channelId: string, emoji: string): Promise + pinMessage(channelId: string, messageId: string): Promise + + // ── Typing ── + startTyping(channelId: string): Promise + stopTyping(channelId: string): Promise + + // ── Auth ── + fetchMemberRoles(userId: string, guildId: string): Promise +} diff --git a/src/connector/util/attachment-cache.ts b/src/connector/util/attachment-cache.ts new file mode 100644 index 0000000..5045b87 --- /dev/null +++ b/src/connector/util/attachment-cache.ts @@ -0,0 +1,198 @@ +/** + * Backend-agnostic attachment cache: download + disk-cache images, fetch text + * attachments. Ported from DiscordConnector's image-cache logic so the portal + * backend can produce identical CachedImage/CachedDocument without discord.js. + * + * (DiscordConnector keeps its own copy for now; a later refactor can point both + * at this class.) + */ +import { existsSync, mkdirSync, readFileSync, writeFileSync } from 'fs' +import { join } from 'path' +import { createHash } from 'crypto' +import sharp from 'sharp' +import { logger } from '../../utils/logger.js' +import type { CachedImage, CachedDocument } from '../../types.js' + +const MAX_TEXT_ATTACHMENT_BYTES = 200_000 // ~200 KB inline text per attachment + +/** Minimal attachment shape (PortalAttachment maps directly). */ +export interface AttachmentLike { + url: string + name: string + contentType?: string | null + size?: number +} + +const TEXT_MIME = [ + 'text/', + 'application/json', + 'application/xml', + 'application/javascript', + 'application/typescript', + 'application/x-yaml', + 'application/yaml', + 'application/x-sh', + 'application/x-python', +] +const TEXT_EXT = [ + '.txt', '.md', '.markdown', '.rst', '.py', '.js', '.ts', '.jsx', '.tsx', '.mjs', '.cjs', + '.json', '.yaml', '.yml', '.toml', '.xml', '.html', '.htm', '.css', '.scss', '.sass', '.less', + '.sh', '.bash', '.zsh', '.fish', '.c', '.cpp', '.h', '.hpp', '.cc', '.cxx', '.java', '.rs', '.go', + '.rb', '.php', '.sql', '.graphql', '.gql', '.lua', '.perl', '.pl', '.r', '.R', '.swift', '.kt', + '.kts', '.scala', '.vim', '.el', '.lisp', '.clj', '.cljs', '.ini', '.cfg', '.conf', '.config', + '.log', '.csv', '.tsv', +] + +export class AttachmentCache { + private imageCache = new Map() + private urlToFilename = new Map() + private mapPath: string + + constructor(private cacheDir: string) { + if (!existsSync(cacheDir)) mkdirSync(cacheDir, { recursive: true }) + this.mapPath = join(cacheDir, 'url-map.json') + this.loadUrlMap() + } + + isTextAttachment(att: AttachmentLike): boolean { + if (att.contentType && TEXT_MIME.some((m) => att.contentType!.startsWith(m))) return true + const name = att.name?.toLowerCase() || '' + return TEXT_EXT.some((ext) => name.endsWith(ext)) + } + + private detectImageType(buffer: Buffer): string | null { + if (buffer.length < 4) return null + if (buffer[0] === 0x89 && buffer[1] === 0x50 && buffer[2] === 0x4e && buffer[3] === 0x47) return 'image/png' + if (buffer[0] === 0xff && buffer[1] === 0xd8 && buffer[2] === 0xff) return 'image/jpeg' + if (buffer[0] === 0x47 && buffer[1] === 0x49 && buffer[2] === 0x46 && buffer[3] === 0x38) return 'image/gif' + if ( + buffer.length >= 12 && + buffer[0] === 0x52 && buffer[1] === 0x49 && buffer[2] === 0x46 && buffer[3] === 0x46 && + buffer[8] === 0x57 && buffer[9] === 0x45 && buffer[10] === 0x42 && buffer[11] === 0x50 + ) return 'image/webp' + return null + } + + private async dimsAndTokens(buffer: Buffer): Promise<{ width?: number; height?: number; tokenEstimate: number }> { + let width: number | undefined + let height: number | undefined + try { + const meta = await sharp(buffer).metadata() + width = meta.width + height = meta.height + } catch { + // dimensions optional + } + const estW = Math.min(width || 1024, 1568) + const estH = Math.min(height || 1024, 1568) + return { width, height, tokenEstimate: Math.ceil((estW * estH) / 750) } + } + + /** True if `url` is already cached (memory or disk-map). */ + has(url: string): boolean { + return this.imageCache.has(url) || this.urlToFilename.has(url) + } + + async cacheImage(url: string, _contentType: string): Promise { + const mem = this.imageCache.get(url) + if (mem) return mem + + // Disk cache via url→filename map. + const cachedFilename = this.urlToFilename.get(url) + if (cachedFilename) { + const filepath = join(this.cacheDir, cachedFilename) + if (existsSync(filepath)) { + try { + const buffer = readFileSync(filepath) + const hash = cachedFilename.split('.')[0] || '' + const ext = cachedFilename.split('.')[1] || 'jpg' + const { width, height, tokenEstimate } = await this.dimsAndTokens(buffer) + const cached: CachedImage = { url, data: buffer, mediaType: `image/${ext}`, hash, width, height, tokenEstimate } + this.imageCache.set(url, cached) + return cached + } catch (error) { + logger.warn({ error, url, filepath }, 'portal: failed to read cached image from disk') + } + } + } + + // Download (cache miss). + try { + const response = await fetch(url) + if (!response.ok) { + logger.warn({ url, status: response.status }, 'portal: image download failed') + return null + } + const buffer = Buffer.from(await response.arrayBuffer()) + const detectedType = this.detectImageType(buffer) + if (!detectedType) { + logger.warn({ url, bufferSize: buffer.length }, 'portal: no valid image magic bytes, skipping') + return null + } + const hash = createHash('sha256').update(buffer).digest('hex') + const ext = detectedType.split('/')[1] || 'jpg' + const filename = `${hash}.${ext}` + const filepath = join(this.cacheDir, filename) + if (!existsSync(filepath)) writeFileSync(filepath, buffer) + this.urlToFilename.set(url, filename) + const { width, height, tokenEstimate } = await this.dimsAndTokens(buffer) + const cached: CachedImage = { url, data: buffer, mediaType: detectedType, hash, width, height, tokenEstimate } + this.imageCache.set(url, cached) + return cached + } catch (error) { + logger.warn({ error, url }, 'portal: failed to cache image') + return null + } + } + + async fetchTextAttachment(att: AttachmentLike, messageId: string): Promise { + if (att.size && att.size > MAX_TEXT_ATTACHMENT_BYTES * 4) { + logger.warn({ size: att.size, url: att.url }, 'portal: skipping oversized text attachment') + return null + } + try { + const response = await fetch(att.url) + if (!response.ok) { + logger.warn({ status: response.status, url: att.url }, 'portal: failed to fetch text attachment') + return null + } + let buffer = Buffer.from(await response.arrayBuffer()) + let truncated = false + if (buffer.length > MAX_TEXT_ATTACHMENT_BYTES) { + buffer = buffer.subarray(0, MAX_TEXT_ATTACHMENT_BYTES) + truncated = true + } + return { + messageId, + url: att.url, + filename: att.name || 'attachment.txt', + contentType: att.contentType || 'text/plain', + size: att.size ?? buffer.length, + text: buffer.toString('utf-8'), + truncated, + } + } catch (error) { + logger.warn({ error, url: att.url }, 'portal: failed to download text attachment') + return null + } + } + + private loadUrlMap(): void { + try { + if (existsSync(this.mapPath)) { + const raw = JSON.parse(readFileSync(this.mapPath, 'utf-8')) as Record + this.urlToFilename = new Map(Object.entries(raw)) + } + } catch (error) { + logger.warn({ error, mapPath: this.mapPath }, 'portal: failed to load url map') + } + } + + saveUrlMap(): void { + try { + writeFileSync(this.mapPath, JSON.stringify(Object.fromEntries(this.urlToFilename))) + } catch (error) { + logger.warn({ error, mapPath: this.mapPath }, 'portal: failed to save url map') + } + } +} diff --git a/src/connector/util/pin-extract.test.ts b/src/connector/util/pin-extract.test.ts new file mode 100644 index 0000000..9cc0d7f --- /dev/null +++ b/src/connector/util/pin-extract.test.ts @@ -0,0 +1,45 @@ +import { describe, it, expect } from 'vitest' +import type { TrackedPin } from '../types.js' +import { extractConfigs, extractSteers, extractSleeps } from './pin-extract.js' + +const pin = (id: string, content: string, over: Partial = {}): TrackedPin => ({ + id, + content, + authorId: 'u', + authorBot: false, + ...over, +}) + +describe('extractConfigs', () => { + it('extracts yaml from .config pins, prepending target', () => { + const pins = [ + pin('2', '.config\n---\nfoo: 1'), + pin('1', '.config Mythos\n---\nbar: 2'), + pin('3', 'not a config'), + ] + // sorted by id: 1 (targeted), 2 (plain) + expect(extractConfigs(pins)).toEqual(['target: Mythos\nbar: 2', 'foo: 1']) + }) + + it('ignores malformed .config (no --- separator)', () => { + expect(extractConfigs([pin('1', '.config\nfoo: 1')])).toEqual([]) + }) +}) + +describe('extractSteers', () => { + it('returns non-bot .steer pins with authorId', () => { + const pins = [ + pin('1', '.steer be nice', { authorId: 'human1' }), + pin('2', '.steer evil', { authorBot: true, authorId: 'bot1' }), + pin('3', '.config\n---\nx: 1'), + ] + expect(extractSteers(pins)).toEqual([{ content: '.steer be nice', authorId: 'human1' }]) + }) +}) + +describe('extractSleeps', () => { + it('returns .sleep pins as full records, sorted by id', () => { + const pins = [pin('2', '.sleep messages: 3'), pin('1', '.sleep'), pin('3', '.config\n---\nx: 1')] + expect(extractSleeps(pins).map((p) => p.id)).toEqual(['1', '2']) + }) +}) diff --git a/src/connector/util/pin-extract.ts b/src/connector/util/pin-extract.ts new file mode 100644 index 0000000..0feb88d --- /dev/null +++ b/src/connector/util/pin-extract.ts @@ -0,0 +1,67 @@ +/** + * Pin classification — `.config` / `.steer` / `.sleep` extraction from tracked + * pins. Ported from DiscordConnector so the portal backend produces identical + * output. Pins are sorted by id (chronological) before extraction, matching the + * discord path. + */ +import type { TrackedPin, PinnedSteer } from '../types.js' +import { pinAddressesBot, type BotIdentity } from '../../agent/pin-target.js' + +function sortById(pins: TrackedPin[]): TrackedPin[] { + return [...pins].sort((a, b) => a.id.localeCompare(b.id)) +} + +/** Identity + roles a connector resolves a pin target against. */ +export interface PinMatchContext { + identity: BotIdentity + ownRoleIds?: readonly string[] +} + +/** + * `.config [target]\n---\n` → yaml. When `ctx` is supplied and the target + * (its text, or a relay-resolved role/persona mention) addresses this bot, the + * target is stripped so parseChannelConfig applies it as a bare config. + * Otherwise the `target:` text is preserved for downstream botId/config-name + * matching. Without `ctx`, behaves as the original (target text preserved). + */ +export function extractConfigs(pins: TrackedPin[], ctx?: PinMatchContext): string[] { + const configs: string[] = [] + for (const pin of sortById(pins)) { + if (!pin.content.startsWith('.config')) continue + const lines = pin.content.split('\n') + if (lines.length > 2 && lines[1] === '---') { + const target = lines[0]!.slice('.config'.length).trim() || undefined + const yaml = lines.slice(2).join('\n') + const addressesMe = ctx + ? pinAddressesBot( + { targetText: target, mentionedPersonaIds: pin.mentionedPersonaIds, mentionedRoleIds: pin.mentionedRoleIds }, + ctx.identity, + ctx.ownRoleIds, + ) + : false + configs.push(target && !addressesMe ? `target: ${target}\n${yaml}` : yaml) + } + } + return configs +} + +/** `.steer` pins not authored by a bot → PinnedSteer (carries resolved mentions). */ +export function extractSteers(pins: TrackedPin[]): PinnedSteer[] { + const out: PinnedSteer[] = [] + for (const pin of sortById(pins)) { + if (pin.content.startsWith('.steer') && !pin.authorBot) { + out.push({ + content: pin.content, + authorId: pin.authorId, + mentionedPersonaIds: pin.mentionedPersonaIds, + mentionedRoleIds: pin.mentionedRoleIds, + }) + } + } + return out +} + +/** `.sleep` pins (full records — sleep-state counters key on pin id). */ +export function extractSleeps(pins: TrackedPin[]): TrackedPin[] { + return sortById(pins).filter((p) => p.content.startsWith('.sleep')) +} diff --git a/src/connector/util/split.ts b/src/connector/util/split.ts new file mode 100644 index 0000000..5e9dcb4 --- /dev/null +++ b/src/connector/util/split.ts @@ -0,0 +1,18 @@ +/** + * Split content into chunks ≤ maxLen, preferring newline then space boundaries. + * Mirrors the discord connector's 1800-char auto-split. Returns [] for empty. + */ +export function splitContent(content: string, maxLen = 1800): string[] { + if (content.length <= maxLen) return content.length ? [content] : [] + const chunks: string[] = [] + let rest = content + while (rest.length > maxLen) { + let cut = rest.lastIndexOf('\n', maxLen) + if (cut < maxLen * 0.5) cut = rest.lastIndexOf(' ', maxLen) + if (cut < maxLen * 0.5) cut = maxLen + chunks.push(rest.slice(0, cut)) + rest = rest.slice(cut).replace(/^\n/, '') + } + if (rest.length) chunks.push(rest) + return chunks +} diff --git a/src/discord/connector.ts b/src/discord/connector.ts index 3256595..1fb52b2 100644 --- a/src/discord/connector.ts +++ b/src/discord/connector.ts @@ -23,36 +23,10 @@ import { fetchChannelMessages, type FetchDeps, } from './context-fetch.js' +import type { IConnector, ConnectorOptions, TrackedPin, FetchContextParams, PinnedSteer } from '../connector/types.js' -export interface ConnectorOptions { - token: string - cacheDir: string - maxBackoffMs: number -} - -/** - * A pinned message tracked via gateway events. - * Holds the minimal set of fields needed to resolve .config / .steer / .sleep - * without calling the Discord /pins endpoint after the initial bootstrap. - */ -export interface TrackedPin { - id: string - content: string - authorId: string - authorBot: boolean - /** Relay-resolved persona ids addressed by this pin (portal backend only). */ - mentionedPersonaIds?: string[] - /** Discord role ids mentioned in this pin (for `<@&roleId>` targeting). */ - mentionedRoleIds?: string[] -} - -/** A pinned `.steer` message with its resolved mention context. */ -export interface PinnedSteer { - content: string - authorId: string - mentionedPersonaIds?: string[] - mentionedRoleIds?: string[] -} +// Re-exported from connector/types.ts for back-compat with existing imports. +export type { ConnectorOptions, TrackedPin, FetchContextParams, PinnedSteer } /** Extract mentioned role ids from a discord.js-ish message `mentions` object. */ function extractMentionedRoleIds( @@ -72,18 +46,7 @@ function snowflakeToTimestamp(id: string): number { return Number(BigInt(id) >> 22n) + DISCORD_EPOCH } -export interface FetchContextParams { - channelId: string - depth: number // Max messages - targetMessageId?: string // Optional: Fetch backward from this message ID (for API range queries) - firstMessageId?: string // Optional: Stop when this message is encountered - authorized_roles?: string[] - pinnedConfigs?: string[] // Optional: Pre-fetched pinned configs (skips fetchPinned call) - maxImages?: number // Optional: Cap image fetching to avoid RAM bloat (default: unlimited) - ignoreHistory?: boolean // Optional: Skip .history command processing (raw fetch) -} - -export class DiscordConnector { +export class DiscordConnector implements IConnector { private client: Client private typingIntervals = new Map() private imageCache = new Map() @@ -368,6 +331,10 @@ export class DiscordConnector { return this.client.user?.id } + isPortal(): boolean { + return false + } + /** * Get bot's Discord username */ diff --git a/src/main.ts b/src/main.ts index 736b1f8..656e90b 100644 --- a/src/main.ts +++ b/src/main.ts @@ -5,10 +5,13 @@ import { readFileSync } from 'fs' import { join } from 'path' +import { parse as parseYaml } from 'yaml' import { EventQueue } from './agent/event-queue.js' import { AgentLoop } from './agent/loop.js' import { ChannelStateManager } from './agent/state-manager.js' import { DiscordConnector } from './discord/connector.js' +import { PortalConnector } from './connector/portal/portal-connector.js' +import type { IConnector } from './connector/types.js' import { ConfigSystem } from './config/system.js' import { ContextBuilder } from './context/builder.js' import { ToolSystem } from './tools/system.js' @@ -57,19 +60,40 @@ async function main() { const toolsPath = process.env.TOOLS_PATH || './tools' const cachePath = process.env.CACHE_PATH || './cache' - // Read Discord token from file - let discordToken: string - - try { - discordToken = readFileSync(tokenFilePath, 'utf-8').trim() - logger.info({ tokenFile: tokenFilePath }, 'Discord token loaded from file') - } catch (error) { - logger.error({ error, tokenFile: tokenFilePath }, 'Failed to read discord_token file') - throw new Error(`Could not read token file: ${tokenFilePath}. Please create it with your bot token.`) + // ── Connector bootstrap settings (config-driven, env overrides) ── + // Read straight from the bot's config.yaml so the whole portal setup lives in + // one place (EMS: //config.yaml). Env vars still win for + // ad-hoc overrides. BOT_NAME is required for config-driven portal mode. + let botBootstrap: Record = {} + if (botNameOverride) { + const botCfgPath = emsPath + ? join(emsPath, botNameOverride, 'config.yaml') + : join(configPath, 'bots', `${botNameOverride}.yaml`) + try { + botBootstrap = parseYaml(readFileSync(botCfgPath, 'utf-8')) || {} + } catch { + logger.warn({ botCfgPath }, 'Could not pre-read bot config for connector settings') + } } - if (!discordToken) { - throw new Error('discord_token file is empty') + // Backend selection: 'portal' routes through the shared relay; default is the + // bot's own discord.js gateway. + const backend = + (process.env.CONNECTOR_BACKEND || botBootstrap.connector_backend) === 'portal' ? 'portal' : 'discord' + + // Read Discord token from file (discord backend only). + let discordToken = '' + if (backend === 'discord') { + try { + discordToken = readFileSync(tokenFilePath, 'utf-8').trim() + logger.info({ tokenFile: tokenFilePath }, 'Discord token loaded from file') + } catch (error) { + logger.error({ error, tokenFile: tokenFilePath }, 'Failed to read discord_token file') + throw new Error(`Could not read token file: ${tokenFilePath}. Please create it with your bot token.`) + } + if (!discordToken) { + throw new Error('discord_token file is empty') + } } logger.info({ configPath, toolsPath, cachePath, emsMode: !!emsPath }, 'Configuration loaded') @@ -99,12 +123,58 @@ async function main() { // Note: MCP servers are initialized on first bot activation // They are configured in bot config and can be overridden per-guild/channel - // Initialize Discord connector - const connector = new DiscordConnector(queue, { - token: discordToken, - cacheDir: cachePath + '/images', - maxBackoffMs: 32000, - }) + // Initialize the connector for the selected backend. + let connector: IConnector + if (backend === 'portal') { + const portalUrl = process.env.PORTAL_URL || botBootstrap.portal_url + if (!portalUrl) throw new Error('portal backend requires portal_url (config) or PORTAL_URL (env)') + let personaId = process.env.PORTAL_PERSONA + let portalToken = process.env.PORTAL_TOKEN + if (!personaId || !portalToken) { + // Enroll ONCE and reuse the persona forever. The creds file must live on a + // path that survives restarts/redeploys — in EMS mode default to the + // (persistent) bot dir, not the (re-deployable) code/cache dir. + const credsPath = + process.env.PORTAL_CREDS_PATH || + botBootstrap.portal_creds_path || + (emsPath && botNameOverride + ? join(emsPath, botNameOverride, 'portal-creds.json') + : join(cachePath, 'portal-creds.json')) + const { loadOrEnrollCreds } = await import('@animalabs/portal-client') + // desiredName is the portal persona display name. Defaults to BOT_NAME + // (the EMS directory name) but can be overridden via env or config for + // personas whose display identity differs from their bot slug. + const desiredName = + process.env.PORTAL_DESIRED_NAME || + botBootstrap.portal_desired_name || + botNameOverride + const creds = await loadOrEnrollCreds({ + url: portalUrl, + credsPath, + invite: process.env.PORTAL_INVITE || botBootstrap.portal_invite, + desiredName, + }) + personaId = creds.personaId + portalToken = creds.token + logger.info({ credsPath, personaId }, 'Portal persona credentials ready (enroll-once)') + } + // No explicit channel subscriptions: channel access is role/permission-based, + // tracked relay-side. + logger.info({ portalUrl, personaId }, 'Using portal backend') + connector = new PortalConnector(queue, { + url: portalUrl, + token: portalToken, + personaId, + cacheDir: cachePath + '/images', + maxBackoffMs: 32000, + }) + } else { + connector = new DiscordConnector(queue, { + token: discordToken, + cacheDir: cachePath + '/images', + maxBackoffMs: 32000, + }) + } await connector.start() @@ -113,7 +183,7 @@ async function main() { const botUsername = connector.getBotUsername() if (!botUserId || !botUsername) { - throw new Error('Failed to get bot identity from Discord') + throw new Error(`Failed to get bot identity from ${backend === 'portal' ? 'portal relay' : 'Discord'}`) } // Use BOT_NAME override (for EMS mode) or Discord username for config loading diff --git a/src/soma/client.ts b/src/soma/client.ts index 7b90ca2..f07d63a 100644 --- a/src/soma/client.ts +++ b/src/soma/client.ts @@ -23,7 +23,7 @@ export interface SomaCheckParams { userId: string // Discord user ID serverId: string // Discord guild ID channelId: string // Discord channel ID (for reactions) - botId: string // Bot's Discord ID + botId: string // Discord user id (account) or EMS bot name (portal) messageId: string // Triggering message ID triggerType: SomaTriggerType userRoles: string[] // User's role IDs for cost multipliers