fix: recover agent turns from unread pod history backlogs

backend/__tests__/service/agentSendFreshnessCommitFence.test.js

@@ -57,4 +57,32 @@ d('agent freshness: in-flight PostgreSQL insert', () => {
       writer.release();
     }
   });
+  it.each([false, true])('catches up a large PostgreSQL backlog without losing the commit fence (cursor: %s)', async (existing) => {
+    await pool.query('DELETE FROM messages WHERE pod_id = $1', [String(podId)]);
+    await Cursor.deleteOne({ _id: `${agentId}:${podId}` });
+    if (existing) await Cursor.create({ _id: `${agentId}:${podId}`, agentUserId: agentId,
+      podId, readAt: new Date(0), readRevision: 0 });
+    await pool.query(`INSERT INTO messages (pod_id, user_id, content)
+      SELECT $1, $2, 'history ' || n FROM generate_series(1, 2400) AS n`, [String(podId), String(peerId)]);
+    const start = await Freshness.readStart(podId);
+    const tail = await pool.query('SELECT id FROM messages WHERE pod_id = $1 ORDER BY id DESC LIMIT 20', [String(podId)]);
+    const ids = tail.rows.map((row) => String(row.id));
+    await Freshness.recordRead(agentId, podId, start, ids);
+    expect((await Freshness.unseenMessages(agentId, podId)).unseenCount).toBe(2400);
+    const writer = await pool.connect();
+    try {
+      await writer.query('BEGIN');
+      await writer.query('INSERT INTO messages (pod_id, user_id, content) VALUES ($1, $2, $3)',
+        [String(podId), String(peerId), 'later committed peer']);
+      expect(await Freshness.recordCatchUp(agentId, podId, start, ids)).toEqual({ skippedCount: 2380 });
+      await writer.query('COMMIT');
+      const hold = await Freshness.unseenMessages(agentId, podId);
+      expect(hold).toMatchObject({ unseenCount: 1 });
+      expect(hold.messages[0].content).toBe('later committed peer');
+    } finally {
+      await writer.query('ROLLBACK').catch(() => undefined);
+      writer.release();
+    }
+  });
+
 });

backend/__tests__/service/agentSendFreshnessMongoCommitFence.test.js

@@ -1,6 +1,7 @@
 /* eslint-disable import/extensions, import/no-unresolved -- backend TypeScript modules expose CJS exports. */
 const mongoose = require('mongoose');
 const Message = require('../../models/Message');
+require('../../models/User');
 const Cursor = require('../../models/AgentPodReadCursor');
 const Freshness = require('../../services/agentSendFreshness');
 const { setupMongoDb, closeMongoDb } = require('../utils/testUtils');
@@ -119,4 +120,32 @@ describe('agent freshness: delayed Mongo-only write', () => {
     expect(hold.unseenCount).toBe(22);
     expect(hold.messages[0].content).toBe('delayed older timestamp');
   });
+  it.each([false, true])('explicitly catches up thousands of rows (existing cursor: %s), preserving later commits', async (existing) => {
+    const old = new Date(Date.now() - 60000);
+    if (existing) await Cursor.create({
+      _id: `${agentId}:${podId}`, agentUserId: agentId, podId, readAt: new Date(0),
+    });
+    await Message.collection.insertMany(Array.from({ length: 2400 }, (_, i) => ({
+      podId, userId: peerId, content: `history ${i}`, messageType: 'text',
+      createdAt: new Date(old.getTime() + i),
+    })));
+    const start = await Freshness.readStart(podId);
+    const tail = await Message.find({ podId }).sort({ createdAt: -1 }).limit(20).lean();
+    const ids = tail.map((row) => String(row._id));
+    await Freshness.recordRead(agentId, podId, start, ids);
+    expect((await Freshness.unseenMessages(agentId, podId)).unseenCount).toBe(2400);
+    await Message.collection.insertOne({
+      podId, userId: peerId, content: 'late commit with old timestamp', messageType: 'text',
+      createdAt: old, freshnessPending: false,
+      freshnessCommittedAt: new Date(start.at.getTime() + 1),
+    });
+    expect(await Freshness.recordCatchUp(agentId, podId, start, ids)).toEqual({ skippedCount: 2380 });
+    const hold = await Freshness.unseenMessages(agentId, podId);
+    expect(hold).toMatchObject({ unseenCount: 1 });
+    expect(hold.messages[0].content).toBe('late commit with old timestamp');
+    // Replaying an older receipt never rolls the cursor back.
+    await Freshness.recordCatchUp(agentId, podId, { at: old }, []);
+    expect((await Cursor.findById(`${agentId}:${podId}`)).readAt).toEqual(start.at);
+  });
+
 });

backend/__tests__/unit/routes/agentsRuntime.post-message.config.test.js

@@ -20,6 +20,7 @@ jest.mock('../../../services/agentMessageService', () => ({
 jest.mock('../../../services/agentSendFreshness', () => ({
   readStart: jest.fn(async () => ({ at: new Date('2026-10-24T00:00:00Z') })),
   recordRead: jest.fn(),
+  recordCatchUp: jest.fn(),
   unseenMessages: jest.fn().mockResolvedValue(null),
 }));
 jest.mock('../../../services/mentionNonMemberNoticeService', () => ({
@@ -38,20 +39,22 @@ jest.mock('../../../routes/uploads', () => ({
 }));
 jest.mock('../../../integrations', () => ({ get: jest.fn() }));
 jest.mock('../../../models/Activity', () => ({}));
-jest.mock('../../../models/User', () => ({ findById: jest.fn() }));
+jest.mock('../../../models/User', () => ({ findById: jest.fn(), findOne: jest.fn() }));
+jest.mock('../../../models/AgentCredential', () => ({ findOne: jest.fn() }));
 jest.mock('../../../models/Post', () => ({ findById: jest.fn() }));
 jest.mock('../../../models/File', () => ({ findOne: jest.fn(), deleteOne: jest.fn() }));
 jest.mock('../../../services/objectStore', () => ({ getObjectStore: jest.fn() }));
 jest.mock('../../../models/Pod', () => ({
   find: jest.fn(),
   findById: jest.fn(() => ({ select: () => ({ lean: async () => ({ type: 'chat' }) }) })),
 }));
-jest.mock('../../../services/dmService', () => ({ getOrCreateAgentDM: jest.fn() }));
+jest.mock('../../../services/dmService', () => ({ getOrCreateAgentDM: jest.fn(), canViewPod: jest.fn() }));
 jest.mock('../../../models/Integration', () => ({ find: jest.fn(), findOne: jest.fn() }));
 jest.mock('../../../models/AgentRegistry', () => ({
-  AgentInstallation: { findOne: jest.fn(), find: jest.fn() },
+  AgentInstallation: { findOne: jest.fn(), find: jest.fn(), updateOne: jest.fn() },
 }));
 
+const DMService = require('../../../services/dmService');
 const AgentMessageService = require('../../../services/agentMessageService');
 const AgentSendFreshness = require('../../../services/agentSendFreshness');
 const PodContextService = require('../../../services/podContextService');
@@ -306,3 +309,118 @@ describe('attachment freshness', () => {
     expect(getObjectStore).not.toHaveBeenCalled();
   });
 });
+
+
+describe('explicit catch-up', () => {
+  const handler = getRouteHandler('/pods/:podId/catch-up', 'post');
+  const installation = { podId: 'pod-1', agentName: 'openclaw' };
+  const req = { params: { podId: 'pod-1' }, body: { confirmSkipHistory: true },
+    agentUser: { _id: 'bot-1' }, agentInstallation: installation };
+  const response = () => ({ status: jest.fn().mockReturnThis(), set: jest.fn(), json: jest.fn() });
+  beforeEach(() => {
+    jest.clearAllMocks();
+    DMService.canViewPod.mockResolvedValue(true);
+    AgentMessageService.getRecentMessages.mockResolvedValue([{ id: '1', content: 'tail' }]);
+    AgentSendFreshness.recordCatchUp.mockResolvedValue({ skippedCount: 14000 });
+  });
+  it('requires explicit confirmation and pod authorization before any reads or cursor writes', async () => {
+    const noConfirm = response();
+    await handler({ ...req, body: {} }, noConfirm);
+    expect(noConfirm.status).toHaveBeenCalledWith(400);
+    const wrongPod = response();
+    await handler({ ...req, params: { podId: 'other' } }, wrongPod);
+    expect(wrongPod.status).toHaveBeenCalledWith(403);
+    expect(AgentSendFreshness.readStart).not.toHaveBeenCalled();
+    expect(AgentSendFreshness.recordCatchUp).not.toHaveBeenCalled();
+  });
+  it('captures the fence before fetching a bounded tail and reports skipped history', async () => {
+    const res = response();
+    const original = process.env.PG_HOST;
+    delete process.env.PG_HOST;
+    try { await handler(req, res); } finally {
+      if (original === undefined) delete process.env.PG_HOST;
+      else process.env.PG_HOST = original;
+    }
+    expect(AgentSendFreshness.readStart.mock.invocationCallOrder[0])
+      .toBeLessThan(AgentMessageService.getRecentMessages.mock.invocationCallOrder[0]);
+    expect(AgentMessageService.getRecentMessages).toHaveBeenCalledWith('pod-1', 20, 'bot-1');
+    expect(AgentSendFreshness.recordCatchUp).toHaveBeenCalledWith('bot-1', 'pod-1', expect.any(Object), ['1']);
+    expect(res.json).toHaveBeenCalledWith({ messages: [{ id: '1', content: 'tail' }],
+      catchUp: { skippedCount: 14000, through: expect.any(Object) } });
+  });
+  it('does not acknowledge a Mongo fallback while PostgreSQL is configured', async () => {
+    const original = process.env.PG_HOST;
+    process.env.PG_HOST = 'configured';
+    const res = response();
+    try { await handler(req, res); } finally {
+      if (original === undefined) delete process.env.PG_HOST;
+      else process.env.PG_HOST = original;
+    }
+    expect(res.status).toHaveBeenCalledWith(503);
+    expect(AgentSendFreshness.recordCatchUp).not.toHaveBeenCalled();
+  });
+  it('returns a retryable failure instead of a receipt when the cursor write fails', async () => {
+    const original = process.env.PG_HOST;
+    delete process.env.PG_HOST;
+    AgentSendFreshness.recordCatchUp.mockRejectedValueOnce(new Error('cursor unavailable'));
+    const res = response();
+    try { await handler(req, res); } finally {
+      if (original === undefined) delete process.env.PG_HOST;
+      else process.env.PG_HOST = original;
+    }
+    expect(res.status).toHaveBeenCalledWith(503);
+    expect(res.json).not.toHaveBeenCalledWith(expect.objectContaining({ catchUp: expect.anything() }));
+  });
+
+  it('refuses a scoped token whose canonical agent cannot view the pod, before reading or advancing', async () => {
+    DMService.canViewPod.mockResolvedValueOnce(false);
+    const res = response();
+    await handler(req, res);
+    expect(DMService.canViewPod).toHaveBeenCalledWith('bot-1', { type: 'chat' });
+    expect(res.status).toHaveBeenCalledWith(403);
+    expect(AgentSendFreshness.readStart).not.toHaveBeenCalled();
+    expect(AgentMessageService.getRecentMessages).not.toHaveBeenCalled();
+    expect(AgentSendFreshness.recordCatchUp).not.toHaveBeenCalled();
+  });
+
+});
+
+
+describe('send freshness through real legacy-token HTTP authentication', () => {
+  const express = require('express');
+  const request = require('supertest');
+  const realAuth = jest.requireActual('../../../middleware/agentRuntimeAuth');
+  const User = require('../../../models/User');
+  const Credential = require('../../../models/AgentCredential');
+  const { AgentInstallation } = require('../../../models/AgentRegistry');
+  const Identity = require('../../../services/agentIdentityService');
+  const { hash } = require('../../../utils/secret');
+  const token = 'cm_agent_legacy_fixture';
+  const app = express();
+  app.get('/api/agents/runtime/pods/:podId/send-freshness', realAuth,
+    getRouteHandler('/pods/:podId/send-freshness', 'get'));
+  beforeEach(() => {
+    jest.clearAllMocks();
+    Credential.findOne.mockResolvedValue(null);
+    User.findOne.mockResolvedValue(null);
+    const installation = { _id: 'legacy-install', podId: 'pod-1', agentName: 'openclaw', instanceId: 'legacy',
+      runtimeTokens: [{ tokenHash: hash(token), lastUsedAt: new Date() }] };
+    AgentInstallation.findOne.mockResolvedValue(installation);
+    AgentInstallation.find.mockResolvedValue([installation]);
+    AgentSendFreshness.unseenMessages.mockResolvedValue(null);
+  });
+  it('resolves a canonical identity when legacy auth has no bot User and returns freshness', async () => {
+    await request(app).get('/api/agents/runtime/pods/pod-1/send-freshness')
+      .set('Authorization', `Bearer ${token}`).expect(200, { ok: true });
+    expect(Identity.getOrCreateAgentUser).toHaveBeenCalledWith('openclaw', {
+      instanceId: 'legacy', displayName: undefined,
+    });
+    expect(AgentSendFreshness.unseenMessages).toHaveBeenCalledWith('bot-1', 'pod-1');
+  });
+  it('still refuses another pod before canonical identity resolution or freshness lookup', async () => {
+    await request(app).get('/api/agents/runtime/pods/other/send-freshness')
+      .set('Authorization', `Bearer ${token}`).expect(403);
+    expect(Identity.getOrCreateAgentUser).not.toHaveBeenCalled();
+    expect(AgentSendFreshness.unseenMessages).not.toHaveBeenCalled();
+  });
+});

backend/routes/agentsRuntime.ts

@@ -1806,6 +1806,41 @@ router.get('/pods/:podId/context', phase4RateLimit, agentRuntimeAuth, async (req
   }
 });
 
+// An explicit history skip, never an implicit side effect of a bounded read.
+router.post('/pods/:podId/catch-up', phase4RateLimit, agentRuntimeAuth, async (req: any, res: any) => {
+  const { podId } = req.params;
+  const installation = resolveInstallationForPod(req.agentInstallations, req.agentInstallation, podId);
+  if (!ensurePodMatch(req.agentInstallations || installation, podId, req.agentAuthorizedPodIds)) {
+    return res.status(403).json({ message: 'Agent token not authorized for this pod' });
+  }
+  if (req.body?.confirmSkipHistory !== true) {
+    return res.status(400).json({ message: 'Catch-up requires confirmSkipHistory:true' });
+  }
+  try {
+    const agentUser = req.agentUser || await AgentIdentityService.getOrCreateAgentUser(installation.agentName, {
+      instanceId: installation.instanceId || 'default', displayName: installation.displayName,
+    });
+    const pod = await Pod.findById(podId).select('type members').lean();
+    if (!await DMService.canViewPod(agentUser._id, pod)) {
+      return res.status(403).json({ message: 'Agent cannot view this pod' });
+    }
+    const start = await AgentSendFreshness.readStart(podId);
+    const messages = await AgentMessageService.getRecentMessages(podId, 20, agentUser._id);
+    if (process.env.PG_HOST && (messages as typeof messages & { readSource?: string }).readSource !== 'postgres') {
+      return res.status(503).json({ message: 'PostgreSQL tail unavailable; catch-up was not applied' });
+    }
+    const catchUp = await AgentSendFreshness.recordCatchUp(
+      agentUser._id, podId, start,
+      messages.map((message: Record<string, unknown>) => String(message.id || message._id || '')),
+    );
+    res.set('Cache-Control', 'no-store');
+    return res.json({ messages, catchUp: { ...catchUp, through: start } });
+  } catch (error) {
+    console.error('Error catching up agent pod history:', error);
+    return res.status(503).json({ message: 'Catch-up unavailable; retry before generating a reply' });
+  }
+});
+
 router.get('/pods/:podId/messages', phase4RateLimit, agentRuntimeAuth, async (req: any, res: any) => {
   try {
     const { podId } = req.params;
@@ -4149,8 +4184,11 @@ router.get('/pods/:podId/send-freshness', phase4RateLimit, agentRuntimeAuth, asy
     if (!ensurePodMatch(req.agentInstallations || req.agentInstallation, podId, req.agentAuthorizedPodIds)) {
       return res.status(403).json({ message: 'Agent token not authorized for this pod' });
     }
-    if (!req.agentUser?._id) return res.status(401).json({ message: 'Agent identity required' });
-    const hold = await AgentSendFreshness.unseenMessages(req.agentUser._id, podId);
+    const installation = resolveInstallationForPod(req.agentInstallations, req.agentInstallation, podId);
+    const agentUser = req.agentUser || await AgentIdentityService.getOrCreateAgentUser(installation.agentName, {
+      instanceId: installation.instanceId || 'default', displayName: installation.displayName,
+    });
+    const hold = await AgentSendFreshness.unseenMessages(agentUser._id, podId);
     return hold ? res.status(409).json(hold) : res.json({ ok: true });
   } catch (error: any) {
     console.error('Agent freshness check error:', error);

backend/services/agentSendFreshness.ts

@@ -38,9 +38,13 @@ async function recordRead(
     .select('readAt readRevision').lean();
   // The first read also needs evidence: a truncated tail cannot acknowledge
   // older history just because no cursor existed yet.
-  if (await hasUnshownMessages(
+  if (await omittedMessages(
     agentUserId, podId, previous || { readAt: new Date(0), readRevision: -1 }, start, observedIds,
   )) return;
+  await advanceCursor(agentUserId, podId, start);
+}
+
+async function advanceCursor(agentUserId: unknown, podId: unknown, start: ReadStart): Promise<void> {
   // A slow earlier read must not roll a newer read's watermark backwards.
   const filter = { _id: cursorId(agentUserId, podId) };
   const maximum = {
@@ -57,25 +61,26 @@ async function recordRead(
   }
 }
 
-async function hasUnshownMessages(
+async function omittedMessages(
   agentUserId: unknown,
   podId: unknown,
   previous: { readAt?: Date; readRevision?: number },
   start: ReadStart,
   observedIds: string[],
-): Promise<boolean> {
+  count = false,
+): Promise<number> {
   const ownId = String(agentUserId);
   if (pool && process.env.PG_HOST) {
     if (start.revision === undefined) throw new Error('PostgreSQL read revision missing');
     const ids = observedIds.filter((id) => /^\d+$/.test(id)).map(Number);
     const result = await pool.query(`
-      SELECT 1 FROM messages m
+      SELECT ${count ? 'COUNT(*)::integer AS skipped_count' : '1'} FROM messages m
       WHERE m.pod_id = $1 AND m.freshness_revision > $2 AND m.freshness_revision <= $3
         AND (m.user_id IS DISTINCT FROM $4 OR m.author_kind = 'kernel')
         AND NOT (m.id = ANY($5::int[]))
-      LIMIT 1
+      ${count ? '' : 'LIMIT 1'}
     `, [String(podId), previous.readRevision ?? -1, start.revision, ownId, ids]);
-    return result.rows.length > 0;
+    return count ? Number(result.rows[0]?.skipped_count || 0) : result.rows.length;
   }
   const observed = observedIds.filter((id) => /^[a-f\d]{24}$/i.test(id));
   const query = {
@@ -93,7 +98,21 @@ async function hasUnshownMessages(
       ] },
     ],
   };
-  return Boolean(await MongoMessage.exists(query));
+  return count ? MongoMessage.countDocuments(query) : Number(Boolean(await MongoMessage.exists(query)));
+}
+
+// Only the explicit catch-up endpoint may acknowledge omitted history. Its
+// snapshot precedes the tail fetch, so later commits remain send-time holds.
+async function recordCatchUp(
+  agentUserId: unknown, podId: unknown, start: ReadStart, observedIds: string[],
+): Promise<{ skippedCount: number }> {
+  const previous = await Cursor.findOne({ _id: cursorId(agentUserId, podId) })
+    .select('readAt readRevision').lean();
+  const skippedCount = await omittedMessages(
+    agentUserId, podId, previous || { readAt: new Date(0), readRevision: -1 }, start, observedIds, true,
+  );
+  await advanceCursor(agentUserId, podId, start);
+  return { skippedCount };
 }
 
 async function unseenMessages(agentUserId: unknown, podId: unknown): Promise<Hold> {
@@ -167,5 +186,5 @@ async function unseenFromMongo(podId: unknown, ownId: string, after: Date): Prom
   };
 }
 
-module.exports = { readStart, recordRead, unseenMessages };
+module.exports = { readStart, recordRead, recordCatchUp, unseenMessages };
 export {};