fix(messages): a configured Postgres is the only message write path (TASK-503)

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

@@ -15,7 +15,9 @@ const request = require('supertest');
 const jwt = require('jsonwebtoken');
 const mongoose = require('mongoose');
 
-const { setupMongoDb, closeMongoDb, clearMongoDb } = require('../utils/testUtils');
+const {
+  setupMongoDb, closeMongoDb, clearMongoDb, setupPgDb, clearPgDb, closePgDb,
+} = 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
@@ -66,6 +68,9 @@ describe('Clawdbot E2E Integration Tests', () => {
   beforeAll(async () => {
     process.env.JWT_SECRET = JWT_SECRET;
     await setupMongoDb();
+    // Agent posts write PostgreSQL only (TASK-503); before it, these suites
+    // passed through the silent MongoDB fallback without a PG schema.
+    await setupPgDb();
 
     // Create Express app with routes
     app = express();
@@ -163,6 +168,7 @@ describe('Clawdbot E2E Integration Tests', () => {
   });
 
   afterAll(async () => {
+    await closePgDb();
     await closeMongoDb();
   });
 
@@ -172,6 +178,9 @@ describe('Clawdbot E2E Integration Tests', () => {
     await AgentEvent.deleteMany({});
     await AgentProfile.deleteMany({});
     await Message.deleteMany({});
+    // Agent posts now land in PG (TASK-503). Clear them too, or one test's
+    // posts count toward the next test's consecutive-run cap.
+    await clearPgDb();
     await Summary.deleteMany({});
     // Reset agent runtime tokens so each beforeEach gets a fresh token
     await User.updateMany({ isBot: true }, { $set: { agentRuntimeTokens: [] } });

backend/__tests__/service/messages.test.js

@@ -102,11 +102,20 @@ describe('Message Routes Integration Tests', () => {
     PGMessage.create.mockResolvedValue(mockMessage);
     const token = generateTestToken(user._id);
 
-    const response = await request(app)
-      .post(`/api/messages/${pod._id}`)
-      .set('Authorization', `Bearer ${token}`)
-      .send({ text: 'Hi there' })
-      .expect(200);
+    // The send writes PostgreSQL only when PG_HOST is configured (TASK-503).
+    const oldPgHost = process.env.PG_HOST;
+    process.env.PG_HOST = 'localhost';
+    let response;
+    try {
+      response = await request(app)
+        .post(`/api/messages/${pod._id}`)
+        .set('Authorization', `Bearer ${token}`)
+        .send({ text: 'Hi there' })
+        .expect(200);
+    } finally {
+      if (oldPgHost === undefined) delete process.env.PG_HOST;
+      else process.env.PG_HOST = oldPgHost;
+    }
 
     expect(response.body.content).toBe('Hi there');
     expect(PGMessage.create).toHaveBeenCalledWith(

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

@@ -21,7 +21,9 @@ const request = require('supertest');
 const express = require('express');
 const jwt = require('jsonwebtoken');
 
-const { setupMongoDb, closeMongoDb } = require('../utils/testUtils');
+const {
+  setupMongoDb, closeMongoDb, setupPgDb, closePgDb,
+} = require('../utils/testUtils');
 
 // Models
 const User = require('../../models/User');
@@ -103,6 +105,9 @@ describe('Two-Way Integration E2E Tests', () => {
   beforeAll(async () => {
     process.env.JWT_SECRET = JWT_SECRET;
     await setupMongoDb();
+    // Agent posts write PostgreSQL only (TASK-503); before it, these suites
+    // passed through the silent MongoDB fallback without a PG schema.
+    await setupPgDb();
 
     // Create Express app with routes
     app = express();
@@ -128,6 +133,7 @@ describe('Two-Way Integration E2E Tests', () => {
   });
 
   afterAll(async () => {
+    await closePgDb();
     await closeMongoDb();
   });
 

backend/__tests__/unit/controllers/messageController.test.js

@@ -172,6 +172,60 @@ describe('messageController', () => {
   });
 
   describe('createMessage', () => {
+    // The write follows PG_HOST like the socket and agent paths; the unit
+    // setup unsets it, so the PostgreSQL cases opt in here.
+    const OLD_PG_HOST = process.env.PG_HOST;
+    beforeEach(() => { process.env.PG_HOST = 'localhost'; });
+    afterEach(() => {
+      if (OLD_PG_HOST === undefined) delete process.env.PG_HOST;
+      else process.env.PG_HOST = OLD_PG_HOST;
+    });
+
+    describe('with no PostgreSQL configured (no PG_HOST)', () => {
+      beforeEach(() => { delete process.env.PG_HOST; });
+
+      it('writes the send to MongoDB, the only store, instead of failing it with 503', async () => {
+        const row = {
+          _id: { toString: () => 'mongo-1' },
+          podId: { toString: () => 'p1' },
+          userId: { toString: () => 'u1' },
+          content: 'hello',
+          messageType: 'text',
+        };
+        Pod.findById.mockResolvedValue({ members: ['u1'], type: 'chat' });
+        MongoMessage.create.mockResolvedValueOnce(row);
+        MongoMessage.findById.mockReturnValueOnce({ populate: jest.fn().mockResolvedValue(row) });
+        const req = { params: { podId: 'p1' }, body: { content: 'hello' }, user: { id: 'u1' } };
+        const res = { status: jest.fn().mockReturnThis(), json: jest.fn() };
+
+        await messageController.createMessage(req, res);
+
+        expect(PGMessage.create).not.toHaveBeenCalled();
+        expect(MongoMessage.create).toHaveBeenCalledWith({
+          podId: 'p1', userId: 'u1', content: 'hello', messageType: 'text',
+        });
+        expect(res.status).not.toHaveBeenCalledWith(503);
+        expect(res.json).toHaveBeenCalledWith(expect.objectContaining({ id: 'mongo-1', content: 'hello' }));
+      });
+
+      it('refuses a reply, which a MongoDB row cannot carry', async () => {
+        Pod.findById.mockResolvedValue({ members: ['u1'], type: 'chat' });
+        const req = {
+          params: { podId: 'p1' },
+          body: { content: 'a reply', replyToMessageId: 'parent-1' },
+          user: { id: 'u1' },
+        };
+        const res = { status: jest.fn().mockReturnThis(), json: jest.fn() };
+
+        await messageController.createMessage(req, res);
+
+        expect(res.status).toHaveBeenCalledWith(503);
+        expect(res.json).toHaveBeenCalledWith(expect.objectContaining({ code: 'REPLY_REQUIRES_POSTGRES' }));
+        expect(MongoMessage.create).not.toHaveBeenCalled();
+        expect(PGMessage.create).not.toHaveBeenCalled();
+      });
+    });
+
     it('turns the creator\'s first answer into one proposal while preserving explicit delivery', async () => {
       Pod.findById.mockResolvedValue({
         members: ['u1'], type: 'chat', createdBy: 'u1',
@@ -356,8 +410,8 @@ describe('messageController', () => {
 
       expect(res.status).toHaveBeenCalledWith(503);
       expect(res.json).toHaveBeenCalledWith({
-        error: 'Replies are temporarily unavailable. Please try again shortly.',
-        code: 'REPLY_REQUIRES_POSTGRES',
+        error: 'Messages are temporarily unavailable. Please try again shortly.',
+        code: 'MESSAGE_STORE_UNAVAILABLE',
       });
       expect(MongoMessage.create).not.toHaveBeenCalled();
       expect(AgentMentionService.enqueueMentions).not.toHaveBeenCalled();
@@ -398,38 +452,25 @@ describe('messageController', () => {
       }));
     });
 
-    it('keeps the Mongo availability fallback for a non-reply when PostgreSQL is unavailable', async () => {
-      const mongoFallbackRow = {
-        _id: { toString: () => 'mongo-message' },
-        podId: { toString: () => 'p1' },
-        userId: {
-          _id: { toString: () => 'u1' }, username: 'alice', profilePicture: 'default',
-        },
-        content: 'ordinary message',
-        messageType: 'text',
-      };
+    it('fails a non-reply with 503 when PostgreSQL is unavailable, writing nothing to MongoDB (TASK-503)', async () => {
       Pod.findById.mockResolvedValue({ members: ['u1'], type: 'chat' });
       PGMessage.create.mockRejectedValueOnce(new Error('PostgreSQL unavailable'));
-      MongoMessage.create.mockResolvedValueOnce(mongoFallbackRow);
-      MongoMessage.findById.mockReturnValueOnce({
-        populate: jest.fn().mockResolvedValue(mongoFallbackRow),
-      });
       const req = {
         params: { podId: 'p1' },
-        body: { content: mongoFallbackRow.content },
+        body: { content: 'ordinary message' },
         user: { id: 'u1' },
       };
       const res = { status: jest.fn().mockReturnThis(), json: jest.fn() };
 
       await messageController.createMessage(req, res);
 
-      expect(MongoMessage.create).toHaveBeenCalledWith({
-        pgFallback: true, podId: 'p1', userId: 'u1', content: mongoFallbackRow.content, messageType: 'text',
+      expect(res.status).toHaveBeenCalledWith(503);
+      expect(res.json).toHaveBeenCalledWith({
+        error: 'Messages are temporarily unavailable. Please try again shortly.',
+        code: 'MESSAGE_STORE_UNAVAILABLE',
       });
-      expect(res.status).not.toHaveBeenCalled();
-      expect(res.json).toHaveBeenCalledWith(expect.objectContaining({
-        id: 'mongo-message', content: mongoFallbackRow.content,
-      }));
+      expect(MongoMessage.create).not.toHaveBeenCalled();
+      expect(AgentMentionService.enqueueMentions).not.toHaveBeenCalled();
     });
 
     it('uses the joined PG author for a JWT-posted chat wake', async () => {

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

@@ -170,6 +170,32 @@ describe('agentsRuntime post message installation config forwarding', () => {
     }
   });
 
+  it('returns a retryable 503 when the message store is unavailable (TASK-503)', async () => {
+    const handler = getRouteHandler('/pods/:podId/messages', 'post');
+    AgentMessageService.postMessage.mockRejectedValueOnce(Object.assign(
+      new Error('Messages are temporarily unavailable. Please try again shortly.'),
+      { code: 'MESSAGE_STORE_UNAVAILABLE', statusCode: 503 },
+    ));
+    const installation = { podId: 'pod-1', agentName: 'openclaw', instanceId: 'tarik', config: {} };
+    const req = {
+      params: { podId: 'pod-1' }, body: { content: 'hello' },
+      agentInstallation: installation, agentInstallations: [installation],
+    };
+    const res = { status: jest.fn().mockReturnThis(), json: jest.fn() };
+    const warn = jest.spyOn(console, 'warn').mockImplementation(() => {});
+
+    try {
+      await handler(req, res);
+      expect(res.status).toHaveBeenCalledWith(503);
+      expect(res.json).toHaveBeenCalledWith({
+        message: 'Messages are temporarily unavailable. Please try again shortly.',
+        code: 'MESSAGE_STORE_UNAVAILABLE',
+      });
+    } finally {
+      warn.mockRestore();
+    }
+  });
+
   // 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

backend/__tests__/unit/services/agentMessageService.threadRootBroadcast.test.js

@@ -175,6 +175,22 @@ describe('agent broadcast carries thread_root_id', () => {
   });
 });
 
+describe('a configured PostgreSQL store is the only write path (TASK-503)', () => {
+  it('fails the post when the PG write throws, writing nothing to MongoDB', async () => {
+    PGMessage.create.mockRejectedValue(new Error('connection terminated'));
+
+    await expect(AgentMessageService.postMessage({
+      agentName: 'sprint-review',
+      instanceId: 'default',
+      podId: POD,
+      content: 'A live agent message',
+    })).rejects.toMatchObject({ code: 'MESSAGE_STORE_UNAVAILABLE', statusCode: 503 });
+
+    expect(Message).not.toHaveBeenCalled();
+    expect(emitted.find((e) => e.event === 'newMessage')).toBeUndefined();
+  });
+});
+
 describe('agent broadcast carries author identity', () => {
   it('preserves the bot marker through Mongo population and the socket whitelist', async () => {
     delete process.env.PG_HOST;
@@ -201,7 +217,9 @@ describe('agent broadcast carries author identity', () => {
     });
 
     const msg = emitted.find((e) => e.event === 'newMessage');
-    expect(Message).toHaveBeenCalledWith(expect.objectContaining({ pgFallback: true }));
+    // No PG configured: a plain MongoDB row, not a pgFallback one (TASK-503).
+    expect(Message).toHaveBeenCalledWith(expect.objectContaining({ content: 'A live agent message' }));
+    expect(Message.mock.calls[0][0]).not.toHaveProperty('pgFallback');
     expect(msg.payload.userId).toEqual(expect.objectContaining({
       _id: 'agent-user-1',
       isBot: true,

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

@@ -0,0 +1,87 @@
+/**
+ * TASK-503: with PostgreSQL configured, the socket send writes PG only. A PG
+ * that is still booting or a failed insert is a retryable error, never a
+ * MongoDB row (those rows carried no pgFallback flag and were never moved).
+ */
+
+jest.mock('../../../models/pg/Message', () => ({ create: jest.fn(), findById: jest.fn() }));
+jest.mock('../../../models/pg/Pod', () => ({ findById: jest.fn() }));
+jest.mock('../../../services/pgPodSyncService', () => ({ syncPodFromMongo: jest.fn() }));
+jest.mock('../../../models/Message', () => jest.fn());
+
+const PGMessage = require('../../../models/pg/Message');
+const PGPod = require('../../../models/pg/Pod');
+const { syncPodFromMongo } = require('../../../services/pgPodSyncService');
+const MongoMessage = require('../../../models/Message');
+const { writeSocketMessage, MESSAGE_STORE_UNAVAILABLE } = require('../../../services/socketMessageWrite');
+
+// The refusal names the send so the client can drop its optimistic copy.
+const REFUSED = {
+  ok: false,
+  error: {
+    ...MESSAGE_STORE_UNAVAILABLE, podId: 'p1', content: 'hi', messageType: 'text',
+  },
+};
+
+const input = (over = {}) => ({
+  pgReady: true, podId: 'p1', userId: 'u1', content: 'hi', messageType: 'text', replyToMessageId: null, ...over,
+});
+
+describe('writeSocketMessage', () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+    jest.spyOn(console, 'error').mockImplementation(() => {});
+    jest.spyOn(console, 'warn').mockImplementation(() => {});
+    PGPod.findById.mockResolvedValue({ id: 'p1' });
+  });
+  afterEach(() => jest.restoreAllMocks());
+
+  it('refuses while PG is still booting, without touching either store', async () => {
+    const result = await writeSocketMessage(input({ pgReady: false }));
+
+    expect(result).toEqual(REFUSED);
+    expect(PGMessage.create).not.toHaveBeenCalled();
+    expect(MongoMessage).not.toHaveBeenCalled();
+  });
+
+  it('returns a retryable error when the PG insert throws, writing nothing to MongoDB', async () => {
+    PGMessage.create.mockRejectedValue(new Error('connection terminated'));
+
+    const result = await writeSocketMessage(input());
+
+    expect(result).toEqual(REFUSED);
+    expect(MongoMessage).not.toHaveBeenCalled();
+  });
+
+  it('returns the joined row on success, passing the reply edge through', async () => {
+    PGMessage.create.mockResolvedValue({ id: 7, content: 'hi' });
+    PGMessage.findById.mockResolvedValue({ id: 7, content: 'hi', username: 'sam' });
+
+    const result = await writeSocketMessage(input({ replyToMessageId: '5' }));
+
+    expect(PGMessage.create).toHaveBeenCalledWith('p1', 'u1', 'hi', 'text', '5');
+    expect(result).toEqual({ ok: true, message: { id: 7, content: 'hi', username: 'sam' } });
+  });
+
+  it('keeps an already-stored message when only the joined re-read fails', async () => {
+    PGMessage.create.mockResolvedValue({ id: 8, content: 'hi' });
+    PGMessage.findById.mockRejectedValue(new Error('read timeout'));
+
+    const result = await writeSocketMessage(input());
+
+    expect(result).toEqual({ ok: true, message: { id: 8, content: 'hi' } });
+    expect(PGMessage.create).toHaveBeenCalledTimes(1);
+  });
+
+  it('heals a MongoDB-only pod into PG before the insert', async () => {
+    PGPod.findById.mockResolvedValue(null);
+    PGMessage.create.mockResolvedValue({ id: 9 });
+    PGMessage.findById.mockResolvedValue({ id: 9 });
+
+    await writeSocketMessage(input());
+
+    expect(syncPodFromMongo).toHaveBeenCalledWith('p1', 'u1');
+    expect(syncPodFromMongo.mock.invocationCallOrder[0])
+      .toBeLessThan(PGMessage.create.mock.invocationCallOrder[0]);
+  });
+});