From c7d2fa7657078fa06c3f6a37b0159bba7c068e49 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sun, 1 Nov 2026 15:44:27 -0800 Subject: [PATCH 1/4] fix(connectors): a drifted hosted-MCP pin reaches its operator On 11-01 Sentry respelled seven search reads. Each was refused as tool_drift, and the only trace was a console.warn, so the connector sat 7 of 11 tools dark until a template capture tripped over it. A fresh drift assessment now goes to an onAssessed hook. For tool_drift it files one Needs-you card per drifted pin set, for the operator named in SystemSetting hostedMcp.driftAlert, in a pod that operator is listed in. No email. Unconfigured or undeliverable drift is a console.error. tool_unavailable files nothing: a row's list depends on its owner's consent, and its detail now says so instead of "upstream no longer lists", which read as a vendor removal. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_015AKgpQ2pr6xRMhN5mcfoHF --- .../services/attentionItemService.test.js | 38 +++++ .../hostedMcpDriftAlertService.test.js | 134 ++++++++++++++++++ .../unit/services/hostedMcpEntry.test.js | 62 +++++++- .../services/hostedMcpToolDefinitions.test.js | 35 +++++ backend/models/AttentionItem.ts | 4 +- backend/services/attentionItemService.ts | 28 +++- .../services/hostedMcpDriftAlertService.ts | 118 +++++++++++++++ backend/services/hostedMcpEntryService.ts | 33 ++++- backend/services/hostedMcpToolDefinitions.ts | 2 + 9 files changed, 443 insertions(+), 11 deletions(-) create mode 100644 backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js create mode 100644 backend/services/hostedMcpDriftAlertService.ts 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..9d99777dc --- /dev/null +++ b/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js @@ -0,0 +1,134 @@ +// 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 config = { recipientUserId: 'operator', podId: 'pod-ops' }; +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-ops' }); + 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.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 pin set files once per process, however often it is assessed', async () => { + const { alert, record, readConfig } = setup(); + await alert(row, sentry, [drifted('search_issues')]); + await alert({ connectionId: 'row-2' }, sentry, [drifted('search_issues'), drifted('search_logs')]); + expect(record).toHaveBeenCalledTimes(1); + expect(readConfig).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, once per pin set, 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' }); + }); + + 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..259c30d14 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, so each drift of each pin set 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 }>; +} + +/** The pin set, not the tool: a re-pin is a new set, so its next drift files afresh. */ +export const driftAlertSourceId = (entry: HostedMcpEntry): string => { + const pins = entry.tools.map((tool) => digestPinnedDefinition(entry.id, tool)).join('\n'); + return `drift:${entry.id}:${createHash('sha256').update(pins).digest('hex').slice(0, 16)}`; +}; + +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 recipientUserId && podId ? { recipientUserId, podId } : null; +}; + +export const createHostedMcpDriftAlert = (deps: HostedMcpDriftAlertDeps) => { + // Pin sets this process has filed or reported. A second row seeing the same + // drift adds nothing, and the card's unique index holds across replicas. + const handled = new Set(); + + 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; + const sourceId = driftAlertSourceId(entry); + if (handled.has(sourceId)) return; + + // 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) => console.error(JSON.stringify({ + event: 'hosted_mcp_drift_alert', + delivery, + ...(reason ? { reason } : {}), + entryId: entry.id, + connectionId: row.connectionId, + tools: drifted, + })); + + try { + const config = configFrom(await deps.readConfig()); + if (!config) { + handled.add(sourceId); + report('unconfigured'); + return; + } + 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(', ')}. ` + + `First seen on connection ${row.connectionId}. Re-pin from a fresh tools/list capture.`, + }); + handled.add(sourceId); + if (!result.recorded) report(result.reason || 'not_recorded'); + } catch (error) { + // Not marked handled: the next fresh assessment tries again. + report('failed', error instanceof Error ? error.message : String(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..bf6640f47 100644 --- a/backend/services/hostedMcpToolDefinitions.ts +++ b/backend/services/hostedMcpToolDefinitions.ts @@ -31,6 +31,7 @@ import { type HostedMcpUpstreamTool, } from './hostedMcpEntryService'; import { HostedMcpCredentialError } from './hostedMcpCredentialError'; +import { alertOnHostedMcpDrift } from './hostedMcpDriftAlertService'; import type { HostedMcpRow } from './hostedMcpCredentialService'; import type { ToolConnection, ToolDefinition } from './toolBrokerService'; @@ -480,6 +481,7 @@ export const createHostedMcpRuntimeDriftCache = ( entries: HostedMcpEntry[] = HOSTED_MCP_ENTRIES, ): HostedMcpDriftCache => createHostedMcpDriftCache({ listUpstreamTools: ({ connectionId }) => listUpstreamTools(connectionId, deps, entries), + onAssessed: alertOnHostedMcpDrift, }); const productionDriftCache = createHostedMcpRuntimeDriftCache({ From e1918d515c429006800b5e564317ae2bb994820b Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sun, 1 Nov 2026 15:48:57 -0800 Subject: [PATCH 2/4] fix(connectors): key a drift card on the drifted tools, recheck undeliverable drift Review on c7d2fa765. The card's source id now hashes the drifted tool names with the pin set, so another tool drifting under the same pins, or one only another row's consent lists, files its own card. An unconfigured or undeliverable drift is looked at again after an hour, so a config set later delivers without a restart. A malformed id counts as unconfigured, and losing the insert race to another replica counts as filed. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_015AKgpQ2pr6xRMhN5mcfoHF --- .../hostedMcpDriftAlertService.test.js | 70 +++++++++++++++++-- backend/services/attentionItemService.ts | 4 +- .../services/hostedMcpDriftAlertService.ts | 59 ++++++++++++---- 3 files changed, 111 insertions(+), 22 deletions(-) diff --git a/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js b/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js index 9d99777dc..2abbf23fe 100644 --- a/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js +++ b/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js @@ -7,7 +7,9 @@ const { HOSTED_MCP_ENTRIES, findHostedMcpEntry } = require('../../../integration const sentry = findHostedMcpEntry(HOSTED_MCP_ENTRIES, 'sentry'); const row = { connectionId: 'row-1' }; -const config = { recipientUserId: 'operator', podId: 'pod-ops' }; +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`, @@ -38,11 +40,12 @@ describe('a drifted pin becomes one card for the operator', () => { expect(record).toHaveBeenCalledTimes(1); const [card] = record.mock.calls[0]; - expect(card).toMatchObject({ recipientUserId: 'operator', podId: 'pod-ops' }); + 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'); }); @@ -52,14 +55,33 @@ describe('a drifted pin becomes one card for the operator', () => { expect(record.mock.calls[0][0].title).toBe('Sentry: 1 pinned tool changed upstream'); }); - test('the same pin set files once per process, however often it is assessed', async () => { + test('the same drift files once per process, however often and on however many rows', async () => { const { alert, record, readConfig } = setup(); await alert(row, sentry, [drifted('search_issues')]); - await alert({ connectionId: 'row-2' }, sentry, [drifted('search_issues'), drifted('search_logs')]); + await alert(row, sentry, [drifted('search_issues')]); + await alert({ connectionId: 'row-2' }, sentry, [drifted('search_issues')]); expect(record).toHaveBeenCalledTimes(1); expect(readConfig).toHaveBeenCalledTimes(1); }); + // 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')]); @@ -97,12 +119,50 @@ describe('a drift that cannot be delivered says so, loudly', () => { }); test('a half-written config is unconfigured, not a card to nobody', async () => { - const { alert, record } = setup({ readConfig: jest.fn(async () => ({ recipientUserId: 'operator' })) }); + 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' })]); + }); + + // Review on c7d2fa765: memoized until restart, so an operator who set the + // config after the first error never got the card. + test('an undeliverable drift checks the config again an hour later', async () => { + let clock = 1000; + const readConfig = jest.fn() + .mockResolvedValueOnce(null) + .mockResolvedValue(config); + const { alert, record } = setup({ readConfig, now: () => clock }); + await alert(row, sentry, [drifted('search_issues')]); + clock += 59 * 60 * 1000; + await alert(row, sentry, [drifted('search_issues')]); + expect(record).not.toHaveBeenCalled(); + clock += 2 * 60 * 1000; + await alert(row, sentry, [drifted('search_issues')]); + expect(record).toHaveBeenCalledTimes(1); + }); + + // 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')]); diff --git a/backend/services/attentionItemService.ts b/backend/services/attentionItemService.ts index 259c30d14..3a2d786e2 100644 --- a/backend/services/attentionItemService.ts +++ b/backend/services/attentionItemService.ts @@ -666,8 +666,8 @@ export const recordInitiativeStall = async (stall: { // 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, so each drift of each pin set is one card; it stays until the -// operator acknowledges it, since another row may still see the drift. +// 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' }> => { diff --git a/backend/services/hostedMcpDriftAlertService.ts b/backend/services/hostedMcpDriftAlertService.ts index 0636e1347..344718491 100644 --- a/backend/services/hostedMcpDriftAlertService.ts +++ b/backend/services/hostedMcpDriftAlertService.ts @@ -7,8 +7,8 @@ * over it. The refusal is right. What was missing is a reader. * * A drift is the catalogue's problem, not the row owner's, so it goes to the - * operator who re-pins: one Needs-you card per drifted pin set, in a pod that - * operator can open. Who that is lives in SystemSetting + * operator who re-pins: one Needs-you card per drift (the pin set and the + * tools that drifted), in a pod that operator can open. Who that is lives in SystemSetting * `hostedMcp.driftAlert` (`{ recipientUserId, podId }`), set per instance, so * no account or pod id is committed. * @@ -49,31 +49,51 @@ export interface HostedMcpDriftCard extends HostedMcpDriftAlertConfig { export interface HostedMcpDriftAlertDeps { readConfig: () => Promise; record: (card: HostedMcpDriftCard) => Promise<{ recorded: boolean; reason?: string }>; + now?: () => number; } -/** The pin set, not the tool: a re-pin is a new set, so its next drift files afresh. */ -export const driftAlertSourceId = (entry: HostedMcpEntry): string => { +/** + * An undeliverable drift (no config, or a recipient the pod does not list) is + * looked at again after this long, so a config set after the first error still + * delivers without a restart. A filed card is never filed again. + */ +export const DRIFT_ALERT_RECHECK_MS = 60 * 60 * 1000; + +/** + * The pin set AND the drifted tools. 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. + */ +export const driftAlertSourceId = (entry: HostedMcpEntry, drifted: string[]): string => { const pins = entry.tools.map((tool) => digestPinnedDefinition(entry.id, tool)).join('\n'); - return `drift:${entry.id}:${createHash('sha256').update(pins).digest('hex').slice(0, 16)}`; + const key = `${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 recipientUserId && podId ? { recipientUserId, podId } : null; + return OBJECT_ID.test(recipientUserId) && OBJECT_ID.test(podId) ? { recipientUserId, podId } : null; }; export const createHostedMcpDriftAlert = (deps: HostedMcpDriftAlertDeps) => { - // Pin sets this process has filed or reported. A second row seeing the same - // drift adds nothing, and the card's unique index holds across replicas. - const handled = new Set(); + const now = deps.now || Date.now; + // Drifts this process has filed (until forever) or reported undeliverable + // (until a recheck). A second row seeing the same drift adds nothing, and + // the card's unique index holds across replicas. + const handledUntil = 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; - const sourceId = driftAlertSourceId(entry); - if (handled.has(sourceId)) return; + const sourceId = driftAlertSourceId(entry, drifted); + if ((handledUntil.get(sourceId) ?? 0) > now()) return; + const filed = () => handledUntil.set(sourceId, Infinity); + const recheckLater = () => handledUntil.set(sourceId, now() + DRIFT_ALERT_RECHECK_MS); // Names come from the pins and the ids from our rows, never from the // vendor's descriptions or schemas. @@ -89,7 +109,7 @@ export const createHostedMcpDriftAlert = (deps: HostedMcpDriftAlertDeps) => { try { const config = configFrom(await deps.readConfig()); if (!config) { - handled.add(sourceId); + recheckLater(); report('unconfigured'); return; } @@ -98,11 +118,20 @@ export const createHostedMcpDriftAlert = (deps: HostedMcpDriftAlertDeps) => { sourceId, title: `${entry.title}: ${drifted.length} pinned ${drifted.length === 1 ? 'tool' : 'tools'} changed upstream`, detail: `Refused as tool_drift until re-pinned: ${drifted.join(', ')}. ` - + `First seen on connection ${row.connectionId}. Re-pin from a fresh tools/list capture.`, + + `Seen on connection ${row.connectionId}. Re-pin from a fresh tools/list capture.`, }); - handled.add(sourceId); - if (!result.recorded) report(result.reason || 'not_recorded'); + if (result.recorded) { + filed(); + } else { + recheckLater(); + 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(); + return; + } // Not marked handled: the next fresh assessment tries again. report('failed', error instanceof Error ? error.message : String(error)); } From c7c4addfce804dca42ccfc90f77d76aebd601ec4 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sun, 1 Nov 2026 16:01:14 -0800 Subject: [PATCH 3/4] fix(connectors): deliver a drift card once the operator can get it; keep mongoose out of the catalogue import - Read the drift-alert config on every fresh drifted assessment, before the memo, and key filed cards by drift, recipient and pod. A newly set or changed operator now gets their card on the next assessment. - Never suppress an undelivered drift. Only its error line is throttled, to once an hour while it lasts. - Load the alert service lazily from hostedMcpToolDefinitions. The eager import pulled SystemSetting (mongoose) into the frontend glyph-catalogue test, which failed with "TextDecoder is not defined". Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_015AKgpQ2pr6xRMhN5mcfoHF --- .../hostedMcpDriftAlertService.test.js | 54 ++++++++++--- .../services/hostedMcpDriftAlertService.ts | 80 +++++++++++-------- backend/services/hostedMcpToolDefinitions.ts | 9 ++- 3 files changed, 97 insertions(+), 46 deletions(-) diff --git a/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js b/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js index 2abbf23fe..165ba9358 100644 --- a/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js +++ b/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js @@ -56,12 +56,27 @@ describe('a drifted pin becomes one card for the operator', () => { }); test('the same drift files once per process, however often and on however many rows', async () => { - const { alert, record, readConfig } = setup(); + 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); - expect(readConfig).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 c7d2fa765: a key of the pin set alone lost a second tool's drift @@ -106,7 +121,7 @@ describe('a drifted pin becomes one card for the operator', () => { }); describe('a drift that cannot be delivered says so, loudly', () => { - test('no configured operator: an error line, once per pin set, and no card', async () => { + 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')]); @@ -135,21 +150,40 @@ describe('a drift that cannot be delivered says so, loudly', () => { expect(logged()).toEqual([expect.objectContaining({ delivery: 'unconfigured' })]); }); - // Review on c7d2fa765: memoized until restart, so an operator who set the - // config after the first error never got the card. - test('an undeliverable drift checks the config again an hour later', async () => { - let clock = 1000; + // 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, now: () => clock }); + 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(record).not.toHaveBeenCalled(); + expect(logged()).toHaveLength(1); clock += 2 * 60 * 1000; await alert(row, sentry, [drifted('search_issues')]); - expect(record).toHaveBeenCalledTimes(1); + expect(logged()).toHaveLength(2); }); // Review on c7d2fa765: two replicas inserting at once make one upsert lose diff --git a/backend/services/hostedMcpDriftAlertService.ts b/backend/services/hostedMcpDriftAlertService.ts index 344718491..640bda91d 100644 --- a/backend/services/hostedMcpDriftAlertService.ts +++ b/backend/services/hostedMcpDriftAlertService.ts @@ -53,11 +53,11 @@ export interface HostedMcpDriftAlertDeps { } /** - * An undeliverable drift (no config, or a recipient the pod does not list) is - * looked at again after this long, so a config set after the first error still - * delivers without a restart. A filed card is never filed again. + * 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_RECHECK_MS = 60 * 60 * 1000; +export const DRIFT_ALERT_RELOG_MS = 60 * 60 * 1000; /** * The pin set AND the drifted tools. A re-pin is a new set, so its next drift @@ -82,37 +82,54 @@ const configFrom = (value: unknown): HostedMcpDriftAlertConfig | null => { export const createHostedMcpDriftAlert = (deps: HostedMcpDriftAlertDeps) => { const now = deps.now || Date.now; - // Drifts this process has filed (until forever) or reported undeliverable - // (until a recheck). A second row seeing the same drift adds nothing, and - // the card's unique index holds across replicas. - const handledUntil = new Map(); + // 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; const sourceId = driftAlertSourceId(entry, drifted); - if ((handledUntil.get(sourceId) ?? 0) > now()) return; - const filed = () => handledUntil.set(sourceId, Infinity); - const recheckLater = () => handledUntil.set(sourceId, now() + DRIFT_ALERT_RECHECK_MS); // 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) => console.error(JSON.stringify({ - event: 'hosted_mcp_drift_alert', - delivery, - ...(reason ? { reason } : {}), - entryId: entry.id, - connectionId: row.connectionId, - tools: drifted, - })); + const report = (delivery: string, reason?: string) => { + const key = `${sourceId}|${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 filedKey = `${sourceId}|${config.recipientUserId}|${config.podId}`; + if (filed.has(filedKey)) return; try { - const config = configFrom(await deps.readConfig()); - if (!config) { - recheckLater(); - report('unconfigured'); - return; - } const result = await deps.record({ ...config, sourceId, @@ -120,20 +137,15 @@ export const createHostedMcpDriftAlert = (deps: HostedMcpDriftAlertDeps) => { 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(); - } else { - recheckLater(); - report(result.reason || 'not_recorded'); - } + 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(); + filed.add(filedKey); return; } - // Not marked handled: the next fresh assessment tries again. - report('failed', error instanceof Error ? error.message : String(error)); + failed(error); } }; }; diff --git a/backend/services/hostedMcpToolDefinitions.ts b/backend/services/hostedMcpToolDefinitions.ts index bf6640f47..3c2c7fac5 100644 --- a/backend/services/hostedMcpToolDefinitions.ts +++ b/backend/services/hostedMcpToolDefinitions.ts @@ -31,7 +31,6 @@ import { type HostedMcpUpstreamTool, } from './hostedMcpEntryService'; import { HostedMcpCredentialError } from './hostedMcpCredentialError'; -import { alertOnHostedMcpDrift } from './hostedMcpDriftAlertService'; import type { HostedMcpRow } from './hostedMcpCredentialService'; import type { ToolConnection, ToolDefinition } from './toolBrokerService'; @@ -481,7 +480,13 @@ export const createHostedMcpRuntimeDriftCache = ( entries: HostedMcpEntry[] = HOSTED_MCP_ENTRIES, ): HostedMcpDriftCache => createHostedMcpDriftCache({ listUpstreamTools: ({ connectionId }) => listUpstreamTools(connectionId, deps, entries), - onAssessed: alertOnHostedMcpDrift, + // 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({ From 7f47bfface1cca93d4a3a41ab2aa4a448ea8f885 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sun, 1 Nov 2026 16:09:58 -0800 Subject: [PATCH 4/4] fix(connectors): a moved drift-alert pod gets its own card The card is unique per recipient and source id and is only ever inserted, so a source id without the pod matched the card in the old pod and left the new pod empty. The pod is now part of the drift's source id. The error-line throttle keys on the drift alone, as before. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_015AKgpQ2pr6xRMhN5mcfoHF --- .../hostedMcpDriftAlertService.test.js | 15 ++++++++++++++ .../services/hostedMcpDriftAlertService.ts | 20 +++++++++++-------- 2 files changed, 27 insertions(+), 8 deletions(-) diff --git a/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js b/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js index 165ba9358..250697a15 100644 --- a/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js +++ b/backend/__tests__/unit/services/hostedMcpDriftAlertService.test.js @@ -79,6 +79,21 @@ describe('a drifted pin becomes one card for the operator', () => { 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 () => { diff --git a/backend/services/hostedMcpDriftAlertService.ts b/backend/services/hostedMcpDriftAlertService.ts index 640bda91d..a56a8ca14 100644 --- a/backend/services/hostedMcpDriftAlertService.ts +++ b/backend/services/hostedMcpDriftAlertService.ts @@ -60,13 +60,15 @@ export interface HostedMcpDriftAlertDeps { export const DRIFT_ALERT_RELOG_MS = 60 * 60 * 1000; /** - * The pin set AND the drifted tools. 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 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[]): string => { +export const driftAlertSourceId = (entry: HostedMcpEntry, drifted: string[], podId: string): string => { const pins = entry.tools.map((tool) => digestPinnedDefinition(entry.id, tool)).join('\n'); - const key = `${pins}\n--\n${[...drifted].sort().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)}`; }; @@ -93,12 +95,13 @@ export const createHostedMcpDriftAlert = (deps: HostedMcpDriftAlertDeps) => { 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; - const sourceId = driftAlertSourceId(entry, drifted); + // 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 = `${sourceId}|${delivery}`; + const key = `${driftKey}|${delivery}`; const last = reportedAt.get(key); if (last !== undefined && now() - last < DRIFT_ALERT_RELOG_MS) return; reportedAt.set(key, now()); @@ -126,7 +129,8 @@ export const createHostedMcpDriftAlert = (deps: HostedMcpDriftAlertDeps) => { report('unconfigured'); return; } - const filedKey = `${sourceId}|${config.recipientUserId}|${config.podId}`; + const sourceId = driftAlertSourceId(entry, drifted, config.podId); + const filedKey = `${sourceId}|${config.recipientUserId}`; if (filed.has(filedKey)) return; try {