feat(connectors): relay decision cards to channels

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();
+  });
+});

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'),

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();
+  });
+});