TASK-441: hold agent writes when pod messages arrive since last read

backend/__tests__/service/agentSendFreshnessCommitFence.test.js

@@ -0,0 +1,60 @@
+/** A timestamp cannot fence an INSERT whose transaction commits after a read. */
+/* eslint-disable import/extensions, import/no-unresolved -- backend TypeScript modules expose CJS exports. */
+const fs = require('fs');
+const path = require('path');
+const mongoose = require('mongoose');
+const { pool } = require('../../config/db-pg');
+const Cursor = require('../../models/AgentPodReadCursor');
+const Freshness = require('../../services/agentSendFreshness');
+
+const d = process.env.INTEGRATION_TEST === 'true' ? describe : describe.skip;
+const podId = new mongoose.Types.ObjectId();
+const agentId = new mongoose.Types.ObjectId();
+const peerId = new mongoose.Types.ObjectId();
+
+d('agent freshness: in-flight PostgreSQL insert', () => {
+  beforeAll(async () => {
+    await mongoose.connect(process.env.MONGO_URI);
+    await pool.query(fs.readFileSync(path.join(__dirname, '../../config/schema.sql'), 'utf8'));
+    await pool.query(
+      "INSERT INTO pods (id, name, type, created_by) VALUES ($1, 'fence', 'chat', $2)",
+      [String(podId), String(peerId)],
+    );
+    await pool.query(
+      "INSERT INTO users (_id, username) VALUES ($1, 'fence-peer')",
+      [String(peerId)],
+    );
+  });
+
+  afterAll(async () => {
+    await Cursor.deleteOne({ _id: `${agentId}:${podId}` });
+    await pool.query('DELETE FROM pods WHERE id = $1', [String(podId)]);
+    await pool.query('DELETE FROM users WHERE _id = $1', [String(peerId)]);
+    await mongoose.disconnect();
+    await pool.end();
+  });
+
+  it('holds a peer message that began before the read but committed afterward', async () => {
+    const writer = await pool.connect();
+    try {
+      await writer.query('BEGIN');
+      const inserted = await writer.query(
+        'INSERT INTO messages (pod_id, user_id, content) VALUES ($1, $2, $3) RETURNING created_at',
+        [String(podId), String(peerId), 'committed after the read'],
+      );
+      await new Promise((resolve) => { setTimeout(resolve, 20); });
+      const start = await Freshness.readStart(podId);
+      expect(inserted.rows[0].created_at.getTime()).toBeLessThanOrEqual(start.at.getTime());
+      expect(await pool.query('SELECT id FROM messages WHERE pod_id = $1', [String(podId)]))
+        .toMatchObject({ rows: [] });
+      await Freshness.recordRead(agentId, podId, start, []);
+      await writer.query('COMMIT');
+      const hold = await Freshness.unseenMessages(agentId, podId);
+      expect(hold).toMatchObject({ code: 'unseen_messages', unseenCount: 1 });
+      expect(hold.messages[0].content).toBe('committed after the read');
+    } finally {
+      await writer.query('ROLLBACK').catch(() => undefined);
+      writer.release();
+    }
+  });
+});

backend/__tests__/service/agentSendFreshnessMongoCommitFence.test.js

@@ -0,0 +1,122 @@
+/* eslint-disable import/extensions, import/no-unresolved -- backend TypeScript modules expose CJS exports. */
+const mongoose = require('mongoose');
+const Message = require('../../models/Message');
+const Cursor = require('../../models/AgentPodReadCursor');
+const Freshness = require('../../services/agentSendFreshness');
+const { setupMongoDb, closeMongoDb } = require('../utils/testUtils');
+
+describe('agent freshness: delayed Mongo-only write', () => {
+  const originalPgHost = process.env.PG_HOST;
+  const podId = new mongoose.Types.ObjectId();
+  const agentId = new mongoose.Types.ObjectId();
+  const peerId = new mongoose.Types.ObjectId();
+
+  beforeAll(async () => {
+    delete process.env.PG_HOST;
+    await setupMongoDb();
+  });
+
+  afterAll(async () => {
+    await closeMongoDb();
+    if (originalPgHost === undefined) delete process.env.PG_HOST;
+    else process.env.PG_HOST = originalPgHost;
+  });
+
+  beforeEach(async () => {
+    await Message.deleteMany({});
+    await Cursor.deleteMany({});
+  });
+
+  it('holds an insert whose createdAt predates a successful read but whose save finishes later', async () => {
+    const peerMessage = new Message({
+      podId,
+      userId: peerId,
+      content: 'late commit',
+      createdAt: new Date(Date.now() - 1000),
+    });
+    const insert = Message.collection.insertOne.bind(Message.collection);
+    let resumeInsert;
+    let notifyInsert;
+    const paused = new Promise((resolve) => { notifyInsert = resolve; });
+    const gate = new Promise((resolve) => { resumeInsert = resolve; });
+    const spy = jest.spyOn(Message.collection, 'insertOne').mockImplementation(async (...args) => {
+      notifyInsert();
+      await gate;
+      return insert(...args);
+    });
+
+    let start;
+    try {
+      const saving = peerMessage.save();
+      await paused;
+      start = await Freshness.readStart(podId);
+      expect(await Message.find({ podId }).lean()).toHaveLength(0);
+      await Freshness.recordRead(agentId, podId, start, []);
+      resumeInsert();
+      await saving;
+    } finally {
+      resumeInsert();
+      spy.mockRestore();
+    }
+
+    const stored = await Message.findById(peerMessage._id).lean();
+    expect(stored.freshnessPending).toBe(false);
+    expect(stored.freshnessCommittedAt.getTime()).toBeGreaterThanOrEqual(start.at.getTime());
+    const hold = await Freshness.unseenMessages(agentId, podId);
+    expect(hold).toMatchObject({ code: 'unseen_messages', unseenCount: 1 });
+    expect(hold.messages[0].content).toBe('late commit');
+  });
+
+  it('holds a persisted write left pending if its commit stamp fails', async () => {
+    const start = await Freshness.readStart(podId);
+    await Freshness.recordRead(agentId, podId, start, []);
+    await Message.collection.insertOne({
+      podId,
+      userId: peerId,
+      content: 'pending commit',
+      messageType: 'text',
+      createdAt: new Date(Date.now() - 1000),
+      freshnessPending: true,
+    });
+    expect(await Freshness.unseenMessages(agentId, podId))
+      .toMatchObject({ code: 'unseen_messages', unseenCount: 1 });
+  });
+
+  it('does not clear a hold when a delayed commit falls outside the returned tail', async () => {
+    const now = Date.now();
+    const previousAt = new Date(now - 60000);
+    await Cursor.create({
+      _id: `${agentId}:${podId}`, agentUserId: agentId, podId, readAt: previousAt,
+    });
+    const newer = Array.from({ length: 21 }, (_, i) => ({
+      podId,
+      userId: peerId,
+      content: `newer ${i}`,
+      messageType: 'text',
+      createdAt: new Date(now - 20000 + i),
+      freshnessPending: false,
+      freshnessCommittedAt: new Date(now - 10000 + i),
+    }));
+    await Message.collection.insertMany([
+      ...newer,
+      {
+        podId,
+        userId: peerId,
+        content: 'delayed older timestamp',
+        messageType: 'text',
+        createdAt: new Date(now - 30000),
+        freshnessPending: false,
+        freshnessCommittedAt: new Date(now - 1000),
+      },
+    ]);
+
+    const start = await Freshness.readStart(podId);
+    const returnedTail = await Message.find({ podId }).sort({ createdAt: -1 }).limit(20).lean();
+    expect(returnedTail.map((message) => message.content)).not.toContain('delayed older timestamp');
+    await Freshness.recordRead(agentId, podId, start, returnedTail.map((message) => String(message._id)));
+    expect((await Cursor.findById(`${agentId}:${podId}`)).readAt).toEqual(previousAt);
+    const hold = await Freshness.unseenMessages(agentId, podId);
+    expect(hold.unseenCount).toBe(22);
+    expect(hold.messages[0].content).toBe('delayed older timestamp');
+  });
+});

backend/__tests__/service/clawdbot-e2e.test.js

@@ -17,6 +17,15 @@ const mongoose = require('mongoose');
 
 const { setupMongoDb, closeMongoDb, clearMongoDb } = require('../utils/testUtils');
 
+// This suite exercises the legacy Mongo-backed Clawdbot flow. Freshness has
+// its own real-database service coverage; Tier 1's PostgreSQL host is present
+// here without a message schema for this fixture.
+jest.mock('../../services/agentSendFreshness', () => ({
+  readStart: jest.fn().mockResolvedValue({ at: new Date(0) }),
+  recordRead: jest.fn().mockResolvedValue(undefined),
+  unseenMessages: jest.fn().mockResolvedValue(null),
+}));
+
 // Models
 const User = require('../../models/User');
 const Pod = require('../../models/Pod');
@@ -652,13 +661,20 @@ describe('Clawdbot E2E Integration Tests', () => {
         .send({ label: 'Message Test Token' });
 
       agentToken = tokenRes.body.token;
+      // Read the current pod tail before writing; send-time freshness holds
+      // a blind first post when other participants already spoke here.
+      const contextRead = await request(app)
+        .get(`/api/agents/runtime/pods/${testPod._id}/messages`)
+        .set('Authorization', `Bearer ${agentToken}`);
+      expect(contextRead.status).toBe(200);
     });
 
     test('should post a message to pod via runtime API', async () => {
       const res = await request(app)
         .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
         .set('Authorization', `Bearer ${agentToken}`)
         .send({
+          continueAnyway: true,
           content: 'Hello from Clawdbot! Based on the recent discussions, here are my insights...',
           messageType: 'text',
           metadata: {
@@ -678,7 +694,7 @@ describe('Clawdbot E2E Integration Tests', () => {
       const bareRes = await request(app)
         .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
         .set('Authorization', `Bearer ${agentToken}`)
-        .send({ content: bareContent, messageType: 'text' });
+        .send({ content: bareContent, messageType: 'text', continueAnyway: true });
 
       expect(bareRes.status).toBe(200);
       expect(bareRes.body.success).toBe(true);
@@ -689,7 +705,7 @@ describe('Clawdbot E2E Integration Tests', () => {
       const escapedRes = await request(app)
         .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
         .set('Authorization', `Bearer ${agentToken}`)
-        .send({ content: escapedContent, messageType: 'text' });
+        .send({ content: escapedContent, messageType: 'text', continueAnyway: true });
 
       expect(escapedRes.status).toBe(200);
       expect(escapedRes.body.success).toBe(true);
@@ -723,6 +739,7 @@ describe('Clawdbot E2E Integration Tests', () => {
         .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
         .set('Authorization', `Bearer ${agentToken}`)
         .send({
+          continueAnyway: true,
           messageType: 'text',
         });
 
@@ -738,6 +755,7 @@ describe('Clawdbot E2E Integration Tests', () => {
         .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
         .set('Authorization', `Bearer ${agentToken}`)
         .send({
+          continueAnyway: true,
           content: 'RGCTX',
           messageType: 'text',
         });
@@ -820,10 +838,14 @@ describe('Clawdbot E2E Integration Tests', () => {
 3. Set up daily standups to track progress`;
 
       // 4. Post response to pod
+      await request(app)
+        .get(`/api/agents/runtime/pods/${testPod._id}/messages`)
+        .set('Authorization', `Bearer ${agentToken}`);
       const postRes = await request(app)
         .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
         .set('Authorization', `Bearer ${agentToken}`)
         .send({
+          continueAnyway: true,
           content: clawdbotResponse,
           messageType: 'text',
           metadata: {
@@ -891,6 +913,7 @@ describe('Clawdbot E2E Integration Tests', () => {
           .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
           .set('Authorization', `Bearer ${agentToken}`)
           .send({
+            continueAnyway: true,
             content: `Response to batch ${i + 1}`,
             messageType: 'text',
           });
@@ -982,22 +1005,30 @@ describe('Clawdbot E2E Integration Tests', () => {
     });
 
     test('should allow both agents to post messages to the same pod', async () => {
+      await request(app)
+        .get(`/api/agents/runtime/pods/${testPod._id}/messages`)
+        .set('Authorization', `Bearer ${clawdbotToken}`);
       // Clawdbot posts
       const clawdbotPost = await request(app)
         .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
         .set('Authorization', `Bearer ${clawdbotToken}`)
         .send({
+          continueAnyway: true,
           content: 'Message from Clawdbot Bridge',
           messageType: 'text',
         });
 
       expect(clawdbotPost.status).toBe(200);
 
+      await request(app)
+        .get(`/api/agents/runtime/pods/${testPod._id}/messages`)
+        .set('Authorization', `Bearer ${commonlyBotToken}`);
       // Commonly Bot posts
       const commonlyBotPost = await request(app)
         .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
         .set('Authorization', `Bearer ${commonlyBotToken}`)
         .send({
+          continueAnyway: true,
           content: 'Message from Commonly Bot',
           messageType: 'text',
         });

backend/__tests__/service/two-way-integration-e2e.test.js

@@ -588,6 +588,7 @@ describe('Two-Way Integration E2E Tests', () => {
         .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
         .set('Authorization', `Bearer ${runtimeToken}`)
         .send({
+          continueAnyway: true,
           content: summaryContent,
           messageType: 'text', // Valid enum values are: text, image
           metadata: {
@@ -784,6 +785,7 @@ describe('Two-Way Integration E2E Tests', () => {
         .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
         .set('Authorization', `Bearer ${runtimeToken}`)
         .send({
+          continueAnyway: true,
           content: summaryContent,
           messageType: 'text',
         });
@@ -898,6 +900,7 @@ describe('Two-Way Integration E2E Tests', () => {
         .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
         .set('Authorization', `Bearer ${runtimeToken}`)
         .send({
+          continueAnyway: true,
           content: `📊 **Discord Update**\n\nPR review discussion between Dev1 and Dev2.`,
           messageType: 'text',
         });
@@ -1181,6 +1184,7 @@ Based on the GroupMe activity, the team is discussing a PR review for the auth f
           .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
           .set('Authorization', `Bearer ${clawdbotToken}`)
           .send({
+            continueAnyway: true,
             content: aiResponse,
             messageType: 'text',
             metadata: {
@@ -1329,6 +1333,7 @@ Based on the GroupMe activity, the team is discussing a PR review for the auth f
           .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
           .set('Authorization', `Bearer ${commonlyToken}`)
           .send({
+            continueAnyway: true,
             content: '📊 **Discord Summary**: Deployment pipeline issue reported.',
             messageType: 'text',
           });
@@ -1338,6 +1343,7 @@ Based on the GroupMe activity, the team is discussing a PR review for the auth f
           .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
           .set('Authorization', `Bearer ${clawdbotToken}`)
           .send({
+            continueAnyway: true,
             content: `🤖 **Clawdbot Analysis**
 
 I noticed the deployment pipeline is failing. Based on similar issues I've seen:
@@ -1427,6 +1433,7 @@ Would you like me to help debug this further?`,
           .post(`/api/agents/runtime/pods/${testPod._id}/messages`)
           .set('Authorization', `Bearer ${tokenRes.body.token}`)
           .send({
+            continueAnyway: true,
             content: `🔍 **Code Review**
 
 \`\`\`javascript

backend/__tests__/unit/routes/agentsRuntime.contextBudget.test.js

@@ -30,6 +30,10 @@ jest.mock('../../../services/agentIdentityService', () => ({
   ensureAgentInPod: jest.fn().mockResolvedValue(undefined),
 }));
 jest.mock('../../../services/agentMessageService', () => ({ getRecentMessages: jest.fn() }));
+jest.mock('../../../services/agentSendFreshness', () => ({
+  readStart: jest.fn().mockResolvedValue({ at: new Date(0) }),
+  recordRead: jest.fn().mockResolvedValue(undefined),
+}));
 jest.mock('../../../services/agentThreadService', () => ({}));
 jest.mock('../../../services/podContextService', () => ({
   getPodContext: jest.fn().mockResolvedValue({ _status: 'success', stats: {} }),