fix(webhook): inline reply posts through the CAP post path (TASK-015)

backend/__tests__/unit/services/agentEventService.webhookInlinePost.test.js

@@ -0,0 +1,169 @@
+/**
+ * TASK-015 — an inline webhook reply must actually post, and a failed post
+ * must not be recorded as posted.
+ *
+ * The old path called agentMessageService.postAgentMessage, which does not
+ * exist, so the TypeError was swallowed and the event still settled as
+ * `posted`. These tests drive the real deliverEventViaWebhook (through the
+ * static seam), with only the network and persistence edges mocked.
+ */
+
+jest.mock('../../../models/AgentEvent', () => ({
+  findByIdAndUpdate: jest.fn().mockResolvedValue({}),
+}));
+
+jest.mock('../../../services/outboundUrlGuard', () => ({
+  postToPublicAddress: jest.fn(),
+}));
+
+jest.mock('../../../services/agentMessageService', () => ({
+  postMessage: jest.fn(),
+}));
+
+jest.mock('../../../services/agentMentionService', () => ({
+  enqueueMentions: jest.fn().mockResolvedValue(undefined),
+  enqueueDmEvent: jest.fn().mockResolvedValue(undefined),
+}));
+
+jest.mock('../../../services/agentSendFreshness', () => ({
+  unseenMessages: jest.fn().mockResolvedValue(null),
+}));
+
+jest.mock('../../../models/Pod', () => ({
+  findById: jest.fn(),
+}));
+
+jest.mock('../../../services/agentIdentityService', () => ({
+  getOrCreateAgentUser: jest.fn(),
+}));
+
+jest.mock('../../../models/AgentRegistry', () => ({
+  AgentInstallation: { find: jest.fn(), findOne: jest.fn() },
+}));
+
+jest.mock('../../../models/Integration', () => ({}));
+jest.mock('../../../models/Gateway', () => ({ findById: jest.fn() }));
+jest.mock('../../../services/agentDigestFoldService', () => ({
+  foldExpiredDigests: jest.fn().mockResolvedValue(0),
+  readFoldedDigest: jest.fn().mockResolvedValue(null),
+  settleFoldedDigest: jest.fn().mockResolvedValue(undefined),
+}));
+jest.mock('../../../services/agentProvisionerService', () => ({
+  clearAgentRuntimeSessions: jest.fn(),
+  restartAgentRuntime: jest.fn(),
+  resolveOpenClawAccountId: jest.fn(),
+}));
+
+const AgentEvent = require('../../../models/AgentEvent');
+const { postToPublicAddress } = require('../../../services/outboundUrlGuard');
+const AgentMessageService = require('../../../services/agentMessageService');
+const AgentMentionService = require('../../../services/agentMentionService');
+const AgentSendFreshness = require('../../../services/agentSendFreshness');
+const PodModel = require('../../../models/Pod');
+const AgentIdentityService = require('../../../services/agentIdentityService');
+const AgentEventService = require('../../../services/agentEventService');
+
+const installation = {
+  displayName: 'Echo',
+  config: { runtime: { webhookUrl: 'https://agent.example.com/hook' } },
+};
+
+const event = {
+  _id: 'evt_inline_001',
+  type: 'chat.mention',
+  podId: 'pod_inline',
+  agentName: 'echo-agent',
+  instanceId: 'default',
+  createdAt: new Date().toISOString(),
+  payload: {},
+};
+
+const agentUser = { _id: 'user_echo', username: 'echo-agent' };
+
+const respondWith = (body) => postToPublicAddress.mockResolvedValue({
+  ok: true,
+  status: 200,
+  text: JSON.stringify(body),
+});
+
+const recordedDelivery = () => {
+  const call = AgentEvent.findByIdAndUpdate.mock.calls.at(-1);
+  return call[1].delivery;
+};
+
+beforeEach(() => {
+  jest.clearAllMocks();
+  AgentEventService.attachLegacyDigest = jest.fn().mockResolvedValue(undefined);
+  AgentEventService.settleLegacyDigest = jest.fn().mockResolvedValue(undefined);
+  AgentIdentityService.getOrCreateAgentUser.mockResolvedValue(agentUser);
+  AgentSendFreshness.unseenMessages.mockResolvedValue(null);
+  PodModel.findById.mockReturnValue({
+    select: () => ({ lean: () => Promise.resolve({ type: 'team' }) }),
+  });
+  AgentMessageService.postMessage.mockResolvedValue({
+    success: true,
+    message: { _id: 'msg_1', content: 'hi' },
+  });
+});
+
+describe('webhook inline reply (TASK-015)', () => {
+  test('posted reply goes through postMessage and is recorded as posted', async () => {
+    respondWith({ outcome: 'posted', content: 'hi there' });
+
+    await AgentEventService.deliverEventViaWebhook(installation, event);
+
+    expect(AgentMessageService.postMessage).toHaveBeenCalledWith(expect.objectContaining({
+      agentName: 'echo-agent',
+      podId: 'pod_inline',
+      content: 'hi there',
+      installationConfig: installation.config,
+    }));
+    expect(recordedDelivery()).toEqual(expect.objectContaining({ outcome: 'posted' }));
+  });
+
+  test('a mention-enqueue runs for the posted reply, as on the CAP route', async () => {
+    respondWith({ outcome: 'posted', content: 'hi there' });
+
+    await AgentEventService.deliverEventViaWebhook(installation, event);
+    await new Promise((resolve) => { setImmediate(resolve); });
+
+    expect(AgentMentionService.enqueueMentions).toHaveBeenCalledWith(expect.objectContaining({
+      podId: 'pod_inline',
+      userId: 'user_echo',
+    }));
+  });
+
+  test('a post that throws is recorded as error, never as posted', async () => {
+    AgentMessageService.postMessage.mockRejectedValue(new Error('db down'));
+    respondWith({ outcome: 'posted', content: 'hi there' });
+
+    await AgentEventService.deliverEventViaWebhook(installation, event);
+
+    expect(recordedDelivery()).toEqual(expect.objectContaining({
+      outcome: 'error',
+      reason: 'inline_post_failed',
+    }));
+  });
+
+  test('a reply held by the freshness gate is recorded as skipped, not posted', async () => {
+    AgentSendFreshness.unseenMessages.mockResolvedValue({ code: 'UNSEEN_MESSAGES' });
+    respondWith({ outcome: 'posted', content: 'hi there' });
+
+    await AgentEventService.deliverEventViaWebhook(installation, event);
+
+    expect(AgentMessageService.postMessage).not.toHaveBeenCalled();
+    expect(recordedDelivery()).toEqual(expect.objectContaining({
+      outcome: 'skipped',
+      reason: 'unseen_messages',
+    }));
+  });
+
+  test('a non-posted outcome never calls postMessage', async () => {
+    respondWith({ outcome: 'no_action', reason: 'nothing_to_add' });
+
+    await AgentEventService.deliverEventViaWebhook(installation, event);
+
+    expect(AgentMessageService.postMessage).not.toHaveBeenCalled();
+    expect(recordedDelivery()).toEqual(expect.objectContaining({ outcome: 'no_action' }));
+  });
+});

backend/__tests__/unit/services/agentSendFreshness.test.js

@@ -97,6 +97,18 @@ describe('agent send-time freshness', () => {
     expect(Message.countDocuments).not.toHaveBeenCalled();
   });
 
+  it('bounds the PostgreSQL unseen query by the trigger id, and only a numeric id', async () => {
+    process.env.PG_HOST = 'configured';
+    pool.query.mockResolvedValue({ rows: [] });
+    await Freshness.unseenMessages('agent-id', 'pod-id', { seenThroughMessageId: '22' });
+    expect(pool.query.mock.calls[0][1]).toEqual(['pod-id', -1, 'agent-id', 22]);
+    expect(pool.query.mock.calls[0][0]).toContain('($4::integer IS NULL OR m.id > $4)');
+    await Freshness.unseenMessages('agent-id', 'pod-id', { seenThroughMessageId: 'abc' });
+    expect(pool.query.mock.calls[1][1][3]).toBeNull();
+    await Freshness.unseenMessages('agent-id', 'pod-id');
+    expect(pool.query.mock.calls[2][1][3]).toBeNull();
+  });
+
   it('captures the PostgreSQL revision before the read and advances it with the cursor', async () => {
     process.env.PG_HOST = 'configured';
     pool.query.mockResolvedValueOnce({ rows: [{ revision: '7' }] }).mockResolvedValueOnce({ rows: [] });
@@ -139,3 +151,41 @@ describe('agent send-time freshness', () => {
 });
 
 });
+
+describe('a delivered trigger is seen through its own message (TASK-015 inline reply)', () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+    delete process.env.PG_HOST;
+    Cursor.getReadWatermark.mockResolvedValue({ readAt: new Date('2026-10-24T00:00:00Z') });
+    Message.find.mockReturnValue({ sort: () => ({ limit: () => ({ populate: () => ({ lean: async () => [] }) }) }) });
+  });
+
+  it('does not hold on the trigger itself: the query excludes it by id', async () => {
+    Message.countDocuments.mockResolvedValue(0);
+    const triggerId = '6adfeccc58eabb3645d76520';
+
+    const hold = await Freshness.unseenMessages('agent-id', 'pod-id', { seenThroughMessageId: triggerId });
+    expect(hold).toBeNull();
+    expect(Message.countDocuments.mock.calls[0][0]._id).toEqual({ $gt: triggerId });
+  });
+
+  it('still holds on a message that landed after the trigger', async () => {
+    Message.countDocuments.mockResolvedValue(1);
+    const hold = await Freshness.unseenMessages('agent-id', 'pod-id', {
+      seenThroughMessageId: '6adfeccc58eabb3645d76520',
+    });
+    expect(hold.unseenCount).toBe(1);
+  });
+
+  it('without a trigger the query has no id bound, so the old behaviour is unchanged', async () => {
+    Message.countDocuments.mockResolvedValue(0);
+    await Freshness.unseenMessages('agent-id', 'pod-id');
+    expect(Message.countDocuments.mock.calls[0][0]).not.toHaveProperty('_id');
+  });
+
+  it('a non-ObjectId trigger is ignored rather than becoming an unbounded filter', async () => {
+    Message.countDocuments.mockResolvedValue(0);
+    await Freshness.unseenMessages('agent-id', 'pod-id', { seenThroughMessageId: { $ne: 1 } });
+    expect(Message.countDocuments.mock.calls[0][0]).not.toHaveProperty('_id');
+  });
+});

backend/services/agentEventService.ts

@@ -194,6 +194,7 @@ interface InstallationDoc {
   _id?: unknown;
   agentName?: string;
   instanceId?: string;
+  displayName?: string;
   podId?: unknown;
   status?: string;
   scopes?: string[];
@@ -315,6 +316,59 @@ const normalizeConfig = (config: unknown): Record<string, unknown> => {
   return config as Record<string, unknown>;
 };
 
+// Inline webhook replies take the same service path as the CAP post route
+// (agentsRuntime.ts): the unseen-messages freshness hold, postMessage with the
+// installation config, then the mention or agent-dm enqueue. Resolves to a
+// recorded outcome only when the reply was held or skipped; throws when the post fails.
+const postInlineReply = async (
+  installation: InstallationDoc, event: EventDoc, content: string,
+): Promise<{ outcome: string; reason: string } | null> => {
+  // eslint-disable-next-line global-require
+  const AgentMessageService = require('./agentMessageService');
+  // eslint-disable-next-line global-require
+  const AgentMentionService = require('./agentMentionService');
+  // eslint-disable-next-line global-require
+  const AgentSendFreshness = require('./agentSendFreshness');
+  // eslint-disable-next-line global-require
+  const PodModel = require('../models/Pod');
+
+  const agentUser = await AgentIdentityService.getOrCreateAgentUser(event.agentName, {
+    instanceId: event.instanceId || 'default',
+    displayName: installation.displayName,
+  });
+  // The event's own message is the one being answered, so only messages that
+  // landed after it hold the reply (see unseenMessages).
+  const hold = await AgentSendFreshness.unseenMessages(agentUser._id, event.podId, {
+    seenThroughMessageId: event.payload?.messageId,
+  });
+  if (hold) return { outcome: 'skipped', reason: 'unseen_messages' };
+
+  const result = await AgentMessageService.postMessage({
+    agentName: event.agentName,
+    instanceId: event.instanceId || 'default',
+    displayName: installation.displayName,
+    podId: event.podId,
+    content,
+    replyToMessageId: null,
+    threadRootId: null,
+    installationConfig: installation.config || null,
+  });
+  if (result?.skipped) return { outcome: 'skipped', reason: String(result.reason || 'skipped') };
+  if (!result?.success || !result.message) throw new Error('post returned no message');
+
+  PodModel.findById(event.podId).select('type').lean()
+    .then((podDoc: { type?: string } | null) => {
+      const args = {
+        podId: event.podId, message: result.message, userId: agentUser._id, username: agentUser.username,
+      };
+      return podDoc?.type === 'agent-dm'
+        ? AgentMentionService.enqueueDmEvent(args)
+        : AgentMentionService.enqueueMentions(args);
+    })
+    .catch((err: Error) => console.warn('[webhook] inline reply enqueue failed:', err.message));
+  return null;
+};
+
 const deliverEventViaWebhook = async (installation: InstallationDoc, event: EventDoc): Promise<void> => {
   const runtimeConfig = (normalizeConfig(installation.config)?.runtime || {}) as Record<string, unknown>;
   const { webhookUrl, webhookSecret } = runtimeConfig as { webhookUrl?: string; webhookSecret?: string };
@@ -366,18 +420,21 @@ const deliverEventViaWebhook = async (installation: InstallationDoc, event: Even
       outcome?: string; content?: string; reason?: string;
     };
 
+    // The endpoint's claim is only what it asked for. If the inline reply does
+    // not land, record that instead of 'posted'.
+    let recordedOutcome = outcome;
+    let recordedReason = reason;
     if (outcome === 'posted' && content) {
       try {
-        // eslint-disable-next-line global-require
-        const agentMessageService = require('./agentMessageService');
-        await agentMessageService.postAgentMessage({
-          agentName: event.agentName,
-          instanceId: event.instanceId,
-          podId: event.podId,
-          content: String(content),
-        });
+        const held = await postInlineReply(installation, event, String(content));
+        if (held) {
+          recordedOutcome = held.outcome;
+          recordedReason = held.reason;
+        }
       } catch (postErr) {
         console.warn(`[webhook] Failed to post message for ${event.agentName}:`, (postErr as Error).message);
+        recordedOutcome = 'error';
+        recordedReason = 'inline_post_failed';
       }
     }
 
@@ -394,9 +451,9 @@ const deliverEventViaWebhook = async (installation: InstallationDoc, event: Even
     await AgentEvent.findByIdAndUpdate(event._id, {
       status: 'acked',
       deliveredAt: new Date(),
-      delivery: AgentEventService.normalizeDeliveryMeta({ outcome, reason }),
+      delivery: AgentEventService.normalizeDeliveryMeta({ outcome: recordedOutcome, reason: recordedReason }),
     });
-    if (digestConsumedBy(outcome, reason)) {
+    if (digestConsumedBy(recordedOutcome, recordedReason)) {
       await AgentEventService.settleLegacyDigest(event).catch((err: Error) => {
         console.warn('[webhook] digest settle failed:', err.message);
       });
@@ -809,6 +866,11 @@ class AgentEventService {
     };
   }
 
+  // Test seam: unit tests drive the real webhook delivery, not a copy of it.
+  static deliverEventViaWebhook(installation: InstallationDoc, event: EventDoc): Promise<void> {
+    return deliverEventViaWebhook(installation, event);
+  }
+
   static normalizeDeliveryMeta(input: unknown = {}): DeliveryMeta | null {
     if (!input || typeof input !== 'object') return null;
     const allowedOutcomes = new Set(['acknowledged', 'posted', 'no_action', 'skipped', 'error']);