diff --git a/backend/__tests__/unit/models/PgMessage.test.js b/backend/__tests__/unit/models/PgMessage.test.js index be071d1288..2db78953e3 100644 --- a/backend/__tests__/unit/models/PgMessage.test.js +++ b/backend/__tests__/unit/models/PgMessage.test.js @@ -81,6 +81,30 @@ describe('PG Message model', () => { }); }); + // The message the posting run is answering (agentAnswerService) reaches the + // attention writer with the row and is never stored: the INSERT carries the + // same seven parameters, and no reply edge. + it("hands the run's answeringMessageId to the attention writer without storing it", async () => { + pool.query + .mockResolvedValueOnce({ + rows: [{ + id: '3', pod_id: 'pod-1', user_id: 'agent', content: '@lee done', message_type: 'text', + }], + }) + .mockResolvedValueOnce({}); + + await Message.create('pod-1', 'agent', '@lee done', 'text', null, null, null, '78385'); + + expect(pool.query).toHaveBeenNthCalledWith( + 1, + expect.stringContaining('INSERT INTO messages'), + ['pod-1', 'agent', '@lee done', 'text', null, null, null], + ); + expect(mockRecordMentionedUsers).toHaveBeenCalledWith(expect.objectContaining({ + id: '3', answeringMessageId: '78385', + })); + }); + it('findByPodId formats returned rows', async () => { pool.query.mockResolvedValueOnce({ rows: [ diff --git a/backend/__tests__/unit/routes/agentsRuntime.post-message.config.test.js b/backend/__tests__/unit/routes/agentsRuntime.post-message.config.test.js index d6534f8441..ba6440f869 100644 --- a/backend/__tests__/unit/routes/agentsRuntime.post-message.config.test.js +++ b/backend/__tests__/unit/routes/agentsRuntime.post-message.config.test.js @@ -94,4 +94,29 @@ describe('agentsRuntime post message installation config forwarding', () => { })); expect(res.json).toHaveBeenCalledWith({ success: true }); }); + + // answeringMessageId tells the answer rule which message a post answers + // (agentAnswerService): it completes "Give your agent its first task" and + // suppresses a needs-you row. Only the native runtime sets it, from its run's + // own trigger. A CAP request (and every MCP tool, which posts through this + // route) cannot name one, or any agent could close someone's onboarding card. + it('never forwards answeringMessageId from a request body', async () => { + const handler = getRouteHandler('/pods/:podId/messages', 'post'); + AgentMessageService.postMessage.mockResolvedValue({ success: true }); + const installation = { + podId: 'pod-1', agentName: 'openclaw', instanceId: 'tarik', displayName: 'Tarik', config: {}, + }; + const req = { + params: { podId: 'pod-1' }, + body: { content: 'hello', messageType: 'text', answeringMessageId: '78385' }, + agentInstallation: installation, + agentInstallations: [installation], + }; + const res = { status: jest.fn().mockReturnThis(), json: jest.fn() }; + + await handler(req, res); + + expect(AgentMessageService.postMessage).toHaveBeenCalledTimes(1); + expect(AgentMessageService.postMessage.mock.calls[0][0]).not.toHaveProperty('answeringMessageId'); + }); }); diff --git a/backend/__tests__/unit/services/agentAnswerService.test.js b/backend/__tests__/unit/services/agentAnswerService.test.js new file mode 100644 index 0000000000..8f69d76329 --- /dev/null +++ b/backend/__tests__/unit/services/agentAnswerService.test.js @@ -0,0 +1,93 @@ +/** + * agentAnswerService.humansAnsweredBy: who an agent's post answers. + * + * QA walk 5 (2026-10-16): since #2269 an agent a person @-mentioned answers + * even when another agent (Scout) won the claim race, so it holds no claim and + * posts no reply edge. Its first answer then answered nobody, and "Give your + * agent its first task" completed only on the second. The native run now names + * the message it is answering (answeringMessageId), which answers its author + * when that message is in the same pod. + */ + +const mockLiveClaims = jest.fn(); +jest.mock('../../../services/messageClaimService', () => ({ liveClaimsHeldBy: (...args) => mockLiveClaims(...args) })); +const mockPgFindById = jest.fn(); +jest.mock('../../../models/pg/Message', () => ({ findById: (...args) => mockPgFindById(...args) })); +const mockTaskExists = jest.fn(); +const mockTaskUpdate = jest.fn(); +jest.mock('../../../models/Task', () => ({ + exists: (...args) => mockTaskExists(...args), + findOneAndUpdate: (...args) => mockTaskUpdate(...args), +})); +const mockPodFindById = jest.fn(); +jest.mock('../../../models/Pod', () => ({ findById: (...args) => mockPodFindById(...args) })); +jest.mock('../../../services/taskEventService', () => ({ emitTaskUpdated: jest.fn() })); + +const { humansAnsweredBy } = require('../../../services/agentAnswerService'); +const { completeFirstTaskOnAnswer, FIRST_TASK_SOURCE_REF } = require('../../../services/starterTaskService'); + +const POD = 'aaaaaaaaaaaaaaaaaaaaaaaa'; +const OWNER = 'cccccccccccccccccccccccc'; +const messages = { + 78385: { id: 78385, pod_id: POD, userId: { _id: OWNER, username: 'stranger' } }, + 78390: { id: 78390, pod_id: 'bbbbbbbbbbbbbbbbbbbbbbbb', userId: { _id: OWNER, username: 'stranger' } }, +}; + +beforeEach(() => { + mockLiveClaims.mockReset().mockResolvedValue([]); + mockPgFindById.mockReset().mockImplementation(async (id) => messages[id] || null); + mockTaskExists.mockReset().mockResolvedValue({ _id: 't2' }); + mockTaskUpdate.mockReset().mockResolvedValue({ taskId: 'TASK-002', status: 'done' }); + mockPodFindById.mockReset().mockReturnValue({ select: () => ({ lean: async () => ({ createdBy: OWNER }) }) }); +}); + +const ask = (extra = {}) => humansAnsweredBy({ + podId: POD, agentName: 'stranger-agent', instanceId: 'default', ...extra, +}); + +describe('humansAnsweredBy', () => { + test('with no claim and no reply edge, a post answers nobody (the walk-5 miss)', async () => { + expect([...await ask()]).toEqual([]); + }); + + test("the message the run is answering answers its author, with no claim held", async () => { + expect([...await ask({ answeringMessageId: '78385' })]).toEqual([OWNER]); + }); + + test("a run's trigger in another pod answers nobody here", async () => { + expect([...await ask({ answeringMessageId: '78390' })]).toEqual([]); + }); + + test('a claimed trigger is read once, not twice', async () => { + mockLiveClaims.mockResolvedValue(['78385']); + expect([...await ask({ answeringMessageId: '78385' })]).toEqual([OWNER]); + expect(mockPgFindById).toHaveBeenCalledTimes(1); + }); + + test('an id that is not a message id is ignored', async () => { + expect([...await ask({ answeringMessageId: 'not-a-message' })]).toEqual([]); + expect(mockPgFindById).not.toHaveBeenCalled(); + }); +}); + +describe('"Give your agent its first task" with the real answer rule', () => { + // The walk-5 shape: Scout won the claim on the stranger's first ask, the + // stranger's hosted agent answered anyway (#2269), and the card must complete + // on that first answer, not the second. + test('an addressed agent that lost the claim completes it with its first answer', async () => { + await completeFirstTaskOnAnswer({ + podId: POD, agentName: 'stranger-agent', instanceId: 'default', agentLabel: 'stranger-agent', + content: 'Here is the summary.', answeringMessageId: '78385', + }); + expect(mockTaskUpdate).toHaveBeenCalledTimes(1); + expect(mockTaskUpdate.mock.calls[0][0].sourceRef).toBe(FIRST_TASK_SOURCE_REF); + }); + + test('without the run naming its trigger, the same answer leaves it open', async () => { + await completeFirstTaskOnAnswer({ + podId: POD, agentName: 'stranger-agent', instanceId: 'default', agentLabel: 'stranger-agent', + content: 'Here is the summary.', + }); + expect(mockTaskUpdate).not.toHaveBeenCalled(); + }); +}); diff --git a/backend/__tests__/unit/services/agentMessageService.threadWrites.test.js b/backend/__tests__/unit/services/agentMessageService.threadWrites.test.js index 63abf97ed1..ab483c73c5 100644 --- a/backend/__tests__/unit/services/agentMessageService.threadWrites.test.js +++ b/backend/__tests__/unit/services/agentMessageService.threadWrites.test.js @@ -153,7 +153,7 @@ describe('agent in-thread posts (threadRootId, no reply edge)', () => { podId: POD, replyToMessageId: null, threadRootId: '57577', }); expect(PGMessage.create).toHaveBeenCalledWith( - POD, 'agent-user-1', 'in-thread continuation', 'text', null, null, 57577, + POD, 'agent-user-1', 'in-thread continuation', 'text', null, null, 57577, undefined, ); const msg = emitted.find((e) => e.event === 'newMessage'); expect(msg.payload.thread_root_id).toBe('57577'); @@ -169,7 +169,7 @@ describe('agent in-thread posts (threadRootId, no reply edge)', () => { expect(threadRootResolver.resolveThreadRoot).not.toHaveBeenCalled(); expect(PGMessage.create).toHaveBeenCalledWith( - POD, 'agent-user-1', 'plain broadcast', 'text', null, null, null, + POD, 'agent-user-1', 'plain broadcast', 'text', null, null, null, undefined, ); }); @@ -197,7 +197,7 @@ describe('agent in-thread posts (threadRootId, no reply edge)', () => { }); expect(PGMessage.create).toHaveBeenCalledWith( - POD, 'agent-user-1', content, 'text', null, null, null, + POD, 'agent-user-1', content, 'text', null, null, null, undefined, ); expect(mockDispatch).toHaveBeenCalledWith('chat.message', expect.objectContaining({ podId: POD, @@ -288,6 +288,26 @@ describe('agent in-thread posts (threadRootId, no reply edge)', () => { }); }); + // Since #2269 an agent a person addressed answers even when another agent won + // the claim, so it holds none. The native run names the message it is + // answering; it reaches both answer readers and is never a reply edge. + it('hands answeringMessageId to the write and the first-task check, never as a reply edge', async () => { + await AgentMessageService.postMessage({ + agentName: 'lee-agent', + instanceId: 'default', + podId: POD, + content: 'Here is the summary.', + displayName: 'Lee Agent', + answeringMessageId: '78385', + }); + expect(PGMessage.create).toHaveBeenCalledWith( + POD, 'agent-user-1', 'Here is the summary.', 'text', null, null, null, '78385', + ); + expect(mockFirstTask).toHaveBeenCalledWith(expect.objectContaining({ + replyToMessageId: null, answeringMessageId: '78385', + })); + }); + it('a ThreadRootError propagates — never swallowed into the Mongo fallback', async () => { threadRootResolver.resolveThreadRoot.mockRejectedValue( new threadRootResolver.ThreadRootError('root is in another pod', 'thread_root_other_pod'), diff --git a/backend/__tests__/unit/services/attentionItemService.test.js b/backend/__tests__/unit/services/attentionItemService.test.js index 778a647af5..b03f414754 100644 --- a/backend/__tests__/unit/services/attentionItemService.test.js +++ b/backend/__tests__/unit/services/attentionItemService.test.js @@ -556,9 +556,10 @@ describe('an agent answering the person who asked', () => { mockUserFind.mockReturnValue(chain(members)); mockUserFindById.mockReturnValue(chain(scoutBot)); mockLiveClaims.mockResolvedValue(['900']); - mockPgMessageFindById.mockImplementation(async (id) => ( - id === '900' ? { id: 900, userId: { _id: 'lee', username: 'lee' } } : null - )); + mockPgMessageFindById.mockImplementation(async (id) => ({ + 900: { id: 900, pod_id: POD, userId: { _id: 'lee', username: 'lee' } }, + 950: { id: 950, pod_id: 'pod-elsewhere', userId: { _id: 'lee', username: 'lee' } }, + })[id] || null); }); const recipients = () => mockUpdateOne.mock.calls.map(([filter]) => filter.recipientUserId); @@ -583,6 +584,21 @@ describe('an agent answering the person who asked', () => { expect(mockUpdateOne).not.toHaveBeenCalled(); }); + // QA walk 5 (2026-10-16): since #2269 an agent a person addressed answers even + // when another agent won the claim, so it holds none. The native run names + // the message it is answering, and that answers its author. + it('an addressed agent that lost the claim answers them: no row', async () => { + mockLiveClaims.mockResolvedValue([]); + await answer('@lee here is the summary.', { answeringMessageId: '900' }); + expect(mockUpdateOne).not.toHaveBeenCalled(); + }); + + it("a run's trigger from another pod answers nobody here", async () => { + mockLiveClaims.mockResolvedValue([]); + await answer('@lee here is the summary.', { answeringMessageId: '950' }); + expect(recipients()).toEqual(['lee']); + }); + it('a lapsed claim, or an unprompted ping, still opens their row', async () => { mockLiveClaims.mockResolvedValue([]); await answer('@lee the deploy failed.'); diff --git a/backend/__tests__/unit/services/starterTaskService.test.js b/backend/__tests__/unit/services/starterTaskService.test.js index e95c43accd..efb1f0df28 100644 --- a/backend/__tests__/unit/services/starterTaskService.test.js +++ b/backend/__tests__/unit/services/starterTaskService.test.js @@ -129,6 +129,12 @@ describe('completeFirstTaskOnAnswer', () => { expect(mockEmitTaskUpdated).toHaveBeenCalledWith(POD_A, expect.objectContaining({ taskId: 'TASK-002' }), 'updated'); }); + test("forwards the run's answeringMessageId to the answer rule", async () => { + await answer({ answeringMessageId: '78385' }); + expect(mockAnswered).toHaveBeenCalledWith(expect.objectContaining({ answeringMessageId: '78385' })); + expect(mockFindOneAndUpdate).toHaveBeenCalledTimes(1); + }); + test('a post that answers someone else, or nobody, leaves it open', async () => { mockAnswered.mockResolvedValue(new Set(['dddddddddddddddddddddddd'])); await answer(); diff --git a/backend/models/pg/Message.ts b/backend/models/pg/Message.ts index df8f18cd44..e1f86f8c3d 100644 --- a/backend/models/pg/Message.ts +++ b/backend/models/pg/Message.ts @@ -104,6 +104,9 @@ class Message { // reply edge, which is the pre-existing behaviour and what the backfill // and explicit replies rely on. resolvedThreadRootId: number | null = null, + // The message the posting run is answering (agentAnswerService), handed to + // the attention writer and never stored: not a column, not a reply edge. + answeringMessageId: string | null = null, ): Promise { console.log('Creating message with params:', { podId, userId, content, messageType, podIdType: typeof podId, @@ -166,7 +169,9 @@ class Message { recordMentionedUsers, resolveMentionAttentionForReply, } = require('../../services/attentionItemService'); - await recordMentionedUsers({ ...row, podId, userId, content, threadRootId: resolvedThreadRootId }); + await recordMentionedUsers({ + ...row, podId, userId, content, threadRootId: resolvedThreadRootId, answeringMessageId, + }); await resolveMentionAttentionForReply({ podId, recipientUserId: userId, diff --git a/backend/services/agentAnswerService.ts b/backend/services/agentAnswerService.ts index cd840745e0..3fa5062fc9 100644 --- a/backend/services/agentAnswerService.ts +++ b/backend/services/agentAnswerService.ts @@ -10,8 +10,12 @@ * * An agent's post answers the authors of the messages it holds a LIVE claim on * in that pod (a native turn claims the message that woke it and carries no - * reply edge), plus the author of a message it explicitly replies to. Reply - * edges are never added to make this fire: that would thread every answer. + * reply edge), the author of a message it explicitly replies to, and the + * author of the message its run is answering (answeringMessageId). That last + * one is set by the native runtime from the run's own trigger, never by the + * model or a CAP request: since #2269 an agent a person addressed answers even + * when another agent won the claim, and then holds no claim to be counted by. + * Reply edges are never added to make this fire: that would thread every answer. * * Callers decide what the answer means; this only names who was answered. Any * failure answers nobody, so every caller falls back to its old behaviour. @@ -19,13 +23,16 @@ // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports const PGMessage = require('../models/pg/Message'); +const isPgMessageId = (id: string): boolean => /^\d{1,10}$/.test(id); + export const humansAnsweredBy = async ({ - podId, agentName, instanceId, replyToMessageId, + podId, agentName, instanceId, replyToMessageId, answeringMessageId, }: { podId: unknown; agentName: string; instanceId?: string; replyToMessageId?: unknown; + answeringMessageId?: unknown; }): Promise> => { const answered = new Set(); if (!podId || !agentName) return answered; @@ -36,12 +43,22 @@ export const humansAnsweredBy = async ({ podId: String(podId), agentName, instanceId: instanceId || 'default', }); const sourceIds = [...new Set([...claimed, ...(replyToMessageId ? [String(replyToMessageId)] : [])])] - .filter((id) => /^\d{1,10}$/.test(id)); - const sources = await Promise.all(sourceIds.map((id) => PGMessage.findById(id))); - sources.forEach((source: any) => { + .filter(isPgMessageId); + const answering = answeringMessageId ? String(answeringMessageId) : ''; + const answeringId = isPgMessageId(answering) && !sourceIds.includes(answering) ? answering : ''; + const [sources, answeringSource] = await Promise.all([ + Promise.all(sourceIds.map((id) => PGMessage.findById(id))), + answeringId ? PGMessage.findById(answeringId) : null, + ]); + const addAuthor = (source: any) => { const author = source?.userId?._id ?? source?.userId; if (author) answered.add(String(author)); - }); + }; + sources.forEach(addAuthor); + // A claim is taken in this pod, and a reply edge is checked against it at + // write (TASK-278). The run's trigger is named by id alone, so its pod is + // checked here: a post never answers a message in another pod. + if (answeringSource && String((answeringSource as any).pod_id) === String(podId)) addAuthor(answeringSource); } catch (error) { // Nothing is added before the last await, so a failure leaves the set empty. console.warn('[agent-answer] answer check failed; treating as answering nobody:', (error as Error).message); diff --git a/backend/services/agentMessageService.ts b/backend/services/agentMessageService.ts index c97b8dff72..a87343c606 100644 --- a/backend/services/agentMessageService.ts +++ b/backend/services/agentMessageService.ts @@ -182,6 +182,10 @@ interface PostMessageOptions { // sets it (TASK-267a, Rhea on #2244), and the relay dispatcher refuses such // a message whatever bridges exist by the time it is stored. noOutboundRelay?: boolean; + // Server-set, never model-set: the message the posting run is answering, + // the native runtime's trigger. Read by the answer rule (agentAnswerService) + // only. Never stored and never a reply edge, so an answer is not threaded. + answeringMessageId?: string | null; } interface PostTargetOptions { @@ -200,6 +204,7 @@ interface PostTargetOptions { skipDeliveryUpdate?: boolean; skipSummaryPersistence?: boolean; noOutboundRelay?: boolean; + answeringMessageId?: string | null; } interface HeartbeatFallbackOptions { @@ -992,6 +997,7 @@ class AgentMessageService { threadRootId = null, installationConfig = null, noOutboundRelay = false, + answeringMessageId, } = options; // Typing indicator cleanup — once postMessage is entered, the runtime has @@ -1572,6 +1578,7 @@ class AgentMessageService { replyToMessageId, threadRootId, noOutboundRelay, + answeringMessageId, }); // The first answer an agent gives the workspace owner completes "Give your @@ -1579,7 +1586,7 @@ class AgentMessageService { // write never delays or fails a post, and it never throws. // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports void require('./starterTaskService').completeFirstTaskOnAnswer({ - podId: String(podId), agentName, instanceId, replyToMessageId, + podId: String(podId), agentName, instanceId, replyToMessageId, answeringMessageId, agentLabel: displayName || agentName, content: sanitizedContent, }); @@ -1653,6 +1660,7 @@ class AgentMessageService { skipDeliveryUpdate = false, skipSummaryPersistence = false, noOutboundRelay = false, + answeringMessageId, } = options; // The installation label belongs to this pod; the User label belongs to @@ -1706,7 +1714,10 @@ class AgentMessageService { console.warn('[agent-msg] PG pod backfill skipped:', (syncErr as Error).message); } const newMessage = await (PGMessage as { - create(podId: string, userId: string, content: string, type: string, replyToMessageId?: string | null, payload?: unknown, resolvedThreadRootId?: number | null): Promise>; + create( + podId: string, userId: string, content: string, type: string, replyToMessageId?: string | null, + payload?: unknown, resolvedThreadRootId?: number | null, answeringMessageId?: string | null, + ): Promise>; }).create( String(podId), String(agentUser._id), @@ -1715,6 +1726,7 @@ class AgentMessageService { replyToMessageId, payload, resolvedThreadRootId, + answeringMessageId, ); // The raw INSERT row carries no reply JOIN, so a replying agent's diff --git a/backend/services/attentionItemService.ts b/backend/services/attentionItemService.ts index 8f1c9f4e21..9401743a20 100644 --- a/backend/services/attentionItemService.ts +++ b/backend/services/attentionItemService.ts @@ -106,6 +106,8 @@ const peopleThisAgentIsAnswering = async ( agentName: String(author.botMetadata?.agentName || author.username || ''), instanceId: String(author.botMetadata?.instanceId || 'default'), replyToMessageId: message?.reply_to_message_id ?? message?.replyToMessageId, + // Carried from the post, never stored on the row (agentAnswerService). + answeringMessageId: message?.answeringMessageId, }); } catch (error) { console.warn('[attention] answer check failed; recording the mention:', (error as Error).message); diff --git a/backend/services/starterTaskService.ts b/backend/services/starterTaskService.ts index 8a4b6bfaa4..305965d714 100644 --- a/backend/services/starterTaskService.ts +++ b/backend/services/starterTaskService.ts @@ -86,12 +86,13 @@ export const completeConnectAgentStarterTask = async ({ * check comes first, so a pod with no open card costs one indexed read. */ export const completeFirstTaskOnAnswer = async ({ - podId, agentName, instanceId, replyToMessageId, agentLabel, content = '', + podId, agentName, instanceId, replyToMessageId, answeringMessageId, agentLabel, content = '', }: { podId: string; agentName: string; instanceId?: string; replyToMessageId?: unknown; + answeringMessageId?: unknown; agentLabel: string; content?: string; }): Promise => { @@ -108,7 +109,7 @@ export const completeFirstTaskOnAnswer = async ({ const owner = pod?.createdBy ? String(pod.createdBy) : ''; if (!owner) return; const answered = await humansAnsweredBy({ - podId, agentName, instanceId, replyToMessageId, + podId, agentName, instanceId, replyToMessageId, answeringMessageId, }); if (!answered.has(owner)) return; await completeStarterTask(