diff --git a/backend/__tests__/unit/services/agentMentionService.threadRootOnMentions.test.js b/backend/__tests__/unit/services/agentMentionService.threadRootOnMentions.test.js new file mode 100644 index 0000000000..d5eac30ee8 --- /dev/null +++ b/backend/__tests__/unit/services/agentMentionService.threadRootOnMentions.test.js @@ -0,0 +1,170 @@ +/** + * TASK-026 / PR #2811: a hosted agent @-mentioned in a thread must receive the + * thread root on the event it is woken with. Before this, only the ambient + * message.posted path carried threadRootId, so a hosted reply landed at the pod + * root. This drives the real enqueueMentions path (no hand-built payload) and + * asserts the field on every chat.mention and thread.mention it enqueues. + */ +jest.mock('../../../services/agentEventService', () => ({ enqueue: jest.fn() })); +jest.mock('../../../models/AgentRegistry', () => ({ + AgentInstallation: { find: jest.fn(), findOne: jest.fn() }, +})); +jest.mock('../../../models/AgentProfile', () => ({ find: jest.fn() })); +jest.mock('../../../models/Pod', () => ({ findById: jest.fn(), find: jest.fn() })); +jest.mock('../../../models/User', () => ({ find: jest.fn(), findById: jest.fn() })); +jest.mock('../../../services/chatSummarizerService', () => ({ + constructor: { getLatestPodSummary: jest.fn() }, summarizePodMessages: jest.fn(), +})); +jest.mock('../../../models/AgentEvent', () => ({ countDocuments: jest.fn() })); +jest.mock('../../../services/welcomeWakeService', () => ({ maybeFireWelcomeWake: jest.fn() })); +jest.mock('../../../models/pg/Message', () => ({ + findById: jest.fn(async (id) => (String(id) === '101' ? { id: 101, user_id: 'bot-user-a' } : null)), +})); +jest.mock('../../../models/pg/ThreadUserState', () => ({ + followByParticipation: jest.fn(), +})); +jest.mock('../../../services/threadWakeScopeService', () => ({ + narrowToThread: jest.fn(async (_root, targets) => targets), + effectiveFollowerIds: jest.fn(async () => new Set()), +})); +jest.mock('../../../services/messageDeliveryModeService', () => ({ + addressedMessageTargets: jest.fn(async () => new Set()), +})); + +const AgentMentionService = require('../../../services/agentMentionService'); +const AgentEventService = require('../../../services/agentEventService'); +const { AgentInstallation } = require('../../../models/AgentRegistry'); +const AgentProfile = require('../../../models/AgentProfile'); +const Pod = require('../../../models/Pod'); +const User = require('../../../models/User'); +const AgentEvent = require('../../../models/AgentEvent'); +const { narrowToThread } = require('../../../services/threadWakeScopeService'); +const { addressedMessageTargets } = require('../../../services/messageDeliveryModeService'); +const ThreadUserState = require('../../../models/pg/ThreadUserState'); +const PGMessage = require('../../../models/pg/Message'); + +// `installedBy` is deliberately a DIFFERENT id from the bot's User row here. +// It is the human installer on five of the write paths (podController:119, +// personaHireService:93, podCurationService:147, authController:201, +// agentProfile:259) and the bot itself on the rest — so a fixture where the +// two coincide cannot tell a correct key from the wrong one. These differ so +// the assertion below discriminates. +const optIn = (agentName, installedBy) => ({ + agentName, + instanceId: 'default', + displayName: agentName, + installedBy, + config: { wakeOnMessage: { enabled: true } }, +}); + +// ONE id per agent across the whole fixture. @sprint-review (57037): seat-a +// used to be `bot-1` through User.findById (as the thread root's author) and +// `bot-user-a` through User.find (as a wake target) — the same agent with two +// identities depending on which lookup you went through. Nothing asserted the +// difference, so nothing failed; but a later test asking "is the seat that +// authored the root scoped out?" would have got contradictory answers from +// the two paths and read it as a scoping bug rather than a fixture bug. +// +// That is the same shape as the defect this suite exists for: an identity +// fixture that quietly disagrees with itself cannot discriminate a correct +// key from a wrong one. +const BOT_USER_ROWS = [ + { _id: 'bot-user-a', botMetadata: { agentName: 'seat-a', instanceId: 'default' } }, + { _id: 'bot-user-b', botMetadata: { agentName: 'seat-b', instanceId: 'default' } }, +]; +const HUMAN_MENTION_ROWS = [ + { _id: 'human-casey', username: 'Casey_Dev', isBot: false }, +]; + +const byType = (t) => AgentEventService.enqueue.mock.calls.map(([a]) => a).filter((a) => a.type === t); + +const send = (message, extra = {}) => AgentMentionService.enqueueMentions({ + podId: 'pod-1', userId: 'user-1', username: 'alice', message, ...extra, +}); + +beforeEach(() => { + jest.clearAllMocks(); + narrowToThread.mockImplementation(async (_root, targets) => targets); + addressedMessageTargets.mockResolvedValue(new Set()); + AgentInstallation.find.mockReturnValue({ + lean: jest.fn().mockResolvedValue([ + optIn('seat-a', 'human-installer-a'), optIn('seat-b', 'human-installer-b'), + ]), + }); + User.find.mockImplementation((filter) => ({ + select: jest.fn().mockReturnValue({ + lean: jest.fn().mockResolvedValue(filter.isBot === false ? HUMAN_MENTION_ROWS : BOT_USER_ROWS), + }), + })); + AgentProfile.find.mockReturnValue({ lean: jest.fn().mockResolvedValue([]) }); + Pod.findById.mockReturnValue({ + select: jest.fn().mockReturnValue({ + lean: jest.fn().mockResolvedValue({ + type: 'chat', members: ['user-1', 'bot-user-a', 'human-casey'], + }), + }), + lean: jest.fn().mockResolvedValue({ + _id: 'pod-1', type: 'chat', members: ['user-1', 'human-casey'], + }), + }); + User.findById.mockImplementation((id) => ({ + select: jest.fn().mockReturnThis(), + lean: jest.fn().mockResolvedValue(String(id) === 'bot-user-a' + ? { _id: 'bot-user-a', isBot: true, botMetadata: { agentName: 'seat-a', instanceId: 'default' } } + : { _id: 'user-1', isBot: false }), + })); + AgentEvent.countDocuments.mockResolvedValue(0); + ThreadUserState.followByParticipation.mockResolvedValue(true); +}); + +const mentionEvents = () => AgentEventService.enqueue.mock.calls + .map(([a]) => a) + .filter((a) => a.type === 'chat.mention' || a.type === 'thread.mention'); + +describe('a mention carries the thread root it was posted in (TASK-026)', () => { + test('a thread.mention to a hosted agent carries threadRootId', async () => { + await send({ + id: 'm2', _id: 'm2', content: '@seat-a please look', source: 'thread', thread_root_id: 101, + }); + const events = mentionEvents().filter((e) => e.agentName === 'seat-a'); + expect(events).toHaveLength(1); + expect(events[0].type).toBe('thread.mention'); + expect(events[0].payload.threadRootId).toBe(101); + }); + + test('a chat-sourced mention inside a thread carries threadRootId', async () => { + await send({ + id: 'm3', _id: 'm3', content: '@seat-a in the thread', source: 'chat', threadRootId: 101, + }); + const events = mentionEvents().filter((e) => e.agentName === 'seat-a'); + expect(events).toHaveLength(1); + expect(events[0].payload.threadRootId).toBe(101); + }); + + test('a human reply inside a thread to a hosted agent carries threadRootId (reply-edge path)', async () => { + await send( + { + id: 'm5', + _id: 'm5', + content: 'and the second one?', + source: 'chat', + thread_root_id: 101, + replyTo: { userId: 'bot-user-a' }, + }, + { replyToMessageId: 'r1' }, + ); + const events = mentionEvents().filter((e) => e.agentName === 'seat-a'); + expect(events).toHaveLength(1); + expect(events[0].payload.implicitReply).toBe(true); + expect(events[0].payload.threadRootId).toBe(101); + }); + + test('a top-level mention carries no threadRootId key at all', async () => { + await send({ + id: 'm4', _id: 'm4', content: '@seat-a at the root', source: 'chat', + }); + const events = mentionEvents().filter((e) => e.agentName === 'seat-a'); + expect(events).toHaveLength(1); + expect(events[0].payload).not.toHaveProperty('threadRootId'); + }); +}); diff --git a/backend/__tests__/unit/services/nativeRuntimeService.threadRoot.test.js b/backend/__tests__/unit/services/nativeRuntimeService.threadRoot.test.js new file mode 100644 index 0000000000..64d5623f17 --- /dev/null +++ b/backend/__tests__/unit/services/nativeRuntimeService.threadRoot.test.js @@ -0,0 +1,182 @@ +/** + * TASK-026 — a hosted agent's reply to a mention made inside a thread must + * carry that thread's root, or it lands at the pod root. + */ +jest.mock('axios', () => ({ __esModule: true, default: { post: jest.fn() } })); +jest.mock('../../../models/Pod', () => ({ + findById: jest.fn(() => ({ lean: jest.fn().mockResolvedValue({ name: 'My Workspace' }) })), +})); +jest.mock('../../../models/AgentRun', () => ({ create: jest.fn(), countDocuments: jest.fn() })); +jest.mock('../../../models/User', () => ({ + findOne: jest.fn(() => ({ select: () => ({ lean: () => Promise.resolve(null) }) })), + findById: jest.fn(() => ({ select: () => ({ lean: () => Promise.resolve({ username: 'ada' }) }) })), +})); +jest.mock('../../../services/hostedRuntimeService', () => ({ + ...jest.requireActual('../../../services/hostedRuntimeService'), + listHostedInstallationsForUser: jest.fn().mockResolvedValue([]), +})); +jest.mock('../../../services/agentTypingService', () => ({ emitAgentTypingStart: jest.fn(), emitAgentTypingStop: jest.fn() })); +jest.mock('../../../services/agentMessageService', () => ({ postMessage: jest.fn().mockResolvedValue({ success: true }) })); +jest.mock('../../../services/messageClaimService', () => ({ + claim: jest.fn().mockResolvedValue({ claimed: true, expiresAt: new Date() }), + release: jest.fn().mockResolvedValue({ released: true }), +})); +jest.mock('../../../services/nativeOutsiderGrantPolicy', () => ({ + podHasOutsiderWritableGrant: jest.fn().mockResolvedValue(false), +})); + +const axios = require('axios').default; +const AgentRun = require('../../../models/AgentRun'); +const AgentMessageService = require('../../../services/agentMessageService'); +const { threadRootOfTrigger, runAgent } = require('../../../services/nativeRuntimeService'); + +describe('threadRootOfTrigger (TASK-026)', () => { + test('a mention inside a thread takes its root (camelCase, as the mention payload writes it)', () => { + expect(threadRootOfTrigger({ messageId: 9, threadRootId: 4 })).toBe(4); + }); + + test('the snake_case field the message row carries is read too', () => { + expect(threadRootOfTrigger({ thread_root_id: '7' })).toBe(7); + }); + + test('a top-level mention has no root, so the answer stays at the pod root', () => { + expect(threadRootOfTrigger({ messageId: 9 })).toBeUndefined(); + expect(threadRootOfTrigger({ messageId: 9, threadRootId: null })).toBeUndefined(); + }); + + test('a missing or malformed payload has no root', () => { + expect(threadRootOfTrigger(null)).toBeUndefined(); + expect(threadRootOfTrigger(undefined)).toBeUndefined(); + expect(threadRootOfTrigger({ threadRootId: 'abc' })).toBeUndefined(); + expect(threadRootOfTrigger({ threadRootId: 0 })).toBeUndefined(); + }); +}); + +// The trigger-to-post handoff: a threaded trigger must reach the post call, +// and a top-level trigger must post with null. Reverting either the tool +// post or the fallback post's threadRootId argument fails one of these. +const POD = '507f1f77bcf86cd799439011'; +const OWNER = '507f1f77bcf86cd799439099'; +const seat = { + podId: POD, + agentName: 'scout', + instanceId: 'u1234567890', + displayName: 'Scout', + installedBy: OWNER, + config: { runtime: { runtimeType: 'native' }, tools: ['commonly_post_message'] }, +}; +const threaded = { + type: 'chat.mention', + eventId: '507f1f77bcf86cd799439012', + payload: { messageId: 9, threadRootId: 4, content: 'answer me in the thread' }, +}; +const topLevel = { + type: 'chat.mention', + eventId: '507f1f77bcf86cd799439013', + payload: { messageId: 10, content: 'answer me at the root' }, +}; +const callsTool = (name, args) => ({ + status: 200, + headers: {}, + data: { + choices: [{ message: { role: 'assistant', content: null, tool_calls: [{ id: 'c1', type: 'function', function: { name, arguments: JSON.stringify(args) } }] } }], + usage: { prompt_tokens: 10, completion_tokens: 8, total_tokens: 18 }, + }, +}); +const says = (content) => ({ + status: 200, + headers: {}, + data: { choices: [{ message: { role: 'assistant', content } }], usage: { prompt_tokens: 12, completion_tokens: 2, total_tokens: 14 } }, +}); + +describe('runAgent carries the trigger thread root to its post calls (TASK-026)', () => { + beforeEach(() => { + jest.clearAllMocks(); + process.env.LITELLM_BASE_URL = 'http://litellm.test'; + process.env.LITELLM_MASTER_KEY = 'test-key'; + process.env.HOSTED_ENGINE = 'native'; + AgentRun.countDocuments.mockResolvedValue(0); + AgentRun.create.mockImplementation(async (doc) => ({ _id: 'run-1', ...doc, save: jest.fn().mockResolvedValue(undefined) })); + }); + + test('the tool post in a thread passes the thread root', async () => { + axios.post + .mockResolvedValueOnce(callsTool('commonly_post_message', { content: 'in the thread' })) + .mockResolvedValueOnce(says('Posted.')); + + await runAgent(seat, threaded); + + expect(AgentMessageService.postMessage).toHaveBeenCalledTimes(1); + expect(AgentMessageService.postMessage.mock.calls[0][0].threadRootId).toBe(4); + }); + + test('the tool post at top level passes null, so the answer stays at the root', async () => { + axios.post + .mockResolvedValueOnce(callsTool('commonly_post_message', { content: 'at the root' })) + .mockResolvedValueOnce(says('Posted.')); + + await runAgent(seat, topLevel); + + expect(AgentMessageService.postMessage).toHaveBeenCalledTimes(1); + expect(AgentMessageService.postMessage.mock.calls[0][0].threadRootId).toBeNull(); + }); + + test('the fallback post in a thread passes the thread root when the model never calls a tool', async () => { + axios.post.mockResolvedValueOnce(says('Plain reply, no tool.')); + + await runAgent(seat, threaded); + + const fallback = AgentMessageService.postMessage.mock.calls.find(([arg]) => arg.metadata?.fallback); + expect(fallback).toBeDefined(); + expect(fallback[0].threadRootId).toBe(4); + }); +}); + +describe('runAgent carries the trigger thread root to the cards and the failed-run notice (TASK-026)', () => { + const hostedSeat = { ...seat, config: { ...seat.config, tools: ['commonly_post_message', 'commonly_offer_hosted_agent'] } }; + const localSeat = { ...seat, config: { ...seat.config, tools: ['commonly_post_message', 'commonly_offer_local_agent'] } }; + + beforeEach(() => { + jest.clearAllMocks(); + process.env.LITELLM_BASE_URL = 'http://litellm.test'; + process.env.LITELLM_MASTER_KEY = 'test-key'; + process.env.HOSTED_ENGINE = 'native'; + AgentRun.countDocuments.mockResolvedValue(0); + AgentRun.create.mockImplementation(async (doc) => ({ _id: 'run-1', ...doc, save: jest.fn().mockResolvedValue(undefined) })); + }); + + // Each test fails if its single threadRootId argument is reverted to null. + test('the hosted-offer starts card in a thread passes the thread root', async () => { + axios.post + .mockResolvedValueOnce(callsTool('commonly_offer_hosted_agent', { content: 'Scout can run hosted.' })) + .mockResolvedValueOnce(says('Offered.')); + + await runAgent(hostedSeat, threaded); + + const card = AgentMessageService.postMessage.mock.calls.find(([arg]) => arg.payload?.kind === 'starts'); + expect(card).toBeDefined(); + expect(card[0].threadRootId).toBe(4); + }); + + test('the BYO approval card in a thread passes the thread root', async () => { + axios.post + .mockResolvedValueOnce(callsTool('commonly_offer_local_agent', { adapter: 'claude', os: 'mac' })) + .mockResolvedValueOnce(says('Offered.')); + + await runAgent(localSeat, threaded); + + const card = AgentMessageService.postMessage.mock.calls.find(([arg]) => arg.metadata?.handoff === 'byo'); + expect(card).toBeDefined(); + expect(card[0].threadRootId).toBe(4); + }); + + test('the failed-run notice for a directly addressed thread mention passes the thread root', async () => { + axios.post.mockRejectedValueOnce(Object.assign(new Error('socket hang up'), { code: 'ECONNRESET' })); + + await runAgent(seat, threaded); + + const notice = AgentMessageService.postMessage.mock.calls.find(([arg]) => arg.metadata?.runFailed); + expect(notice).toBeDefined(); + expect(notice[0].threadRootId).toBe(4); + }); +}); diff --git a/backend/services/agentMentionService.ts b/backend/services/agentMentionService.ts index 06f737a9e0..8af1d33f33 100644 --- a/backend/services/agentMentionService.ts +++ b/backend/services/agentMentionService.ts @@ -1579,6 +1579,7 @@ const enqueueMentions = async ({ const rawContent = message?.content || message?.text || ''; const source = message?.source || 'chat'; const eventType: (typeof MENTION_EVENT_TYPES)[number] = source === 'thread' ? 'thread.mention' : 'chat.mention'; + const threadRootId = resolveThreadRootId(message); // Author/age frame inputs — identical for every target of this // message, so resolved once here rather than at each of the four // enqueue sites below. Same values that already go into the envelope @@ -1723,6 +1724,7 @@ const enqueueMentions = async ({ messageType: message?.messageType || message?.message_type || 'text', createdAt: message?.createdAt || message?.created_at || new Date(), thread: message?.thread || null, + ...(threadRootId ? { threadRootId } : {}), implicitReply: true, adjacentReply: true, }, @@ -1981,6 +1983,7 @@ const enqueueMentions = async ({ messageType: message?.messageType || message?.message_type || 'text', createdAt: message?.createdAt || message?.created_at || new Date(), thread: message?.thread || null, + ...(threadRootId ? { threadRootId } : {}), }, }); recordEnqueued(directMatch); @@ -2032,6 +2035,7 @@ const enqueueMentions = async ({ messageType: message?.messageType || message?.message_type || 'text', createdAt: message?.createdAt || message?.created_at || new Date(), thread: message?.thread || null, + ...(threadRootId ? { threadRootId } : {}), autoJoined: true, }, }); @@ -2106,6 +2110,7 @@ const enqueueMentions = async ({ messageType: message?.messageType || message?.message_type || 'text', createdAt: message?.createdAt || message?.created_at || new Date(), thread: message?.thread || null, + ...(threadRootId ? { threadRootId } : {}), }, }); recordEnqueued({ agentName: agentType, instanceId: match.instanceId || 'default' }); @@ -2124,7 +2129,6 @@ const enqueueMentions = async ({ // One BOT User lookup serves both the delivered agent mention below and the // routed message's ambient wake fan-out. An auto-joined target is not in // the initial installation list yet, so include the actual delivery set. - const threadRootId = resolveThreadRootId(message); let resolvedBotUserIds: Map | undefined; if (threadRootId) { const followedUserIds = await resolveHumanMentionUserIds(humanMentionHandles, podMemberIds); @@ -2181,6 +2185,7 @@ const enqueueMentions = async ({ messageType: message?.messageType || message?.message_type || 'text', createdAt: message?.createdAt || message?.created_at || new Date(), thread: message?.thread || null, + ...(threadRootId ? { threadRootId } : {}), replyToMessageId, implicitReply: true, }, diff --git a/backend/services/nativeRuntimeService.ts b/backend/services/nativeRuntimeService.ts index 251d380263..08ddf158c8 100644 --- a/backend/services/nativeRuntimeService.ts +++ b/backend/services/nativeRuntimeService.ts @@ -677,11 +677,26 @@ interface DispatchContext { * event, never from the model. Every post the turn makes carries it so the * answer is credited to that message's author even when the turn runs * unclaimed (an addressed agent that lost the claim race, #2269, or a claim - * service that failed open). It is not a reply edge and threads nothing. + * service that failed open). It credits the author; it does not place the + * reply. Placement is answeringThreadRootId. */ answeringMessageId?: string; + /** + * The thread the trigger was posted in, when it was posted in one. Every + * post the turn makes carries it, so a mention inside a thread gets its + * answer in that thread (TASK-026). Undefined for a top-level trigger, which + * keeps the answer at the pod root. + */ + answeringThreadRootId?: number; } +/** The thread root a trigger event was posted in, or undefined at the pod root. */ +export const threadRootOfTrigger = (payload: unknown): number | undefined => { + const p = (payload || {}) as { threadRootId?: unknown; thread_root_id?: unknown }; + const root = Number(p.threadRootId ?? p.thread_root_id); + return Number.isInteger(root) && root > 0 ? root : undefined; +}; + // The agent's own bot User, found the way the posting path derives it (the // same two functions as the broker's seat lookup below). Its id is what marks // the agent's own pod posts. Not its names: a display name drifts by design (a @@ -823,6 +838,7 @@ export async function dispatchTool( // bridge appears between this check and message storage. noOutboundRelay: Boolean(ctx.webTainted || ctx.brokerCallSeen), answeringMessageId: ctx.answeringMessageId, + threadRootId: ctx.answeringThreadRootId ?? null, runId: ctx.runId, ...(ctx.runId ? { runPodId: String(ctx.podId) } : {}), }); @@ -888,6 +904,7 @@ export async function dispatchTool( payload: { kind: 'starts', starts: [{ id: 'run-hosted', label: HOSTED_OFFER_LABEL, href }] }, noOutboundRelay: Boolean(ctx.webTainted || ctx.brokerCallSeen), answeringMessageId: ctx.answeringMessageId, + threadRootId: ctx.answeringThreadRootId ?? null, runId: ctx.runId, ...(ctx.runId ? { runPodId: String(ctx.podId) } : {}), }); @@ -937,6 +954,7 @@ export async function dispatchTool( payload, noOutboundRelay: Boolean(ctx.webTainted || ctx.brokerCallSeen), answeringMessageId: ctx.answeringMessageId, + threadRootId: ctx.answeringThreadRootId ?? null, runId: ctx.runId, ...(ctx.runId ? { runPodId: String(ctx.podId) } : {}), authorKind: 'kernel', @@ -1873,6 +1891,7 @@ export async function runAgent( ownUserId: null, givenAddresses: new GivenAddresses(), answeringMessageId: claimMessageId || undefined, + answeringThreadRootId: threadRootOfTrigger(trigger.payload), }; // Who wrote what matters only to commonly_fetch_url, so a turn that cannot // fetch makes neither lookup (its own user, the trigger's sender). @@ -2356,6 +2375,7 @@ export async function runAgent( metadata: { source: 'native-runtime', fallback: true }, noOutboundRelay: Boolean(dispatchCtx.webTainted || dispatchCtx.brokerCallSeen), answeringMessageId: dispatchCtx.answeringMessageId, + threadRootId: dispatchCtx.answeringThreadRootId ?? null, runId, runPodId: String(podId), }); @@ -2407,6 +2427,7 @@ export async function runAgent( metadata: { source: 'native-runtime', runFailed: true }, noOutboundRelay: Boolean(dispatchCtx.webTainted || dispatchCtx.brokerCallSeen), answeringMessageId: dispatchCtx.answeringMessageId, + threadRootId: dispatchCtx.answeringThreadRootId ?? null, runId, runPodId: String(podId), });