diff --git a/backend/__tests__/unit/services/attentionItemService.test.js b/backend/__tests__/unit/services/attentionItemService.test.js index 28cd9547b..640c35498 100644 --- a/backend/__tests__/unit/services/attentionItemService.test.js +++ b/backend/__tests__/unit/services/attentionItemService.test.js @@ -954,3 +954,41 @@ describe('an agent mention nobody answered for a week leaves the queue as missed expect(AttentionItemService.AGENT_MENTION_MISSED_DAYS).toBe(7); }); }); + +// F48: a hosted-MCP pin that drifted upstream used to reach only a log line. +// The card goes to one configured operator, in a pod that operator can open. +describe('a hosted-MCP drift card reaches its operator', () => { + const drift = { + recipientUserId: 'operator', podId: 'pod-ops', sourceId: 'drift:sentry:abc', + title: 'Sentry: 7 pinned tools changed upstream', detail: 'Paused until re-pinned: search_issues.', + }; + + beforeEach(() => { + jest.clearAllMocks(); + mockUpdateOne.mockResolvedValue({ upsertedId: 'item-1' }); + mockPodFindById.mockReturnValue(chain({ _id: 'pod-ops', name: 'Connectors', members: ['operator', 'bot'] })); + mockUserFind.mockReturnValue(chain([{ _id: 'operator', username: 'connector-ops', isBot: false }])); + }); + + it('records one blocked card for a human member, with no email', async () => { + await expect(AttentionItemService.recordHostedMcpDrift(drift)).resolves.toEqual({ recorded: true }); + + expect(mockUpdateOne).toHaveBeenCalledTimes(1); + expect(mockUpdateOne).toHaveBeenCalledWith( + { recipientUserId: 'operator', 'source.type': 'hosted_mcp', 'source.id': 'drift:sentry:abc' }, + { $setOnInsert: expect.objectContaining({ + kind: 'blocked', podId: 'pod-ops', podName: 'Connectors', status: 'open', + title: drift.title, detail: drift.detail, + }) }, + { upsert: true }, + ); + expect(mockSendNeedsYouEmail).not.toHaveBeenCalled(); + }); + + it('records nothing for a recipient the pod does not list, and says why', async () => { + mockUserFind.mockReturnValue(chain([])); + await expect(AttentionItemService.recordHostedMcpDrift(drift)) + .resolves.toEqual({ recorded: false, reason: 'recipient_not_member' }); + expect(mockUpdateOne).not.toHaveBeenCalled(); + }); +}); diff --git a/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js b/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js new file mode 100644 index 000000000..250697a15 --- /dev/null +++ b/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js @@ -0,0 +1,243 @@ +// F48: on 11-01 Sentry's seven search reads drifted upstream and were refused +// as tool_drift on prod. The only signal was a console.warn nobody reads, so a +// connector sat 7/11 dark until a template capture tripped over it. A drift now +// becomes one Needs-you card for the operator who re-pins. +const { createHostedMcpDriftAlert } = require('../../../services/hostedMcpDriftAlertService'); +const { HOSTED_MCP_ENTRIES, findHostedMcpEntry } = require('../../../integrations/hostedMcp/entries'); + +const sentry = findHostedMcpEntry(HOSTED_MCP_ENTRIES, 'sentry'); +const row = { connectionId: 'row-1' }; +const OPERATOR = '6a1b2c3d4e5f60718293a4b5'; +const POD = '6a8f6dc7a1dccf2e02f31f99'; +const config = { recipientUserId: OPERATOR, podId: POD }; +const drifted = (name) => ({ + name, namespacedName: `sentry.${name}`, verdict: 'tool_drift', + detail: `upstream inputSchema for ${name} differs from the pin`, +}); +const unavailable = (name) => ({ + name, namespacedName: `sentry.${name}`, verdict: 'tool_unavailable', + detail: `this connection's tools/list does not include ${name}`, +}); +const ok = (name) => ({ name, namespacedName: `sentry.${name}`, verdict: 'ok' }); + +const setup = (over = {}) => { + const readConfig = jest.fn(async () => config); + const record = jest.fn(async () => ({ recorded: true })); + const alert = createHostedMcpDriftAlert(Object.assign({ readConfig, record }, over)); + return { alert, readConfig: over.readConfig || readConfig, record: over.record || record }; +}; + +let error; +beforeEach(() => { error = jest.spyOn(console, 'error').mockImplementation(() => {}); }); +afterEach(() => error.mockRestore()); + +const logged = () => error.mock.calls.map(([line]) => JSON.parse(line)); + +describe('a drifted pin becomes one card for the operator', () => { + test('drift files a card naming the entry and every drifted tool', async () => { + const { alert, record } = setup(); + await alert(row, sentry, [drifted('search_issues'), drifted('search_errors'), ok('find_organizations')]); + + expect(record).toHaveBeenCalledTimes(1); + const [card] = record.mock.calls[0]; + expect(card).toMatchObject({ recipientUserId: OPERATOR, podId: POD }); + expect(card.sourceId).toMatch(/^drift:sentry:[0-9a-f]{16}$/); + expect(card.title).toBe('Sentry: 2 pinned tools changed upstream'); + expect(card.detail).toContain('search_issues, search_errors'); + expect(card.detail).toContain('row-1'); + expect(card.detail).not.toMatch(/first seen/i); + expect(card.detail).not.toContain('find_organizations'); + }); + + test('one drifted tool reads in the singular', async () => { + const { alert, record } = setup(); + await alert(row, sentry, [drifted('search_issues')]); + expect(record.mock.calls[0][0].title).toBe('Sentry: 1 pinned tool changed upstream'); + }); + + test('the same drift files once per process, however often and on however many rows', async () => { + const { alert, record } = setup(); + await alert(row, sentry, [drifted('search_issues')]); + await alert(row, sentry, [drifted('search_issues')]); + await alert({ connectionId: 'row-2' }, sentry, [drifted('search_issues')]); + expect(record).toHaveBeenCalledTimes(1); + }); + + // Review on e1918d515: a memo keyed by the drift alone sent nothing to an + // operator configured after the first card. + test('a changed operator gets their own card, and the first is not filed twice', async () => { + const NEXT = '6a1b2c3d4e5f60718293a4c6'; + let current = config; + const { alert, record } = setup({ readConfig: jest.fn(async () => current) }); + await alert(row, sentry, [drifted('search_issues')]); + current = { recipientUserId: NEXT, podId: POD }; + await alert(row, sentry, [drifted('search_issues')]); + current = config; + await alert(row, sentry, [drifted('search_issues')]); + + expect(record.mock.calls.map(([card]) => card.recipientUserId)).toEqual([OPERATOR, NEXT]); + expect(record.mock.calls[1][0].sourceId).toBe(record.mock.calls[0][0].sourceId); + }); + + // Review on c7c4addfc: the card's unique key is recipient + source id and the + // write only inserts, so a source id without the pod matched the card in the + // old pod and left the new one empty. + test('a moved alert pod gets its own card, for the same operator', async () => { + const NEXT_POD = '6a8f6dc7a1dccf2e02f31fa1'; + let current = config; + const { alert, record } = setup({ readConfig: jest.fn(async () => current) }); + await alert(row, sentry, [drifted('search_issues')]); + current = { recipientUserId: OPERATOR, podId: NEXT_POD }; + await alert(row, sentry, [drifted('search_issues')]); + + expect(record.mock.calls.map(([card]) => card.podId)).toEqual([POD, NEXT_POD]); + expect(record.mock.calls[1][0].sourceId).not.toBe(record.mock.calls[0][0].sourceId); + }); + + // Review on c7d2fa765: a key of the pin set alone lost a second tool's drift + // under the same pins, and a drift only another row's consent can see. + test('another tool drifting under the same pins files its own card', async () => { + const { alert, record } = setup(); + await alert(row, sentry, [drifted('search_issues')]); + await alert({ connectionId: 'row-2' }, sentry, [drifted('search_issues'), drifted('update_issue')]); + expect(record).toHaveBeenCalledTimes(2); + expect(record.mock.calls[1][0].sourceId).not.toBe(record.mock.calls[0][0].sourceId); + expect(record.mock.calls[1][0].detail).toContain('search_issues, update_issue'); + }); + + test('the key does not depend on the order the tools were assessed in', async () => { + const { alert, record } = setup(); + await alert(row, sentry, [drifted('search_issues'), drifted('search_errors')]); + await alert(row, sentry, [drifted('search_errors'), drifted('search_issues')]); + expect(record).toHaveBeenCalledTimes(1); + }); + + test('a re-pinned entry is a new pin set, so its drift files again', async () => { + const { alert, record } = setup(); + await alert(row, sentry, [drifted('search_issues')]); + const repinned = Object.assign({}, sentry, { + tools: sentry.tools.map((tool, index) => (index === 0 + ? Object.assign({}, tool, { description: `${tool.description} (re-pinned)` }) : tool)), + }); + await alert(row, repinned, [drifted('search_issues')]); + expect(record).toHaveBeenCalledTimes(2); + expect(record.mock.calls[1][0].sourceId).not.toBe(record.mock.calls[0][0].sourceId); + }); + + // A row's list depends on what its owner consented to, so an absent tool is + // not evidence about the vendor (the Linear smoke row lists 26 as absent + // because its scope is read-only). Only drift is unambiguous. + test('an absent or unchanged tool files nothing and reads no config', async () => { + const { alert, record, readConfig } = setup(); + await alert(row, sentry, [unavailable('update_issue'), ok('search_issues')]); + expect(record).not.toHaveBeenCalled(); + expect(readConfig).not.toHaveBeenCalled(); + }); +}); + +describe('a drift that cannot be delivered says so, loudly', () => { + test('no configured operator: an error line, and no card', async () => { + const { alert, record } = setup({ readConfig: jest.fn(async () => null) }); + await alert(row, sentry, [drifted('search_issues')]); + await alert(row, sentry, [drifted('search_issues')]); + + expect(record).not.toHaveBeenCalled(); + expect(logged()).toEqual([expect.objectContaining({ + event: 'hosted_mcp_drift_alert', delivery: 'unconfigured', entryId: 'sentry', + connectionId: 'row-1', tools: ['search_issues'], + })]); + }); + + test('a half-written config is unconfigured, not a card to nobody', async () => { + const { alert, record } = setup({ readConfig: jest.fn(async () => ({ recipientUserId: OPERATOR })) }); + await alert(row, sentry, [drifted('search_issues')]); + expect(record).not.toHaveBeenCalled(); + expect(logged()[0]).toMatchObject({ delivery: 'unconfigured' }); + }); + + // Review on c7d2fa765: a bad id threw a CastError, which logged `failed` and + // retried on every fresh assessment of every drifted row. + test('a malformed id is unconfigured, not a failure retried forever', async () => { + const { alert, record } = setup({ readConfig: jest.fn(async () => ({ recipientUserId: 'connector-ops', podId: POD })) }); + await alert(row, sentry, [drifted('search_issues')]); + await alert(row, sentry, [drifted('search_issues')]); + expect(record).not.toHaveBeenCalled(); + expect(logged()).toEqual([expect.objectContaining({ delivery: 'unconfigured' })]); + }); + + // Reviews on c7d2fa765 and e1918d515: memoizing an undelivered drift, until + // restart and then for an hour, held back the card from an operator who set + // the config, or joined the pod, after the first error. + test('a config set after an unconfigured drift files on the next assessment', async () => { + const readConfig = jest.fn() + .mockResolvedValueOnce(null) + .mockResolvedValue(config); + const { alert, record } = setup({ readConfig }); + await alert(row, sentry, [drifted('search_issues')]); + await alert(row, sentry, [drifted('search_issues')]); + expect(record).toHaveBeenCalledTimes(1); + }); + + test('an operator who joins the pod after the first error gets the card next time', async () => { + const record = jest.fn() + .mockResolvedValueOnce({ recorded: false, reason: 'recipient_not_member' }) + .mockResolvedValue({ recorded: true }); + const { alert } = setup({ record }); + await alert(row, sentry, [drifted('search_issues')]); + await alert(row, sentry, [drifted('search_issues')]); + await alert(row, sentry, [drifted('search_issues')]); + expect(record).toHaveBeenCalledTimes(2); + }); + + test('an undeliverable drift logs once an hour while it lasts, not on every assessment', async () => { + let clock = 1000; + const { alert } = setup({ readConfig: jest.fn(async () => null), now: () => clock }); + await alert(row, sentry, [drifted('search_issues')]); + clock += 59 * 60 * 1000; + await alert(row, sentry, [drifted('search_issues')]); + expect(logged()).toHaveLength(1); + clock += 2 * 60 * 1000; + await alert(row, sentry, [drifted('search_issues')]); + expect(logged()).toHaveLength(2); + }); + + // Review on c7d2fa765: two replicas inserting at once make one upsert lose + // on the unique index. The card exists, so that is not a failure. + test('losing an insert race to another replica counts as filed', async () => { + const record = jest.fn(async () => { throw Object.assign(new Error('E11000 duplicate key'), { code: 11000 }); }); + const { alert } = setup({ record }); + await alert(row, sentry, [drifted('search_issues')]); + await alert(row, sentry, [drifted('search_issues')]); + expect(record).toHaveBeenCalledTimes(1); + expect(error).not.toHaveBeenCalled(); + }); + + test('an operator the pod does not list: an error line naming why', async () => { + const { alert } = setup({ record: jest.fn(async () => ({ recorded: false, reason: 'recipient_not_member' })) }); + await alert(row, sentry, [drifted('search_issues')]); + expect(logged()).toEqual([expect.objectContaining({ delivery: 'recipient_not_member', entryId: 'sentry' })]); + }); + + test('a failed write is logged and retried on the next assessment', async () => { + const record = jest.fn() + .mockRejectedValueOnce(new Error('mongo unavailable')) + .mockResolvedValueOnce({ recorded: true }); + const { alert } = setup({ record }); + await alert(row, sentry, [drifted('search_issues')]); + await alert(row, sentry, [drifted('search_issues')]); + + expect(record).toHaveBeenCalledTimes(2); + expect(logged()[0]).toMatchObject({ delivery: 'failed', reason: 'mongo unavailable' }); + }); + + test('a failed config read is logged and retried too', async () => { + const readConfig = jest.fn() + .mockRejectedValueOnce(new Error('mongo unavailable')) + .mockResolvedValueOnce(config); + const { alert, record } = setup({ readConfig }); + await alert(row, sentry, [drifted('search_issues')]); + await alert(row, sentry, [drifted('search_issues')]); + expect(record).toHaveBeenCalledTimes(1); + expect(logged()[0]).toMatchObject({ delivery: 'failed' }); + }); +}); diff --git a/backend/__tests__/unit/services/hostedMcpEntry.test.js b/backend/__tests__/unit/services/hostedMcpEntry.test.js index 0428b8ff0..d876ef95d 100644 --- a/backend/__tests__/unit/services/hostedMcpEntry.test.js +++ b/backend/__tests__/unit/services/hostedMcpEntry.test.js @@ -469,7 +469,7 @@ describe('the comparison is cached per row, and an outage is not cacheable', () name: 'read_issue', namespacedName: 'linear.read_issue', verdict: 'tool_unavailable', - detail: 'upstream no longer lists get_issue', + detail: "this connection's tools/list does not include get_issue", }], }); expect(line).not.toContain(VENDOR_TEXT); @@ -478,6 +478,66 @@ describe('the comparison is cached per row, and an outage is not cacheable', () } }); + test('an absent tool is named against this connection, not as a vendor removal', async () => { + // A row's list depends on what its owner consented to (Sentry gates + // update_issue behind Triage), so absence on one row proves nothing about + // the vendor. "Upstream no longer lists" read as a removal on 11-01 (F48). + const [verdict] = assessEntryTools(linear(), upstreamOk().filter((tool) => tool.name !== 'get_issue')) + .filter((tool) => tool.verdict !== 'ok'); + expect(verdict.verdict).toBe('tool_unavailable'); + expect(verdict.detail).toBe("this connection's tools/list does not include get_issue"); + expect(verdict.detail).not.toMatch(/no longer/); + }); + + test('a fresh assessment is handed to onAssessed once, and a cached one is not', async () => { + const onAssessed = jest.fn(); + const warn = jest.spyOn(console, 'warn').mockImplementation(() => {}); + try { + const drift = cache(async () => upstreamOk().filter((tool) => tool.name !== 'get_issue'), { onAssessed }); + const first = await drift.assess(row, linear()); + await drift.assess(row, linear()); + + expect(onAssessed).toHaveBeenCalledTimes(1); + const [seenRow, seenEntry, seenTools] = onAssessed.mock.calls[0]; + expect(seenRow).toEqual(row); + expect(seenEntry.id).toBe('linear'); + expect(seenTools).toBe(first.tools); + } finally { + warn.mockRestore(); + } + }); + + test('a clean assessment is handed to onAssessed too', async () => { + const onAssessed = jest.fn(); + await cache(async () => upstreamOk(), { onAssessed }).assess(row, linear()); + expect(onAssessed).toHaveBeenCalledTimes(1); + expect(onAssessed.mock.calls[0][2].every((tool) => tool.verdict === 'ok')).toBe(true); + }); + + test('an outage is not handed to onAssessed', async () => { + const onAssessed = jest.fn(); + const outcome = await cache(async () => { throw new Error('ECONNRESET'); }, { onAssessed }) + .assess(row, linear()); + expect(outcome.state).toBe('provider_unavailable'); + expect(onAssessed).not.toHaveBeenCalled(); + }); + + test.each([ + ['throws', () => { throw new Error('alert store down'); }], + ['rejects', async () => { throw new Error('alert store down'); }], + ])('an onAssessed that %s cannot change the outcome', async (_label, onAssessed) => { + const warn = jest.spyOn(console, 'warn').mockImplementation(() => {}); + try { + const outcome = await cache(async () => upstreamOk(), { onAssessed }).assess(row, linear()); + expect(outcome).toMatchObject({ state: 'assessed', cached: false }); + // Let a rejected hook settle before asserting what it logged. + await new Promise((resolve) => setImmediate(resolve)); + expect(warn.mock.calls.map(([line]) => String(line)).join('\n')).toContain('alert store down'); + } finally { + warn.mockRestore(); + } + }); + test('a clean assessment logs nothing', async () => { const warn = jest.spyOn(console, 'warn').mockImplementation(() => {}); try { diff --git a/backend/__tests__/unit/services/hostedMcpToolDefinitions.test.js b/backend/__tests__/unit/services/hostedMcpToolDefinitions.test.js index 350cc67fb..f28015a60 100644 --- a/backend/__tests__/unit/services/hostedMcpToolDefinitions.test.js +++ b/backend/__tests__/unit/services/hostedMcpToolDefinitions.test.js @@ -19,6 +19,13 @@ const { } = require('../../../services/hostedMcpToolDefinitions'); const { AIRTABLE_ENTRY } = require('../../../integrations/hostedMcp/airtable'); +// The runtime cache hands each fresh assessment to the drift alert (F48). The +// alert's own behaviour is pinned in hostedMcpDriftAlertService.test.js. +const mockAlertOnHostedMcpDrift = jest.fn(); +jest.mock('../../../services/hostedMcpDriftAlertService', () => ({ + alertOnHostedMcpDrift: (...args) => mockAlertOnHostedMcpDrift(...args), +})); + const ENTRY = { id: 'linear', title: 'Linear', @@ -169,6 +176,34 @@ describe('hosted-MCP tool definitions', () => { } }); + it('hands a fresh assessment to the drift alert', async () => { + mockAlertOnHostedMcpDrift.mockClear(); + const fetcher = jest.fn(async (_url, init) => { + if (init.method === 'DELETE') return new Response(null, { status: 204 }); + const body = JSON.parse(init.body); + if (body.method === 'initialize') { + return new Response(JSON.stringify({ jsonrpc: '2.0', id: body.id, result: { + protocolVersion: '2025-11-25', capabilities: { tools: {} }, serverInfo: { name: 'fake', version: '1' }, + } }), { headers: { 'content-type': 'application/json', 'mcp-session-id': 's-1' } }); + } + if (body.method === 'notifications/initialized') return new Response(null, { status: 202 }); + if (body.method === 'tools/list') { + return jsonResponse({ jsonrpc: '2.0', id: body.id, result: { + tools: [{ name: 'list_issues', inputSchema: { type: 'object', properties: { limit: { type: 'string' } } } }], + } }); + } + throw new Error(`unexpected ${body.method}`); + }); + const deps = makeDeps(); + deps.fetcher = fetcher; + deps.driftCache = createHostedMcpRuntimeDriftCache(deps, [ENTRY]); + const outcome = await deps.driftCache.assess({ connectionId: ROW_ID }, ENTRY); + + expect(mockAlertOnHostedMcpDrift).toHaveBeenCalledTimes(1); + expect(mockAlertOnHostedMcpDrift).toHaveBeenCalledWith({ connectionId: ROW_ID }, ENTRY, outcome.tools); + expect(outcome.tools.find((tool) => tool.name === 'list_issues').verdict).toBe('tool_drift'); + }); + it('opens a fresh session after the vendor expires one before tools/list', async () => { let opened = 0; let listed = 0; diff --git a/backend/models/AttentionItem.ts b/backend/models/AttentionItem.ts index 9e53ccfc7..1c69545bc 100644 --- a/backend/models/AttentionItem.ts +++ b/backend/models/AttentionItem.ts @@ -1,7 +1,7 @@ import mongoose, { Document, Model, Schema, Types } from 'mongoose'; export type AttentionKind = 'mention' | 'approval' | 'decision' | 'handoff' | 'blocked'; -export type AttentionSourceType = 'message' | 'approval' | 'decision_request' | 'task' | 'initiative'; +export type AttentionSourceType = 'message' | 'approval' | 'decision_request' | 'task' | 'initiative' | 'hosted_mcp'; export interface IAttentionItem extends Document { recipientUserId: Types.ObjectId; @@ -52,7 +52,7 @@ const attentionItemSchema = new Schema({ podId: { type: Schema.Types.ObjectId, ref: 'Pod', required: true }, kind: { type: String, enum: ['mention', 'approval', 'decision', 'handoff', 'blocked'], required: true }, source: { - type: { type: String, enum: ['message', 'approval', 'approval_action', 'decision_request', 'task', 'initiative'], required: true }, + type: { type: String, enum: ['message', 'approval', 'approval_action', 'decision_request', 'task', 'initiative', 'hosted_mcp'], required: true }, id: { type: String, required: true }, }, title: { type: String, required: true }, diff --git a/backend/services/attentionItemService.ts b/backend/services/attentionItemService.ts index eab0e1caa..3a2d786e2 100644 --- a/backend/services/attentionItemService.ts +++ b/backend/services/attentionItemService.ts @@ -23,7 +23,8 @@ const { humansAnsweredBy } = require('./agentAnswerService'); // eslint-disable-next-line global-require const { isListedPodMember } = require('../utils/isPodMember'); -type SourceType = 'message' | 'approval' | 'approval_action' | 'decision_request' | 'task' | 'initiative'; +type SourceType = 'message' | 'approval' | 'approval_action' | 'decision_request' | 'task' | 'initiative' + | 'hosted_mcp'; type Kind = 'mention' | 'approval' | 'decision' | 'handoff' | 'blocked'; type MentionOptions = { isAlreadyAcknowledged?: (recipientUserId: unknown, legacyMentionId: string) => boolean; @@ -661,6 +662,27 @@ export const recordInitiativeStall = async (stall: { } }; +// A hosted-MCP pin that drifted upstream (F48). It goes to one configured +// operator, in a pod that operator is listed in, because the queue hides a row +// whose recipient cannot open its pod. No email: the queue is the delivery, and +// the operator account's mailbox may be someone else's. The source id carries +// the pin set and the drifted tools, so each drift is one card; it stays until +// the operator acknowledges it, since another row may still see the drift. +export const recordHostedMcpDrift = async (drift: { + recipientUserId: unknown; podId: unknown; sourceId: string; title: string; detail: string; +}): Promise<{ recorded: boolean; reason?: 'recipient_not_member' }> => { + const recipient = (await currentHumanMembers(drift.podId)) + .find((member) => String(member._id) === String(drift.recipientUserId)); + if (!recipient) return { recorded: false, reason: 'recipient_not_member' }; + const pod = await Pod.findById(drift.podId).select('name').lean(); + await recordForRecipients([recipient], { + podId: drift.podId, kind: 'blocked' as Kind, sourceType: 'hosted_mcp' as SourceType, + sourceId: drift.sourceId, title: compact(drift.title, 200), detail: compact(drift.detail, 600), + podName: pod?.name || 'Pod', suppressEmail: true, + }); + return { recorded: true }; +}; + // Closes this initiative's open stall rows, except the current episode's. A // card for a stall that has ended would otherwise sit in the queue forever. export const resolveInitiativeStalls = async (initiativeId: unknown, keepSourceId?: string): Promise => { @@ -1015,6 +1037,6 @@ export const hasEverHadAttention = async (recipientUserId: unknown): Promise Promise; + record: (card: HostedMcpDriftCard) => Promise<{ recorded: boolean; reason?: string }>; + now?: () => number; +} + +/** + * An undeliverable drift (no config, a recipient the pod does not list, a + * failed write) logs again after this long while it lasts. It is never + * suppressed: the next fresh assessment reads the config and tries to file. + */ +export const DRIFT_ALERT_RELOG_MS = 60 * 60 * 1000; + +/** + * The pin set, the drifted tools AND the pod the card goes to. A re-pin is a + * new set, so its next drift files afresh; another tool drifting under the + * same pins, or one only another row's consent can see, is a new drift with its + * own card. The card is unique per recipient and source id and is only ever + * inserted, so a moved alert pod needs its own id or the old pod keeps the card. + */ +export const driftAlertSourceId = (entry: HostedMcpEntry, drifted: string[], podId: string): string => { + const pins = entry.tools.map((tool) => digestPinnedDefinition(entry.id, tool)).join('\n'); + const key = `${podId}\n--\n${pins}\n--\n${[...drifted].sort().join('\n')}`; + return `drift:${entry.id}:${createHash('sha256').update(key).digest('hex').slice(0, 16)}`; +}; + +const OBJECT_ID = /^[0-9a-f]{24}$/i; + +// A malformed id would throw a CastError on every write, so it counts as unset. +const configFrom = (value: unknown): HostedMcpDriftAlertConfig | null => { + const config = value as Partial | null | undefined; + const recipientUserId = String(config?.recipientUserId || '').trim(); + const podId = String(config?.podId || '').trim(); + return OBJECT_ID.test(recipientUserId) && OBJECT_ID.test(podId) ? { recipientUserId, podId } : null; +}; + +export const createHostedMcpDriftAlert = (deps: HostedMcpDriftAlertDeps) => { + const now = deps.now || Date.now; + // Cards this process has filed, keyed by the drift and who it went to: a + // second row seeing the same drift adds nothing, while a newly configured + // operator gets their own card. The card's unique index holds across replicas. + const filed = new Set(); + // When each undeliverable drift last logged, so a drift that persists across + // many assessments logs hourly rather than on every one. + const reportedAt = new Map(); + + return async (row: HostedMcpDriftRow, entry: HostedMcpEntry, tools: HostedMcpToolAssessment[]): Promise => { + const drifted = tools.filter((tool) => tool.verdict === 'tool_drift').map((tool) => tool.name); + if (drifted.length === 0) return; + // The log throttle's key: this drift, whoever it would go to. + const driftKey = driftAlertSourceId(entry, drifted, ''); + + // Names come from the pins and the ids from our rows, never from the + // vendor's descriptions or schemas. + const report = (delivery: string, reason?: string) => { + const key = `${driftKey}|${delivery}`; + const last = reportedAt.get(key); + if (last !== undefined && now() - last < DRIFT_ALERT_RELOG_MS) return; + reportedAt.set(key, now()); + console.error(JSON.stringify({ + event: 'hosted_mcp_drift_alert', + delivery, + ...(reason ? { reason } : {}), + entryId: entry.id, + connectionId: row.connectionId, + tools: drifted, + })); + }; + const failed = (error: unknown) => report('failed', error instanceof Error ? error.message : String(error)); + + // Read on every fresh drifted assessment, before the memo: the operator + // may have been set or changed since the last one. + let config: HostedMcpDriftAlertConfig | null; + try { + config = configFrom(await deps.readConfig()); + } catch (error) { + failed(error); + return; + } + if (!config) { + report('unconfigured'); + return; + } + const sourceId = driftAlertSourceId(entry, drifted, config.podId); + const filedKey = `${sourceId}|${config.recipientUserId}`; + if (filed.has(filedKey)) return; + + try { + const result = await deps.record({ + ...config, + sourceId, + title: `${entry.title}: ${drifted.length} pinned ${drifted.length === 1 ? 'tool' : 'tools'} changed upstream`, + detail: `Refused as tool_drift until re-pinned: ${drifted.join(', ')}. ` + + `Seen on connection ${row.connectionId}. Re-pin from a fresh tools/list capture.`, + }); + if (result.recorded) filed.add(filedKey); + else report(result.reason || 'not_recorded'); + } catch (error) { + // Another replica inserted the same card first: it exists, so it is filed. + if ((error as { code?: unknown })?.code === 11000) { + filed.add(filedKey); + return; + } + failed(error); + } + }; +}; + +export const alertOnHostedMcpDrift = createHostedMcpDriftAlert({ + readConfig: async () => { + const row = await SystemSetting.findOne({ key: DRIFT_ALERT_SETTING_KEY }).lean() as { value?: unknown } | null; + return row?.value ?? null; + }, + record: (card) => AttentionItemService.recordHostedMcpDrift(card), +}); diff --git a/backend/services/hostedMcpEntryService.ts b/backend/services/hostedMcpEntryService.ts index d3155824f..657315b1c 100644 --- a/backend/services/hostedMcpEntryService.ts +++ b/backend/services/hostedMcpEntryService.ts @@ -23,10 +23,12 @@ * - It caches that comparison per row behind a TTL, so the vendor is polled on * a schedule rather than on every `tools/list` or `tools/call`. * - * Drift is a named refusal, never a silent change: a pinned tool that is gone - * upstream is `tool_unavailable`, and one whose schema or annotations moved - * incompatibly is `tool_drift` until a PR re-pins it. A compatible read-schema - * change keeps the pin and emits a fresh-assessment signal. + * Drift is a named refusal, never a silent change: a pinned tool absent from + * this row's list is `tool_unavailable`, and one whose schema or annotations + * moved incompatibly is `tool_drift` until a PR re-pins it. A compatible + * read-schema change keeps the pin and emits a fresh-assessment signal. Only + * `tool_drift` reaches an operator (hostedMcpDriftAlertService): absence can + * be the row's own consent, so it says nothing about the vendor. */ import { createHash } from 'crypto'; import type { HostedToolDefinition } from './toolBrokerService'; @@ -334,7 +336,10 @@ export const assessEntryTools = ( name: tool.name, namespacedName, verdict: 'tool_unavailable' as const, - detail: `upstream no longer lists ${tool.upstreamName}`, + // Named against this row: a list depends on what its owner consented to + // (Sentry gates update_issue behind Triage), so absence here is not a + // vendor removal. "Upstream no longer lists" read as one on 11-01. + detail: `this connection's tools/list does not include ${tool.upstreamName}`, }; } const schemaChange = tool.class === 'read' @@ -395,6 +400,11 @@ export interface HostedMcpDriftCacheOptions { * not be reached; the caller must not read that as a clean comparison. */ listUpstreamTools: (row: HostedMcpDriftRow) => Promise; + /** + * Sees each fresh assessment, clean or not; never a cached one or an outage. + * It runs detached: what it throws is logged and cannot change the outcome. + */ + onAssessed?: (row: HostedMcpDriftRow, entry: HostedMcpEntry, tools: HostedMcpToolAssessment[]) => unknown; ttlMs?: number; now?: () => number; } @@ -482,6 +492,19 @@ export const createHostedMcpDriftCache = (options: HostedMcpDriftCacheOptions): })); } entries.set(key, { at, tools }); + if (options.onAssessed) { + const { onAssessed } = options; + Promise.resolve() + .then(() => onAssessed(row, entry, tools)) + .catch((error) => { + console.warn(JSON.stringify({ + event: 'hosted_mcp_assessment_hook_failed', + entryId: entry.id, + connectionId: row.connectionId, + reason: error instanceof Error ? error.message : String(error), + })); + }); + } return { state: 'assessed', entryId: entry.id, at, cached: false, tools }; }, }; diff --git a/backend/services/hostedMcpToolDefinitions.ts b/backend/services/hostedMcpToolDefinitions.ts index e7335fab8..3c2c7fac5 100644 --- a/backend/services/hostedMcpToolDefinitions.ts +++ b/backend/services/hostedMcpToolDefinitions.ts @@ -480,6 +480,13 @@ export const createHostedMcpRuntimeDriftCache = ( entries: HostedMcpEntry[] = HOSTED_MCP_ENTRIES, ): HostedMcpDriftCache => createHostedMcpDriftCache({ listUpstreamTools: ({ connectionId }) => listUpstreamTools(connectionId, deps, entries), + // Loaded on first use: the alert reads SystemSetting, and the frontend + // glyph-catalogue test imports this module without mongoose. + onAssessed: (row, entry, tools) => { + // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports + const { alertOnHostedMcpDrift } = require('./hostedMcpDriftAlertService'); + return alertOnHostedMcpDrift(row, entry, tools); + }, }); const productionDriftCache = createHostedMcpRuntimeDriftCache({