diff --git a/backend/__tests__/unit/models/ChannelVerdict.test.js b/backend/__tests__/unit/models/ChannelVerdict.test.js new file mode 100644 index 0000000000..e4e0141c8c --- /dev/null +++ b/backend/__tests__/unit/models/ChannelVerdict.test.js @@ -0,0 +1,57 @@ +// This model suite uses the in-memory Mongo helper; keep its JWT dependency +// loadable on Node 26, as the other database-backed unit suites do. +jest.mock('jsonwebtoken', () => ({})); + +const mongoose = require('mongoose'); +const ChannelVerdict = require('../../../models/ChannelVerdict'); +const { setupMongoDb, closeMongoDb, clearMongoDb } = require('../../utils/testUtils'); + +describe('ChannelVerdict', () => { + beforeAll(async () => { + await setupMongoDb(); + await ChannelVerdict.syncIndexes(); + }); + afterAll(() => closeMongoDb()); + afterEach(() => clearMongoDb()); + + const row = (overrides = {}) => ({ + integrationId: new mongoose.Types.ObjectId(), + podId: new mongoose.Types.ObjectId(), + provider: 'telegram', + event: { kind: 'decision_request', podMessageId: '700' }, + verdict: 'interrupt', + reason: 'card', + at: new Date(), + ...overrides, + }); + + it('keeps one connector receipt per workspace message and preserves the D7 read indexes', async () => { + const first = row(); + await ChannelVerdict.create(first); + await expect(ChannelVerdict.create({ ...first, reason: 'retry' })) + .rejects.toMatchObject({ code: 11000 }); + + const indexes = ChannelVerdict.schema.indexes(); + expect(indexes).toEqual(expect.arrayContaining([ + [{ integrationId: 1, at: -1 }, expect.any(Object)], + [{ podId: 1, at: -1 }, expect.any(Object)], + [{ 'event.podMessageId': 1 }, expect.any(Object)], + [{ expiresAt: 1 }, expect.objectContaining({ expireAfterSeconds: 0 })], + ])); + }); + + it('accepts a lazily-linked decision id and a later channel receipt', async () => { + const verdict = new ChannelVerdict(row({ + provider: 'slack', + event: { + kind: 'decision_request', + podMessageId: '701', + decisionId: new mongoose.Types.ObjectId(), + }, + reachedHumanAt: new Date(), + ruledVia: 'slack', + expiresAt: new Date(Date.now() + (90 * 24 * 60 * 60 * 1000)), + })); + await expect(verdict.validate()).resolves.toBeUndefined(); + }); +}); diff --git a/backend/__tests__/unit/services/agentMessageService.threadWrites.test.js b/backend/__tests__/unit/services/agentMessageService.threadWrites.test.js index c8603e6457..7688d0c149 100644 --- a/backend/__tests__/unit/services/agentMessageService.threadWrites.test.js +++ b/backend/__tests__/unit/services/agentMessageService.threadWrites.test.js @@ -18,6 +18,7 @@ const DMService = require('../../../services/dmService'); const PGMessage = require('../../../models/pg/Message'); const File = require('../../../models/File'); const threadRootResolver = require('../../../services/threadRootResolver'); +const mockDispatch = jest.fn(); jest.mock('../../../models/Message'); jest.mock('../../../models/Summary', () => ({ @@ -88,6 +89,9 @@ jest.mock('../../../services/threadRootResolver', () => { } return { resolveThreadRoot: jest.fn(), ThreadRootError }; }); +jest.mock('../../../services/installable/eventHandlers', () => ({ + dispatch: (...args) => mockDispatch(...args), +})); const POD = '6a0da39bae757028b39f87a6'; let emitted; @@ -165,6 +169,39 @@ describe('agent in-thread posts (threadRootId, no reply edge)', () => { ); }); + it('forwards a relay-only decision card to the outbound dispatcher without persisting it', async () => { + const card = { + title: 'Choose a train', + question: 'Which rollout should ship?', + options: [{ label: 'Canary' }, { label: 'Fast lane' }], + }; + const content = 'Choose a train\n\nWhich rollout should ship?'; + PGMessage.create.mockResolvedValueOnce({ + id: '57602', + content, + message_type: 'text', + created_at: new Date('2026-08-23T00:00:00Z'), + thread_root_id: null, + }); + + await AgentMessageService.postMessage({ + agentName: 'sprint-review', + instanceId: 'default', + podId: POD, + content, + relayCard: card, + }); + + expect(PGMessage.create).toHaveBeenCalledWith( + POD, 'agent-user-1', content, 'text', null, null, null, + ); + expect(mockDispatch).toHaveBeenCalledWith('chat.message', expect.objectContaining({ + podId: POD, + content, + card, + })); + }); + it('a ThreadRootError propagates — never swallowed into the Mongo fallback', async () => { threadRootResolver.resolveThreadRoot.mockRejectedValue( new threadRootResolver.ThreadRootError('root is in another pod', 'thread_root_other_pod'), diff --git a/backend/__tests__/unit/services/channelVerdictService.test.js b/backend/__tests__/unit/services/channelVerdictService.test.js new file mode 100644 index 0000000000..799cc688c8 --- /dev/null +++ b/backend/__tests__/unit/services/channelVerdictService.test.js @@ -0,0 +1,108 @@ +const mockChannelVerdict = { updateOne: jest.fn(), updateMany: jest.fn() }; +jest.mock('../../../models/ChannelVerdict', () => mockChannelVerdict); + +const { + record, markReachedHuman, markRuled, CHANNEL_VERDICT_RETENTION_MS, +} = require('../../../services/channelVerdictService'); + +describe('channelVerdictService', () => { + beforeEach(() => { + jest.clearAllMocks(); + mockChannelVerdict.updateOne.mockResolvedValue({ acknowledged: true }); + mockChannelVerdict.updateMany.mockResolvedValue({ matchedCount: 1 }); + }); + + it('upserts an interrupt receipt on its channel and workspace message', async () => { + const at = new Date('2026-09-05T13:00:00.000Z'); + await record({ + integrationId: 'integration-a', + installationId: 'installation-a', + podId: 'pod-a', + provider: 'telegram', + event: { kind: 'decision_request', podMessageId: '700' }, + verdict: 'interrupt', + reason: 'card', + at, + }); + + expect(mockChannelVerdict.updateOne).toHaveBeenCalledWith( + { integrationId: 'integration-a', 'event.podMessageId': '700' }, + { + $setOnInsert: expect.objectContaining({ + integrationId: 'integration-a', installationId: 'installation-a', podId: 'pod-a', + provider: 'telegram', event: { kind: 'decision_request', podMessageId: '700' }, + verdict: 'interrupt', reason: 'card', at, + }), + }, + { upsert: true }, + ); + expect(mockChannelVerdict.updateOne.mock.calls[0][1].$setOnInsert).not.toHaveProperty('expiresAt'); + }); + + it('expires completed verdicts a quarter after their own event time', async () => { + const at = new Date('2026-09-05T13:00:00.000Z'); + await record({ + integrationId: 'integration-a', podId: 'pod-a', provider: 'telegram', + event: { kind: 'chat.message', podMessageId: '701' }, + verdict: 'hold', reason: 'muted', at, + }); + + expect(mockChannelVerdict.updateOne.mock.calls[0][1].$setOnInsert.expiresAt) + .toEqual(new Date(at.getTime() + CHANNEL_VERDICT_RETENTION_MS)); + }); + + it('stamps only the channel that supplied a ruling', async () => { + mockChannelVerdict.updateOne.mockResolvedValueOnce({ matchedCount: 1 }); + await markReachedHuman({ + integrationId: 'integration-b', podMessageId: '700', ruledVia: 'slack', + }); + + expect(mockChannelVerdict.updateOne).toHaveBeenCalledWith( + { integrationId: 'integration-b', 'event.podMessageId': '700' }, + { $set: expect.objectContaining({ ruledVia: 'slack', reachedHumanAt: expect.any(Date) }) }, + ); + expect(mockChannelVerdict.updateMany).toHaveBeenCalledWith( + { 'event.podMessageId': '700' }, + { $set: expect.objectContaining({ ruledVia: 'slack', expiresAt: expect.any(Date) }) }, + ); + }); + + it('marks every copy ruled without inventing a channel receipt', async () => { + await markRuled({ podMessageId: '700', ruledVia: 'workspace' }); + + expect(mockChannelVerdict.updateMany).toHaveBeenCalledWith( + { 'event.podMessageId': '700' }, + { $set: expect.objectContaining({ ruledVia: 'workspace', expiresAt: expect.any(Date) }) }, + ); + expect(mockChannelVerdict.updateOne).not.toHaveBeenCalled(); + }); + + it('logs a missing channel instead of stamping its sibling copies', async () => { + const warn = jest.spyOn(console, 'warn').mockImplementation(() => {}); + mockChannelVerdict.updateOne.mockResolvedValueOnce({ matchedCount: 0 }); + + await markReachedHuman({ + integrationId: 'expired-integration', podMessageId: 'expired-card', ruledVia: 'telegram', + }); + + expect(mockChannelVerdict.updateMany).not.toHaveBeenCalled(); + expect(warn).toHaveBeenCalledWith( + '[channel-verdict] reach stamp matched no channel:', 'expired-card', + ); + warn.mockRestore(); + }); + + it('keeps a failed observational write out of the bridge outcome', async () => { + const warn = jest.spyOn(console, 'warn').mockImplementation(() => {}); + mockChannelVerdict.updateOne.mockRejectedValueOnce(new Error('database unavailable')); + + await expect(record({ + integrationId: 'integration-a', podId: 'pod-a', provider: 'telegram', + event: { kind: 'decision_request', podMessageId: '700' }, + verdict: 'interrupt', reason: 'card', + })).resolves.toBeUndefined(); + + expect(warn).toHaveBeenCalledWith('[channel-verdict] record failed:', 'database unavailable'); + warn.mockRestore(); + }); +}); diff --git a/backend/__tests__/unit/services/decisionCardRelay.bridges.test.js b/backend/__tests__/unit/services/decisionCardRelay.bridges.test.js new file mode 100644 index 0000000000..e55cbfebbc --- /dev/null +++ b/backend/__tests__/unit/services/decisionCardRelay.bridges.test.js @@ -0,0 +1,209 @@ +jest.mock('../../../models/Integration', () => ({ + findOne: jest.fn(), + findByIdAndUpdate: jest.fn(), +})); +jest.mock('../../../models/Pod', () => ({ findById: jest.fn() })); +jest.mock('../../../services/telegramService', () => ({ sendMessage: jest.fn() })); +jest.mock('../../../services/connectorSecrets', () => ({ get: jest.fn() })); +jest.mock('../../../services/slackApi', () => jest.fn()); +jest.mock('../../../services/connectorRelayPolicy', () => ({ shouldEscalate: jest.fn(() => false) })); +jest.mock('../../../services/channelVerdictService', () => ({ record: jest.fn() })); + +const Integration = require('../../../models/Integration'); +const Pod = require('../../../models/Pod'); +const telegramSend = require('../../../services/telegramService'); +const connectorSecrets = require('../../../services/connectorSecrets'); +const SlackApi = require('../../../services/slackApi'); +const { shouldEscalate } = require('../../../services/connectorRelayPolicy'); +const verdicts = require('../../../services/channelVerdictService'); +const { + relayAgentMessageToTelegram, + renderTelegramDecisionCard, +} = require('../../../services/telegramBridgeService'); +const { + relayAgentMessageToSlack, + renderSlackDecisionCard, +} = require('../../../services/slackBridgeService'); + +const CARD = { + title: 'Choose the rollout & safely', + question: 'Should we take the low-risk path?', + options: [ + { label: 'Later', description: 'L'.repeat(280) }, + { label: 'Canary ', description: 'C'.repeat(280), recommended: true }, + { label: 'Full rollout', description: 'F'.repeat(280) }, + { label: 'Pause', description: 'P'.repeat(280) }, + ], + context: 'This private context must never cross the bridge.', +}; + +const telegramIntegration = (config = {}) => ({ + _id: 'telegram-1', + installationId: 'install-telegram', + podId: 'pod-1', + scope: 'user', + type: 'telegram', + isActive: true, + config: { + liveRelay: true, + chatType: 'private', + chatId: 'telegram-chat', + gates: { 'pod-1': { enabled: true } }, + ...config, + }, +}); + +const slackIntegration = (config = {}) => ({ + _id: 'slack-1', + installationId: 'install-slack', + podId: 'pod-1', + scope: 'user', + type: 'slack', + isActive: true, + config: { + liveRelay: true, + chatType: 'im', + chatId: 'slack-chat', + teamId: 'T1', + botTokenRef: 'slack-secret', + gates: { 'pod-1': { enabled: true } }, + ...config, + }, +}); + +const cardSend = (integration, card = CARD) => ({ + podId: 'pod-1', + agentUsername: 'release-agent', + displayName: 'Release Agent', + content: 'The workspace copy remains ordinary text.', + podMessageId: '731', + card, + integration, +}); + +describe('decision-card outbound bridges', () => { + beforeEach(() => { + jest.clearAllMocks(); + process.env.TELEGRAM_BOT_TOKEN = 'telegram-token'; + process.env.PUBLIC_APP_URL = 'https://commonly.test'; + telegramSend.sendMessage.mockResolvedValue({ messageId: 42 }); + connectorSecrets.get.mockResolvedValue('xoxb-token'); + SlackApi.mockImplementation(() => ({ postMessage: jest.fn().mockResolvedValue({ ok: true, ts: '1700.0001' }) })); + Pod.findById.mockReturnValue({ select: jest.fn().mockReturnValue({ lean: jest.fn().mockResolvedValue({ name: 'Launch week' }) }) }); + Integration.findByIdAndUpdate.mockResolvedValue(undefined); + verdicts.record.mockResolvedValue(undefined); + }); + + afterAll(() => { + delete process.env.TELEGRAM_BOT_TOKEN; + delete process.env.PUBLIC_APP_URL; + }); + + test('Telegram sends a card as an interrupt without consulting prose escalation and records it once', async () => { + await relayAgentMessageToTelegram(cardSend(telegramIntegration())); + + expect(shouldEscalate).not.toHaveBeenCalled(); + expect(telegramSend.sendMessage).toHaveBeenCalledTimes(1); + const text = telegramSend.sendMessage.mock.calls[0][2]; + expect(text.length).toBeLessThanOrEqual(1900); + expect(text).toContain('Release Agent needs a ruling'); + expect(text).toContain('<a href="https://untrusted.test">the rollout</a> & safely'); + expect(text).not.toContain('This private context must never cross the bridge.'); + expect(text.indexOf('1. Later')).toBeLessThan(text.indexOf('2. Canary <fast>')); + expect(text.indexOf('2. Canary <fast>')).toBeLessThan(text.indexOf('3. Full rollout')); + expect((text.match(/★ recommended/g) || [])).toHaveLength(1); + expect(text).toContain(`${'L'.repeat(99)}…`); + expect(text).toContain('?message=731'); + expect(verdicts.record).toHaveBeenCalledWith(expect.objectContaining({ + integrationId: 'telegram-1', + provider: 'telegram', + event: { kind: 'decision_request', podMessageId: '731' }, + verdict: 'interrupt', + reason: 'card', + })); + }); + + test('Slack sends the same ordered card with mrkdwn-safe fields', async () => { + await relayAgentMessageToSlack(cardSend(slackIntegration())); + + expect(shouldEscalate).not.toHaveBeenCalled(); + const api = SlackApi.mock.results[0].value; + const text = api.postMessage.mock.calls[0][1]; + expect(text.length).toBeLessThanOrEqual(1900); + expect(text).toContain('*Release Agent needs a ruling · Choose <a href="https://untrusted.test">the rollout</a> & safely*'); + expect(text).toContain('2. *Canary <fast>* ★ recommended'); + expect(text).not.toContain('This private context must never cross the bridge.'); + expect(text).toContain(`${'L'.repeat(99)}…`); + expect(verdicts.record).toHaveBeenCalledWith(expect.objectContaining({ + integrationId: 'slack-1', provider: 'slack', verdict: 'interrupt', reason: 'card', + })); + }); + + test('Slack escapes every agent-authored card field before rendering mrkdwn', () => { + const hostile = ' &'; + const text = renderSlackDecisionCard({ + card: { + title: `${hostile} title`, + question: `${hostile} question`, + options: [{ label: `${hostile} label`, description: `${hostile} description` }], + }, + displayName: `${hostile} agent`, + agentUsername: 'release-agent', + link: 'https://commonly.test/v2/pods/pod-1?message=731', + }); + + expect(text).not.toContain(''); + expect(text.match(/<https:\/\/evil\.example\|open in Commonly> &/g)).toHaveLength(5); + expect(text).toContain('https://commonly.test/v2/pods/pod-1?message=731'); + }); + + test.each([ + ['Telegram', 'muted', relayAgentMessageToTelegram, telegramIntegration, { relayMutedUntil: new Date(Date.now() + 60_000) }], + ['Telegram', 'paused', relayAgentMessageToTelegram, telegramIntegration, { adminPause: { reason: 'Safety review' } }], + ['Telegram', 'gate_off', relayAgentMessageToTelegram, telegramIntegration, { gates: { 'pod-1': { enabled: false } } }], + ['Slack', 'muted', relayAgentMessageToSlack, slackIntegration, { relayMutedUntil: new Date(Date.now() + 60_000) }], + ['Slack', 'paused', relayAgentMessageToSlack, slackIntegration, { adminPause: { reason: 'Safety review' } }], + ['Slack', 'gate_off', relayAgentMessageToSlack, slackIntegration, { gates: { 'pod-1': { enabled: false } } }], + ])('%s holds a card when %s', async (_provider, reason, relay, integrationFor, config) => { + await relay(cardSend(integrationFor(config))); + + expect(telegramSend.sendMessage).not.toHaveBeenCalled(); + expect(SlackApi).not.toHaveBeenCalled(); + expect(verdicts.record).toHaveBeenCalledWith(expect.objectContaining({ verdict: 'hold', reason })); + }); + + test('keeps ordinary Telegram and Slack relay content at 900 characters', async () => { + const ordinary = `[DECISION] ${'x'.repeat(1_000)}`; + shouldEscalate.mockReturnValue(true); + await relayAgentMessageToTelegram({ ...cardSend(telegramIntegration()), card: undefined, content: ordinary }); + await relayAgentMessageToSlack({ ...cardSend(slackIntegration()), card: undefined, content: ordinary }); + + const telegramText = telegramSend.sendMessage.mock.calls[0][2]; + expect(telegramText).toContain('x'.repeat(889)); + expect(telegramText).not.toContain('x'.repeat(890)); + const slackText = SlackApi.mock.results[0].value.postMessage.mock.calls[0][1]; + expect(slackText).toBe(`[Launch week] Release Agent: ${ordinary.slice(0, 900)}`); + }); + + test('keeps a maximum valid card below 1900 without trimming its title, question, or labels', () => { + const maximum = { + title: 'T'.repeat(160), + question: 'Q'.repeat(1000), + options: Array.from({ length: 4 }, (_, index) => ({ + label: String(index + 1).repeat(80), description: 'D'.repeat(280), + })), + }; + const telegram = renderTelegramDecisionCard({ card: maximum, displayName: 'Agent', agentUsername: 'agent', link: 'https://commonly.test/link' }); + const slack = renderSlackDecisionCard({ card: maximum, displayName: 'Agent', agentUsername: 'agent', link: 'https://commonly.test/link' }); + + for (const text of [telegram, slack]) { + expect(text.length).toBeLessThanOrEqual(1900); + expect(text).toContain(maximum.title); + expect(text).toContain(maximum.question); + maximum.options.forEach((option, index) => { + expect(text).toContain(`${index + 1}.`); + expect(text).toContain(option.label); + }); + } + }); +}); diff --git a/backend/__tests__/unit/services/decisionRequestService.test.js b/backend/__tests__/unit/services/decisionRequestService.test.js index 93fd1f6c84..3aafcc3c52 100644 --- a/backend/__tests__/unit/services/decisionRequestService.test.js +++ b/backend/__tests__/unit/services/decisionRequestService.test.js @@ -81,8 +81,18 @@ describe('DecisionRequestService', () => { expect(mockPostMessage).toHaveBeenCalledWith(expect.objectContaining({ agentName: 'release-agent', instanceId: 'seat-1', podId: 'pod-1', threadRootId: '612', metadata: { source: 'decision-request' }, + relayCard: { + title: 'Choose the release train', + question: 'Which rollout should I run?', + options: [ + { label: 'Canary', description: 'Small cohort.', recommended: true }, + { label: 'Fast lane', description: 'Ship once green.' }, + ], + context: 'The branch is green.', + }, })); expect(mockPostMessage.mock.calls[0][0].content).toContain('Choose the release train'); + expect(mockPostMessage.mock.calls[0][0].payload).toBeUndefined(); expect(mockDecision.create).toHaveBeenCalledWith(expect.objectContaining({ agentUserId: 'agent-user-1', agentName: 'release-agent', instanceId: 'seat-1', decisionClass: 'implementation', messageId: '700', status: 'pending', diff --git a/backend/__tests__/unit/services/installableEventHandlers.test.js b/backend/__tests__/unit/services/installableEventHandlers.test.js index 00f70d4a7e..e6afe30185 100644 --- a/backend/__tests__/unit/services/installableEventHandlers.test.js +++ b/backend/__tests__/unit/services/installableEventHandlers.test.js @@ -98,6 +98,38 @@ describe('installable event dispatcher', () => { expect(String(relay.mock.calls[0][0].integration.podId)).toBe(podA); }); + it('forwards a server-composed decision card without changing the chat content', async () => { + const podId = freshId(); + const ownerId = freshId(); + await createPod(podId, ownerId); + await install({ installableId: 'telegram', installedBy: ownerId, podId }); + const relay = jest.fn().mockResolvedValue(undefined); + eventHandlers['telegram.relay'] = relay; + const card = { + title: 'Choose a train', + question: 'Which rollout should ship?', + options: [ + { label: 'Canary' }, + { label: 'Fast lane', recommended: true }, + ], + context: 'Private workspace detail', + }; + + await dispatch('chat.message', { + podId, + agentUsername: 'kai', + displayName: 'Kai', + content: 'Choose a train\n\nWhich rollout should ship?', + podMessageId: 'message-card', + card, + }); + + expect(relay).toHaveBeenCalledWith(expect.objectContaining({ + content: 'Choose a train\n\nWhich rollout should ship?', + card, + })); + }); + it('relays each same-pod connector to its own selected Telegram chat', async () => { const podId = freshId(); const firstOwner = freshId(); @@ -144,6 +176,41 @@ describe('installable event dispatcher', () => { } }); + it('fans a decision card to every member connector with an enabled gate', async () => { + const podId = freshId(); + const firstOwner = freshId(); + const secondOwner = freshId(); + await createPod(podId, firstOwner, [secondOwner]); + const [first, second] = await Promise.all([ + install({ installableId: 'telegram', installedBy: firstOwner, podId }), + install({ installableId: 'telegram', installedBy: secondOwner, podId }), + ]); + await Integration.updateOne( + { _id: second.integration._id }, + { $set: { [`config.gates.${podId}.enabled`]: true } }, + ); + const relay = jest.fn().mockResolvedValue(undefined); + eventHandlers['telegram.relay'] = relay; + + await dispatch('chat.message', { + podId, + agentUsername: 'kai', + displayName: 'Kai', + content: 'The workspace ask remains plain text.', + podMessageId: 'message-card-fanout', + card: { + title: 'Choose a rollout', + question: 'Which path should ship?', + options: [{ label: 'Canary' }, { label: 'Fast lane' }], + }, + }); + + expect(relay).toHaveBeenCalledTimes(2); + expect(new Set(relay.mock.calls.map(([call]) => String(call.integration._id)))).toEqual( + new Set([String(first.integration._id), String(second.integration._id)]), + ); + }); + it('selects a user connector only for an enabled gate whose owner remains a member', async () => { const ownerId = freshId(); const otherOwnerId = freshId(); @@ -191,6 +258,42 @@ describe('installable event dispatcher', () => { expect(relay).toHaveBeenCalledTimes(2); }); + it('still selects a muted, paused, or gate-off card binding so the bridge can record its hold verdict', async () => { + const podId = freshId(); + const ownerId = freshId(); + await createPod(podId, ownerId); + const installed = await install({ installableId: 'telegram', installedBy: ownerId, podId }); + await Integration.updateOne( + { _id: installed.integration._id }, + { + $set: { + [`config.gates.${podId}.enabled`]: false, + 'config.adminPause': { reason: 'Safety review', at: new Date() }, + }, + }, + ); + const relay = jest.fn().mockResolvedValue(undefined); + eventHandlers['telegram.relay'] = relay; + + await dispatch('chat.message', { + podId, + agentUsername: 'kai', + displayName: 'Kai', + content: 'Workspace content stays unchanged', + podMessageId: 'message-card-hold', + card: { + title: 'Choose a rollout', + question: 'Which path should ship?', + options: [{ label: 'Canary' }, { label: 'Fast lane' }], + }, + }); + + expect(relay).toHaveBeenCalledWith(expect.objectContaining({ + card: expect.objectContaining({ title: 'Choose a rollout' }), + integration: expect.objectContaining({ _id: expect.anything() }), + })); + }); + it('does not invoke a handler when the event pod has no installation', async () => { const relay = jest.fn().mockResolvedValue(undefined); eventHandlers['telegram.relay'] = relay; diff --git a/backend/models/ChannelVerdict.ts b/backend/models/ChannelVerdict.ts new file mode 100644 index 0000000000..91a4e1b175 --- /dev/null +++ b/backend/models/ChannelVerdict.ts @@ -0,0 +1,87 @@ +import mongoose, { Document, Model, Schema, Types } from 'mongoose'; + +/** + * One bridge verdict for one connector binding. This is deliberately neither + * an AttentionItem (which is a recipient queue) nor an AuditLog (which is an + * administrator action): it records what a particular channel did with a + * workspace event. + */ +export const CHANNEL_VERDICT_PROVIDERS = ['telegram', 'slack'] as const; +export type ChannelVerdictProvider = typeof CHANNEL_VERDICT_PROVIDERS[number]; + +export const CHANNEL_VERDICTS = ['interrupt', 'digest', 'hold'] as const; +export type ChannelVerdictKind = typeof CHANNEL_VERDICTS[number]; + +export const CHANNEL_VERDICT_EVENT_KINDS = ['decision_request', 'chat.message'] as const; +export type ChannelVerdictEventKind = typeof CHANNEL_VERDICT_EVENT_KINDS[number]; + +export const CHANNEL_VERDICT_RULED_VIA = ['telegram', 'slack', 'workspace'] as const; +export type ChannelVerdictRuledVia = typeof CHANNEL_VERDICT_RULED_VIA[number]; + +export interface IChannelVerdict extends Document { + integrationId: Types.ObjectId; + installationId?: Types.ObjectId; + podId: Types.ObjectId; + provider: ChannelVerdictProvider; + event: { + kind: ChannelVerdictEventKind; + podMessageId: string; + decisionId?: Types.ObjectId; + }; + verdict: ChannelVerdictKind; + reason: string; + at: Date; + reachedHumanAt?: Date; + ruledVia?: ChannelVerdictRuledVia; + expiresAt?: Date; +} + +const channelVerdictEventSchema = new Schema( + { + kind: { type: String, required: true, enum: CHANNEL_VERDICT_EVENT_KINDS }, + podMessageId: { type: String, required: true, trim: true, maxlength: 128 }, + decisionId: { type: Schema.Types.ObjectId, ref: 'DecisionRequest' }, + }, + { _id: false }, +); + +const channelVerdictSchema = new Schema( + { + integrationId: { type: Schema.Types.ObjectId, ref: 'Integration', required: true }, + installationId: { type: Schema.Types.ObjectId, ref: 'InstallableInstallation' }, + podId: { type: Schema.Types.ObjectId, ref: 'Pod', required: true }, + provider: { type: String, required: true, enum: CHANNEL_VERDICT_PROVIDERS }, + event: { type: channelVerdictEventSchema, required: true }, + verdict: { type: String, required: true, enum: CHANNEL_VERDICTS }, + // Reasons are deliberately extensible: ADR-029's later writers use the + // same collection rather than adding one model per relay policy. + reason: { type: String, required: true, trim: true, maxlength: 80 }, + at: { type: Date, required: true, default: Date.now }, + reachedHumanAt: { type: Date }, + ruledVia: { type: String, enum: CHANNEL_VERDICT_RULED_VIA }, + // Open decision cards stay addressable until the card itself is settled. + // All other rows set this at record time; the two settlement paths set it + // for every copy of a decision card. + expiresAt: { type: Date }, + }, + { collection: 'channel_verdicts' }, +); + +channelVerdictSchema.index({ integrationId: 1, at: -1 }); +channelVerdictSchema.index({ podId: 1, at: -1 }); +channelVerdictSchema.index({ 'event.podMessageId': 1 }); +// Finished facts remain visible for one quarter. An open decision card has no +// expiry until its ruling path sets one, so the ledger cannot forget the +// channel receipt before the decision itself can finish. +channelVerdictSchema.index({ expiresAt: 1 }, { expireAfterSeconds: 0 }); +// A connector can retry a relay after a transient failure. Its ledger fact is +// still one fact per channel and workspace message, even if competing workers +// race the retry; the service upserts on this same key. +channelVerdictSchema.index({ integrationId: 1, 'event.podMessageId': 1 }, { unique: true }); + +const ChannelVerdict: Model = (mongoose.models.ChannelVerdict as Model) + || mongoose.model('ChannelVerdict', channelVerdictSchema); + +export default ChannelVerdict; +// eslint-disable-next-line @typescript-eslint/no-require-imports +module.exports = ChannelVerdict; diff --git a/backend/services/agentMessageService.ts b/backend/services/agentMessageService.ts index a37777e808..2a87933ee1 100644 --- a/backend/services/agentMessageService.ts +++ b/backend/services/agentMessageService.ts @@ -1,3 +1,5 @@ +import type { DecisionRelayCard } from './decisionCardRelay'; + // eslint-disable-next-line global-require const socketConfig = require('../config/socket'); // eslint-disable-next-line global-require @@ -164,6 +166,9 @@ interface PostMessageOptions { // sanitizeAgentContent by design — content stays the sanitized fallback // text; payload is server-composed, never model prose. payload?: unknown; + // Relay-only structured data composed by a trusted server workflow. It is + // not persisted on the pod message or exposed to arbitrary runtime output. + relayCard?: DecisionRelayCard | null; instanceId?: string; displayName?: string; replyToMessageId?: string | null; @@ -182,6 +187,7 @@ interface PostTargetOptions { messageType?: string; metadata?: MetadataDoc; payload?: unknown; + relayCard?: DecisionRelayCard | null; agentUser: AgentUserDoc; displayName?: string; replyToMessageId?: string | null; @@ -969,6 +975,7 @@ class AgentMessageService { metadata = {}, messageType = 'text', payload = null, + relayCard = null, instanceId = 'default', displayName, replyToMessageId = null, @@ -1506,6 +1513,7 @@ class AgentMessageService { messageType, metadata, payload, + relayCard, agentUser, displayName, replyToMessageId, @@ -1565,8 +1573,20 @@ class AgentMessageService { static async _postToTarget(options: PostTargetOptions): Promise<{ message: MessageNormalized; summary: unknown }> { const { - agentName, instanceId = 'default', podId, content, messageType = 'text', - metadata = {}, payload = null, agentUser, displayName, replyToMessageId = null, threadRootId = null, skipDeliveryUpdate = false, skipSummaryPersistence = false, + agentName, + instanceId = 'default', + podId, + content, + messageType = 'text', + metadata = {}, + payload = null, + relayCard = null, + agentUser, + displayName, + replyToMessageId = null, + threadRootId = null, + skipDeliveryUpdate = false, + skipSummaryPersistence = false, } = options; // The installation label belongs to this pod; the User label belongs to @@ -1792,6 +1812,7 @@ class AgentMessageService { displayName: senderDisplayName || bridgeUsername, content: String((message as MessageNormalized)?.content ?? ''), podMessageId: (message as MessageNormalized)?._id || (message as MessageNormalized)?.id || null, + ...(relayCard ? { card: relayCard } : {}), }); } catch (bridgeError) { console.warn('[tg-bridge] outbound hook failed:', (bridgeError as Error).message); diff --git a/backend/services/channelVerdictService.ts b/backend/services/channelVerdictService.ts new file mode 100644 index 0000000000..8984faaddf --- /dev/null +++ b/backend/services/channelVerdictService.ts @@ -0,0 +1,143 @@ +import type { + ChannelVerdictEventKind, + ChannelVerdictKind, + ChannelVerdictProvider, + ChannelVerdictRuledVia, +} from '../models/ChannelVerdict'; + +// eslint-disable-next-line @typescript-eslint/no-require-imports +const ChannelVerdict = require('../models/ChannelVerdict'); + +export const CHANNEL_VERDICT_RETENTION_MS = 90 * 24 * 60 * 60 * 1000; + +export interface RecordChannelVerdict { + integrationId: unknown; + installationId?: unknown; + podId: unknown; + provider: ChannelVerdictProvider; + event: { + kind: ChannelVerdictEventKind; + podMessageId: string; + decisionId?: unknown; + }; + verdict: ChannelVerdictKind; + reason: string; + at?: Date; +} + +export interface MarkReachedHuman { + // The same decision card can reach several member-owned channels. This is + // mandatory so one person's reply never claims those sibling channels did + // the reaching. + integrationId: unknown; + podMessageId: string; + ruledVia: Extract; +} + +export interface MarkRuled { + podMessageId: string; + ruledVia: ChannelVerdictRuledVia; +} + +const matchedCount = (result: { matchedCount?: number; n?: number } | null | undefined): number => ( + Number(result?.matchedCount ?? result?.n ?? 0) +); + +const expiryAfter = (at: Date): Date => new Date(at.getTime() + CHANNEL_VERDICT_RETENTION_MS); + +const isOpenDecisionCard = (entry: RecordChannelVerdict): boolean => ( + entry.event.kind === 'decision_request' + && entry.verdict === 'interrupt' + && entry.reason === 'card' +); + +/** + * Writes are observational. A bridge has already sent (or deliberately held) + * the card before this is called, so a ledger failure must never change relay + * delivery. Upserting the model's unique integration/message key makes a + * retry one durable fact rather than a second card receipt. + */ +export const record = async (entry: RecordChannelVerdict): Promise => { + try { + const at = entry.at || new Date(); + await ChannelVerdict.updateOne( + { + integrationId: entry.integrationId, + 'event.podMessageId': entry.event.podMessageId, + }, + { + $setOnInsert: { + integrationId: entry.integrationId, + ...(entry.installationId ? { installationId: entry.installationId } : {}), + podId: entry.podId, + provider: entry.provider, + event: entry.event, + verdict: entry.verdict, + reason: entry.reason, + at, + // A card does not become a historical fact until it is ruled. The + // same model also records completed relays and holds, which expire + // a quarter after their original event time. + ...(!isOpenDecisionCard(entry) ? { expiresAt: expiryAfter(at) } : {}), + }, + }, + { upsert: true }, + ); + } catch (error) { + console.warn('[channel-verdict] record failed:', (error as Error).message); + } +}; + +/** + * Record a channel-originated ruling. The selector is intentionally scoped to + * the channel binding, not only the workspace message, because a card may + * fan out to several gated connector owners. + */ +export const markReachedHuman = async (entry: MarkReachedHuman): Promise => { + try { + const now = new Date(); + const reached = await ChannelVerdict.updateOne( + { + integrationId: entry.integrationId, + 'event.podMessageId': entry.podMessageId, + }, + { + $set: { reachedHumanAt: now, ruledVia: entry.ruledVia }, + }, + ); + if (!matchedCount(reached)) { + console.warn('[channel-verdict] reach stamp matched no channel:', entry.podMessageId); + return; + } + + // This is a fork-level fact, unlike reachedHumanAt: every gated recipient + // sees how the decision settled, but only the replying channel records + // that its own human performed the action. + await ChannelVerdict.updateMany( + { 'event.podMessageId': entry.podMessageId }, + { $set: { ruledVia: entry.ruledVia, expiresAt: expiryAfter(now) } }, + ); + } catch (error) { + console.warn('[channel-verdict] reach stamp failed:', (error as Error).message); + } +}; + +/** + * A workspace ruling settles every channel copy without claiming that any + * channel human was reached. PR 3 calls this path; its counterpart above is + * deliberately narrower for a Telegram or Slack ruling. + */ +export const markRuled = async (entry: MarkRuled): Promise => { + try { + const now = new Date(); + const ruled = await ChannelVerdict.updateMany( + { 'event.podMessageId': entry.podMessageId }, + { $set: { ruledVia: entry.ruledVia, expiresAt: expiryAfter(now) } }, + ); + if (!matchedCount(ruled)) { + console.warn('[channel-verdict] ruling stamp matched no channels:', entry.podMessageId); + } + } catch (error) { + console.warn('[channel-verdict] ruling stamp failed:', (error as Error).message); + } +}; diff --git a/backend/services/decisionCardRelay.ts b/backend/services/decisionCardRelay.ts new file mode 100644 index 0000000000..0ee431f469 --- /dev/null +++ b/backend/services/decisionCardRelay.ts @@ -0,0 +1,20 @@ +/** + * A decision card is a relay-only view of a server-validated decision + * request. It crosses the outbound dispatch seam, but deliberately never + * becomes part of the workspace message payload. + */ +export interface DecisionCardOption { + label: string; + description?: string; + recommended?: boolean; +} + +export interface DecisionRelayCard { + // The request is posted before its DecisionRequest row exists. Channel + // replies bind through the persisted pod message id, not this optional id. + decisionId?: string; + title: string; + question: string; + options: DecisionCardOption[]; + context?: string; +} diff --git a/backend/services/decisionRequestService.ts b/backend/services/decisionRequestService.ts index 9c9429aeec..1f4b60aa0a 100644 --- a/backend/services/decisionRequestService.ts +++ b/backend/services/decisionRequestService.ts @@ -1,4 +1,5 @@ import crypto from 'crypto'; +import type { DecisionCardOption, DecisionRelayCard } from './decisionCardRelay'; // eslint-disable-next-line global-require const DecisionRequest = require('../models/DecisionRequest'); @@ -35,11 +36,7 @@ export class DecisionRequestError extends Error { } } -export interface DecisionOptionInput { - label: string; - description?: string; - recommended?: boolean; -} +export type DecisionOptionInput = DecisionCardOption; export type DecisionClass = 'strategy' | 'implementation' | 'prioritization'; const ADVISORY_DECISION_CLASSES: readonly DecisionClass[] = [ @@ -182,6 +179,16 @@ export const requestDecision = async (input: RequestDecisionOptions): Promise ( && typeof eventHandlers[reference.slice('internal:'.length)] === 'function' ); -const activeHandlersForPod = async (podId: string): Promise> => { @@ -55,26 +57,48 @@ const activeHandlersForPod = async (podId: string): Promise String(id)))).map((id) => new Types.ObjectId(id)); if (!memberIds.length) return []; const gateEnabledPath = `config.gates.${String(podId)}.enabled`; + // A decision card must record why it was deliberately withheld. Ordinary + // chat retains the narrow selector below; cards additionally reach paused + // and gate-off member bindings so their bridge can write a `hold` verdict + // without sending anything to the channel. + const selection = includeCardHolds + ? { + type: { $in: ['telegram', 'slack'] }, + isActive: true, + status: { $ne: 'error' }, + 'config.liveRelay': true, + $or: [ + { + scope: { $ne: 'user' }, + podId: new Types.ObjectId(podId), + }, + { + scope: 'user', + createdBy: { $in: memberIds }, + }, + ], + } + : { + type: { $in: ['telegram', 'slack'] }, + isActive: true, + status: { $ne: 'error' }, + 'config.liveRelay': true, + 'config.adminPause': { $exists: false }, + $or: [ + { + scope: { $ne: 'user' }, + podId: new Types.ObjectId(podId), + }, + { + scope: 'user', + createdBy: { $in: memberIds }, + [gateEnabledPath]: true, + }, + ], + }; return Integration.aggregate([ { - $match: { - type: { $in: ['telegram', 'slack'] }, - isActive: true, - status: { $ne: 'error' }, - 'config.liveRelay': true, - 'config.adminPause': { $exists: false }, - $or: [ - { - scope: { $ne: 'user' }, - podId: new Types.ObjectId(podId), - }, - { - scope: 'user', - createdBy: { $in: memberIds }, - [gateEnabledPath]: true, - }, - ], - }, + $match: selection, }, { $lookup: { @@ -158,7 +182,7 @@ export const dispatch = async ( let selected: Array<{ integration: IIntegration; handler: string }>; try { - selected = await activeHandlersForPod(payload.podId); + selected = await activeHandlersForPod(payload.podId, Boolean(payload.card)); } catch (error) { console.warn('[installable-dispatch] selector failed:', (error as Error).message); return; diff --git a/backend/services/slackBridgeService.ts b/backend/services/slackBridgeService.ts index f0829aeb0c..19f9719e0c 100644 --- a/backend/services/slackBridgeService.ts +++ b/backend/services/slackBridgeService.ts @@ -10,9 +10,13 @@ const isPodMember = require('../utils/isPodMember'); const connectorSecrets = require('./connectorSecrets'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { shouldEscalate } = require('./connectorRelayPolicy'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const channelVerdictService = require('./channelVerdictService'); +import type { DecisionRelayCard } from './decisionCardRelay'; const RELAY_MAP_CAP = 100; -const OUTBOUND_TEXT_CAP = 2_800; +const OUTBOUND_TEXT_CAP = 900; +const CARD_OUTBOUND_TEXT_CAP = 1_900; interface SlackIntegrationDoc { _id: unknown; @@ -60,6 +64,52 @@ const isRelayableIntegration = (integration: SlackIntegrationDoc, podId: string) && Boolean(integration.config?.botTokenRef) ); +const truncateWithEllipsis = (value: string, limit: number): string => { + if (value.length <= limit) return value; + if (limit <= 1) return '…'.slice(0, limit); + return `${value.slice(0, limit - 1)}…`; +}; + +// Slack mrkdwn treats these as control characters: escaping keeps agent-authored +// card fields from creating links, mentions, or other markup in a human's DM. +const escapeSlackMrkdwn = (raw: string): string => String(raw) + .replace(/&/g, '&') + .replace(//g, '>'); + +export const renderSlackDecisionCard = (opts: { + card: DecisionRelayCard; + displayName: string; + agentUsername: string; + link: string; +}): string => { + const { card, displayName, agentUsername, link } = opts; + const footer = `\n\nReply to this message with a number, or write your ruling.\nopen in Commonly → ${link}`; + const optionLines = card.options.map((option, index) => ( + `\n${index + 1}. *${escapeSlackMrkdwn(option.label)}*${option.recommended ? ' ★ recommended' : ''}` + )); + const title = escapeSlackMrkdwn(card.title); + const question = escapeSlackMrkdwn(card.question); + const agent = escapeSlackMrkdwn(displayName || agentUsername); + const header = `*${agent} needs a ruling · ${title}*\n${question}\n`; + let remaining = CARD_OUTBOUND_TEXT_CAP - header.length - footer.length + - optionLines.reduce((sum, line) => sum + line.length, 0); + const descriptions = card.options.map((option) => ( + option.description ? escapeSlackMrkdwn(truncateWithEllipsis(option.description, 100)) : '' + )); + const descriptionLines = descriptions.map((description, index) => { + if (!description || remaining <= 4) return ''; + const nonEmptyRemaining = descriptions.slice(index).filter(Boolean).length; + const allowance = Math.max(0, Math.floor(remaining / Math.max(1, nonEmptyRemaining)) - 4); + if (!allowance) return ''; + const fitted = truncateWithEllipsis(description, allowance); + const line = `\n ${fitted}`; + remaining -= line.length; + return line; + }); + return header + optionLines.map((line, index) => `${line}${descriptionLines[index]}`).join('') + footer; +}; + // The enabled gate is only the pod → connector subscription. Inbound follows // the owner's active pod selection and validates membership at receive time. const isInboundRelayableIntegration = (integration: SlackIntegrationDoc, podId: string): boolean => ( @@ -112,6 +162,7 @@ export const relayAgentMessageToSlack = async (opts: { displayName: string; content: string; podMessageId?: string | null; + card?: DecisionRelayCard; integration?: SlackIntegrationDoc; }): Promise => { const { @@ -119,17 +170,58 @@ export const relayAgentMessageToSlack = async (opts: { } = opts; try { const integration = opts.integration ?? await findLiveIntegration(podId); - if (!integration || !isRelayableIntegration(integration, podId)) return; + if (!integration) return; + const cardHoldReason = opts.card + ? (integration.config?.adminPause + ? 'paused' + : integration.scope === 'user' && integration.config?.gates?.[podId]?.enabled !== true + ? 'gate_off' + : null) + : null; + if (cardHoldReason) { + if (podMessageId) { + await channelVerdictService.record({ + integrationId: integration._id, + ...(integration.installationId ? { installationId: integration.installationId } : {}), + podId, + provider: 'slack', + event: { kind: 'decision_request', podMessageId }, + verdict: 'hold', + reason: cardHoldReason, + }); + } + return; + } + if (!isRelayableIntegration(integration, podId)) return; const mutedUntil = integration.config?.relayMutedUntil; - if (mutedUntil && new Date(mutedUntil) > new Date()) return; - if (!shouldEscalate({ content, agentUsername, integration, podId })) return; + if (mutedUntil && new Date(mutedUntil) > new Date()) { + if (opts.card && podMessageId) { + await channelVerdictService.record({ + integrationId: integration._id, + ...(integration.installationId ? { installationId: integration.installationId } : {}), + podId, + provider: 'slack', + event: { kind: 'decision_request', podMessageId }, + verdict: 'hold', + reason: 'muted', + }); + } + return; + } + if (!opts.card && !shouldEscalate({ content, agentUsername, integration, podId })) return; const [pod, token] = await Promise.all([ Pod.findById(podId).select('name').lean(), connectorSecrets.get(String(integration.config!.botTokenRef)), ]); const podName = String(pod?.name || 'Commonly'); - const text = `[${podName}] ${displayName || agentUsername}: ${String(content).slice(0, OUTBOUND_TEXT_CAP)}`; + const base = (process.env.PUBLIC_APP_URL || 'https://commonly.me').replace(/\/$/, ''); + const link = opts.card && podMessageId + ? `${base}/v2/pods/${encodeURIComponent(podId)}?message=${encodeURIComponent(podMessageId)}` + : `${base}/v2/pods/${podId}`; + const text = opts.card + ? renderSlackDecisionCard({ card: opts.card, displayName, agentUsername, link }) + : `[${podName}] ${displayName || agentUsername}: ${String(content).slice(0, OUTBOUND_TEXT_CAP)}`; const result = await new SlackApi(token).postMessage(String(integration.config!.chatId), text); if (!result.ok || !result.ts) { throw new Error(`chat.postMessage failed: ${String(result.error || 'unknown error')}`); @@ -144,6 +236,17 @@ export const relayAgentMessageToSlack = async (opts: { }, }, }); + if (opts.card && podMessageId) { + await channelVerdictService.record({ + integrationId: integration._id, + ...(integration.installationId ? { installationId: integration.installationId } : {}), + podId, + provider: 'slack', + event: { kind: 'decision_request', podMessageId }, + verdict: 'interrupt', + reason: 'card', + }); + } } catch (error) { const config = opts.integration?.config; console.warn( @@ -309,6 +412,7 @@ module.exports = { relayAgentMessageToSlack, relaySlackMessageToPod, routeSlackReplyContent, + renderSlackDecisionCard, isRelayableIntegration, isInboundRelayableIntegration, }; diff --git a/backend/services/telegramBridgeService.ts b/backend/services/telegramBridgeService.ts index 2dd47ebfe5..79ea386ee1 100644 --- a/backend/services/telegramBridgeService.ts +++ b/backend/services/telegramBridgeService.ts @@ -26,9 +26,13 @@ const isPodMember = require('../utils/isPodMember'); const telegramSend = require('./telegramService'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { shouldEscalate } = require('./connectorRelayPolicy'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const channelVerdictService = require('./channelVerdictService'); +import type { DecisionRelayCard } from './decisionCardRelay'; const RELAY_MAP_CAP = 100; const OUTBOUND_TEXT_CAP = 900; +const CARD_OUTBOUND_TEXT_CAP = 1_900; export interface RelayMapEntry { tgMessageId: string; @@ -38,6 +42,7 @@ export interface RelayMapEntry { interface TelegramIntegrationDoc { _id: unknown; + installationId?: unknown; podId: unknown; scope?: 'pod' | 'user'; type?: string; @@ -62,6 +67,53 @@ const escapeHtml = (raw: string): string => String(raw) export { shouldEscalate }; +const truncateWithEllipsis = (value: string, limit: number): string => { + if (value.length <= limit) return value; + if (limit <= 1) return '…'.slice(0, limit); + return `${value.slice(0, limit - 1)}…`; +}; + +/** + * Decision cards have their own cap: ordinary relay messages remain bounded + * at 900 characters. Keep the title, question, and option labels intact; if + * a valid maximum-size card needs space, descriptions yield first. + */ +export const renderTelegramDecisionCard = (opts: { + card: DecisionRelayCard; + displayName: string; + agentUsername: string; + link: string; +}): string => { + const { card, displayName, agentUsername, link } = opts; + const title = escapeHtml(card.title); + const question = escapeHtml(card.question); + const agent = escapeHtml(displayName || agentUsername); + const footer = '\n\nReply to this message with a number, or write your ruling.\n' + + `open in Commonly`; + const optionLines = card.options.map((option, index) => ( + `\n${index + 1}. ${escapeHtml(option.label)}${option.recommended ? ' ★ recommended' : ''}` + )); + const header = `${agent} needs a ruling · ${title}\n${question}\n`; + let remaining = CARD_OUTBOUND_TEXT_CAP - header.length - footer.length + - optionLines.reduce((sum, line) => sum + line.length, 0); + const descriptions = card.options.map((option) => ( + option.description ? escapeHtml(truncateWithEllipsis(option.description, 100)) : '' + )); + + const descriptionLines = descriptions.map((description, index) => { + if (!description || remaining <= 4) return ''; + const nonEmptyRemaining = descriptions.slice(index).filter(Boolean).length; + const allowance = Math.max(0, Math.floor(remaining / Math.max(1, nonEmptyRemaining)) - 4); + if (!allowance) return ''; + const fitted = truncateWithEllipsis(description, allowance); + const line = `\n ${fitted}`; + remaining -= line.length; + return line; + }); + + return header + optionLines.map((line, index) => `${line}${descriptionLines[index]}`).join('') + footer; +}; + // Prefix an inbound Telegram quote-reply with the @mention of the agent whose // relayed line was quoted, so the normal mention pipeline routes it. Pure — // unit-tested without any I/O. @@ -161,6 +213,7 @@ export const relayAgentMessageToTelegram = async (opts: { displayName: string; content: string; podMessageId?: string | null; + card?: DecisionRelayCard; integration?: TelegramIntegrationDoc; }): Promise => { const { @@ -170,22 +223,61 @@ export const relayAgentMessageToTelegram = async (opts: { // The dispatcher selects each pod-scoped subscription. Its row is the // authority for this send; the fallback preserves legacy direct rows. const integration = opts.integration ?? await findLiveIntegration(podId); - if (!integration || !isRelayableIntegration(integration, podId)) return; + if (!integration) return; + const cardHoldReason = opts.card + ? (integration.config?.adminPause + ? 'paused' + : integration.scope === 'user' && integration.config?.gates?.[podId]?.enabled !== true + ? 'gate_off' + : null) + : null; + if (cardHoldReason) { + if (podMessageId) { + await channelVerdictService.record({ + integrationId: integration._id, + ...(integration.installationId ? { installationId: integration.installationId } : {}), + podId, + provider: 'telegram', + event: { kind: 'decision_request', podMessageId }, + verdict: 'hold', + reason: cardHoldReason, + }); + } + return; + } + if (!isRelayableIntegration(integration, podId)) return; // /mute pauses ALL outbound relay to the chat, escalations included — // mute means mute; /status shows it, and it self-expires. const mutedUntil = (integration.config as { relayMutedUntil?: Date | string })?.relayMutedUntil; - if (mutedUntil && new Date(mutedUntil) > new Date()) return; - if (!shouldEscalate({ content, agentUsername, integration, podId })) return; + if (mutedUntil && new Date(mutedUntil) > new Date()) { + if (opts.card && podMessageId) { + await channelVerdictService.record({ + integrationId: integration._id, + ...(integration.installationId ? { installationId: integration.installationId } : {}), + podId, + provider: 'telegram', + event: { kind: 'decision_request', podMessageId }, + verdict: 'hold', + reason: 'muted', + }); + } + return; + } + if (!opts.card && !shouldEscalate({ content, agentUsername, integration, podId })) return; const botToken = process.env.TELEGRAM_BOT_TOKEN; const chatId = integration.config?.chatId; if (!botToken || !chatId) return; const base = (process.env.PUBLIC_APP_URL || 'https://commonly.me').replace(/\/$/, ''); - const link = `${base}/v2/pods/${podId}`; + const link = opts.card && podMessageId + ? `${base}/v2/pods/${encodeURIComponent(podId)}?message=${encodeURIComponent(podMessageId)}` + : `${base}/v2/pods/${podId}`; const body = escapeHtml(String(content).slice(0, OUTBOUND_TEXT_CAP)); - const text = `${escapeHtml(displayName || agentUsername)}: ${body}` - + `\n\nopen in Commonly`; + const text = opts.card + ? renderTelegramDecisionCard({ card: opts.card, displayName, agentUsername, link }) + : `${escapeHtml(displayName || agentUsername)}: ${body}` + + `\n\nopen in Commonly`; const result = await telegramSend.sendMessage(botToken, chatId, text); const tgMessageId = result && result.messageId != null ? String(result.messageId) : null; @@ -199,6 +291,17 @@ export const relayAgentMessageToTelegram = async (opts: { }, }, }); + if (opts.card && podMessageId) { + await channelVerdictService.record({ + integrationId: integration._id, + ...(integration.installationId ? { installationId: integration.installationId } : {}), + podId, + provider: 'telegram', + event: { kind: 'decision_request', podMessageId }, + verdict: 'interrupt', + reason: 'card', + }); + } } catch (err) { console.warn('[tg-bridge] outbound relay failed:', (err as Error).message); } @@ -398,6 +501,7 @@ module.exports = { routeReplyContent, isRelayableIntegration, isInboundRelayableIntegration, + renderTelegramDecisionCard, relayAgentMessageToTelegram, relayTelegramMessageToPod, };