fix(messages): a reply must target a message in its own pod (TASK-278)

backend/__tests__/service/threading.derivation.test.js

@@ -44,6 +44,7 @@ let pool;
 // has an FK to it. The first run of this suite failed on exactly that, which
 // is the point: a regex over the SQL cannot discover a foreign key.
 const POD = 'aaaaaaaaaaaaaaaaaaaaaa01';
+const OTHER_POD = 'aaaaaaaaaaaaaaaaaaaaaa02';
 const USER = 'bbbbbbbbbbbbbbbbbbbbbb01';
 
 const connect = () => new Pool({
@@ -59,10 +60,13 @@ const seed = async () => {
   await pool.query(
     `INSERT INTO users (_id, username) VALUES ($1,'tester') ON CONFLICT (_id) DO NOTHING`, [USER],
   );
-  await pool.query(
-    `INSERT INTO pods (id, name, type, created_by) VALUES ($1,'threading','chat',$2)
-     ON CONFLICT (id) DO NOTHING`, [POD, USER],
-  );
+  for (const podId of [POD, OTHER_POD]) {
+    // eslint-disable-next-line no-await-in-loop
+    await pool.query(
+      `INSERT INTO pods (id, name, type, created_by) VALUES ($1,'threading','chat',$2)
+       ON CONFLICT (id) DO NOTHING`, [podId, USER],
+    );
+  }
 };
 
 const insert = (content, replyTo = null) => PGMessage.create(POD, USER, content, 'text', replyTo);
@@ -83,7 +87,7 @@ d('thread_root_id derivation, executed', () => {
     if (pool) await pool.end();
     await closeMongoDb();
   });
-  beforeEach(async () => { await pool.query('DELETE FROM messages WHERE pod_id = $1', [POD]); });
+  beforeEach(async () => { await pool.query('DELETE FROM messages WHERE pod_id = ANY($1)', [[POD, OTHER_POD]]); });
 
   test('a root has no thread_root_id — NULL, not self-referential', async () => {
     const root = await insert('the root');
@@ -173,6 +177,44 @@ d('thread_root_id derivation, executed', () => {
     expect(ra.thread_root_id).not.toBe(rb.thread_root_id);
   });
 
+  // TASK-278, executed (Cofounder's HOLD on #2272): the INSERT's same-pod
+  // guard first shipped as `$1 IN (SELECT …)`, which pg-mem accepted and real
+  // Postgres refused for EVERY write with 42P08 (inconsistent types deduced
+  // for parameter $1). These cases are the ones that would have shown it, and
+  // they only mean something here.
+  describe('the INSERT writes a reply only when its target is in the pod', () => {
+    const countIn = async (podId) => Number((await pool.query(
+      'SELECT count(*)::int AS n FROM messages WHERE pod_id = $1', [podId],
+    )).rows[0].n);
+
+    test('a plain post is written', async () => {
+      const row = await insert('a plain post');
+      expect(row.id).toBeTruthy();
+      expect(await countIn(POD)).toBe(1);
+    });
+
+    test('a same-pod reply is written and derives its root', async () => {
+      const root = await insert('a root');
+      const reply = await insert('a reply', root.id);
+      expect(reply.thread_root_id).toBe(root.id);
+      expect(await countIn(POD)).toBe(2);
+    });
+
+    test('a cross-pod reply writes nothing and is refused', async () => {
+      const elsewhere = await PGMessage.create(OTHER_POD, USER, 'elsewhere', 'text', null);
+      await expect(insert('a cross-pod reply', elsewhere.id))
+        .rejects.toMatchObject({ name: 'ThreadRootError', code: 'reply_target_unavailable' });
+      expect(await countIn(POD)).toBe(0);
+    });
+
+    test('a root named in another pod writes nothing either', async () => {
+      const elsewhere = await PGMessage.create(OTHER_POD, USER, 'elsewhere', 'text', null);
+      await expect(PGMessage.create(POD, USER, 'aimed elsewhere', 'text', null, null, elsewhere.id))
+        .rejects.toMatchObject({ name: 'ThreadRootError', code: 'reply_target_unavailable' });
+      expect(await countIn(POD)).toBe(0);
+    });
+  });
+
   test('a reply to a missing parent is REJECTED, not silently rooted at NULL', async () => {
     // Corrects the backfill script's own comment. It says orphaned chains
     // "stay NULL by design", implying a bad edge can be written and later
@@ -189,8 +231,14 @@ d('thread_root_id derivation, executed', () => {
     // Must fail for the RIGHT reason. The suite's first run "passed" this
     // assertion because pod_id's FK rejected every insert — a control that
     // shares the failure mode proves nothing.
+    //
+    // Since TASK-278 the INSERT's same-pod guard refuses it first, so the FK
+    // is now the second line rather than the first. Same outcome: no row.
     await expect(insert('reply to nothing', 999999))
-      .rejects.toThrow(/messages_reply_to_message_id_fkey|violates foreign key/);
+      .rejects.toMatchObject({ name: 'ThreadRootError', code: 'reply_target_unavailable' });
+    expect(Number((await pool.query(
+      "SELECT count(*)::int AS n FROM messages WHERE content = 'reply to nothing'",
+    )).rows[0].n)).toBe(0);
     // Positive control: the same call with a REAL parent succeeds, so the
     // rejection above is about the missing parent and nothing else.
     const root = await insert('a real root');

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

@@ -27,12 +27,13 @@ jest.mock('../../../config/socket', () => ({
 }));
 jest.mock('../../../services/threadRootResolver', () => ({
   resolveThreadRoot: jest.fn(),
+  assertReplyTargetInPod: jest.fn().mockResolvedValue(undefined),
 }));
 jest.mock('../../../services/onboardingMarks', () => ({ markPodOpened: jest.fn(), markStartPicked: jest.fn() }));
 const { markPodOpened, markStartPicked } = require('../../../services/onboardingMarks');
 
 const socketConfig = require('../../../config/socket');
-const { resolveThreadRoot } = require('../../../services/threadRootResolver');
+const { resolveThreadRoot, assertReplyTargetInPod } = require('../../../services/threadRootResolver');
 
 describe('messageController', () => {
   beforeEach(() => {
@@ -166,6 +167,42 @@ describe('messageController', () => {
       });
     });
 
+    // TASK-278 (Cofounder's security read): a reply's target must be in the
+    // reply's pod, or this pod's preview shows another pod's message.
+    it('refuses a reply whose target is in another pod, before anything is written', async () => {
+      Pod.findById.mockResolvedValue({ members: ['u1'], type: 'chat' });
+      const err = Object.assign(new Error('the message being replied to is not in this pod'), {
+        name: 'ThreadRootError', code: 'reply_target_unavailable',
+      });
+      assertReplyTargetInPod.mockRejectedValueOnce(err);
+      const req = { params: { podId: 'p1' }, body: { content: 'a reply', replyToMessageId: '500' }, user: { id: 'u1' } };
+      const res = { status: jest.fn().mockReturnThis(), json: jest.fn() };
+
+      await messageController.createMessage(req, res);
+
+      expect(assertReplyTargetInPod).toHaveBeenCalledWith({ podId: 'p1', replyToMessageId: '500' });
+      expect(res.status).toHaveBeenCalledWith(400);
+      expect(res.json).toHaveBeenCalledWith({
+        error: 'the message being replied to is not in this pod', code: 'reply_target_unavailable',
+      });
+      expect(PGMessage.create).not.toHaveBeenCalled();
+      expect(AgentMentionService.enqueueMentions).not.toHaveBeenCalled();
+    });
+
+    it('a store failure in that check is left to the write, which answers it', async () => {
+      Pod.findById.mockResolvedValue({ members: ['u1'], type: 'chat' });
+      assertReplyTargetInPod.mockRejectedValueOnce(new Error('PostgreSQL unavailable'));
+      PGMessage.create.mockResolvedValueOnce({ id: 'pg-9', content: 'a reply' });
+      PGMessage.findById.mockResolvedValueOnce({ id: 'pg-9', content: 'a reply' });
+      const req = { params: { podId: 'p1' }, body: { content: 'a reply', replyToMessageId: '100' }, user: { id: 'u1' } };
+      const res = { status: jest.fn().mockReturnThis(), json: jest.fn() };
+
+      await messageController.createMessage(req, res);
+
+      expect(PGMessage.create).toHaveBeenCalled();
+      expect(res.status).not.toHaveBeenCalledWith(400);
+    });
+
     it('rejects a reply when PostgreSQL is unavailable instead of silently dropping its parent edge', async () => {
       const unsafeFallbackRow = {
         _id: { toString: () => 'mongo-reply' },

backend/__tests__/unit/models/threadRootResolver.test.js

@@ -15,7 +15,7 @@ const mockDb = newDb();
 const mockPool = new (mockDb.adapters.createPg().Pool)();
 jest.mock('../../../config/db-pg', () => ({ pool: mockPool }));
 
-const { resolveThreadRoot, ThreadRootError } = require('../../../services/threadRootResolver');
+const { resolveThreadRoot, assertReplyTargetInPod, ThreadRootError } = require('../../../services/threadRootResolver');
 const PGMessageTop = require('../../../models/pg/Message');
 
 const POD = 'pod-1';
@@ -225,3 +225,59 @@ describe('the resolver is not consulted when there is nothing to reconcile', ()
     expect(reply.thread_root_id).toBe(root.id);
   });
 });
+
+// Cofounder's security read, 2026-10-16 (TASK-278): with no threadRootId, a
+// reply's target was never checked to be in the reply's pod. The reply joined
+// the other pod's thread, and this pod's preview showed that message's text
+// and author. Each case runs the real check and the real SQL.
+describe('a reply must target a message in its own pod', () => {
+  const codeOf = (promise) => promise.then(() => 'resolved', (err) => err.code);
+
+  test('a target in the same pod passes', async () => {
+    await expect(assertReplyTargetInPod({ podId: POD, replyToMessageId: 100 })).resolves.toBeUndefined();
+    await expect(assertReplyTargetInPod({ podId: POD })).resolves.toBeUndefined();
+  });
+
+  test.each([
+    ['in another pod', 300, 'reply_target_unavailable'],
+    // The same code as another pod's message: a reply must not reveal which
+    // ids exist elsewhere.
+    ['that does not exist', 999, 'reply_target_unavailable'],
+    ['that is not an id', 'abc', 'reply_target_invalid'],
+  ])('a target %s is refused with %s', async (_label, replyToMessageId, code) => {
+    expect(await codeOf(assertReplyTargetInPod({ podId: POD, replyToMessageId }))).toBe(code);
+  });
+
+  test('the resolver no longer derives a root from another pod\'s message', async () => {
+    expect(await codeOf(resolveThreadRoot({ podId: POD, replyToMessageId: 300 }))).toBe('reply_target_unavailable');
+  });
+
+  test('the INSERT refuses a cross-pod reply, so a path that skips the check writes nothing', async () => {
+    const before = (await mockPool.query("SELECT count(*)::int AS n FROM messages WHERE content = 'cross-pod-reply'")).rows[0].n;
+    const err = await PGMessageTop.create(POD, 'u', 'cross-pod-reply', 'text', '300').catch((e) => e);
+    expect(err).toBeInstanceOf(ThreadRootError);
+    expect(err.code).toBe('reply_target_unavailable');
+    const after = (await mockPool.query("SELECT count(*)::int AS n FROM messages WHERE content = 'cross-pod-reply'")).rows[0].n;
+    expect(after).toBe(before);
+  });
+
+  test('CONTROL: a same-pod reply is still written and derives its root', async () => {
+    const reply = await PGMessageTop.create(POD, 'u', 'same-pod-reply', 'text', '100');
+    expect(reply.thread_root_id).toBe(100);
+  });
+
+  test('an existing cross-pod reply row no longer previews the other pod\'s message', async () => {
+    await msg(410, OTHER);                // the other pod's message
+    await msg(411, POD, null, 410);       // a reply written before this fix
+    await msg(412, POD, null, 100);       // CONTROL: an ordinary reply in the pod
+    const rows = await PGMessageTop.findByPodId(POD, 200);
+    const leaked = rows.find((row) => String(row.id) === '411');
+    const ordinary = rows.find((row) => String(row.id) === '412');
+    expect(leaked).toBeTruthy();
+    expect(leaked.replyTo).toBeNull();
+    expect(ordinary.replyTo).toEqual(expect.objectContaining({ id: '100' }));
+    // The single-message read (what a new message is broadcast from) too.
+    expect((await PGMessageTop.findById('411')).replyTo).toBeNull();
+    expect((await PGMessageTop.findById('412')).replyTo).toEqual(expect.objectContaining({ id: '100' }));
+  });
+});

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

@@ -87,7 +87,7 @@ jest.mock('../../../services/threadRootResolver', () => {
       this.code = code;
     }
   }
-  return { resolveThreadRoot: jest.fn(), ThreadRootError };
+  return { resolveThreadRoot: jest.fn(), assertReplyTargetInPod: jest.fn().mockResolvedValue(undefined), ThreadRootError };
 });
 jest.mock('../../../services/installable/eventHandlers', () => ({
   dispatch: (...args) => mockDispatch(...args),
@@ -305,6 +305,22 @@ describe('agent in-thread posts (threadRootId, no reply edge)', () => {
     expect(emitted).toHaveLength(0);
   });
 
+  // TASK-278: an agent's reply to a message in another pod is refused before
+  // the write, and the refusal propagates instead of falling into Mongo.
+  it('refuses a reply whose target is in another pod, before anything is written', async () => {
+    threadRootResolver.assertReplyTargetInPod.mockRejectedValueOnce(
+      new threadRootResolver.ThreadRootError('the message being replied to is not in this pod', 'reply_target_unavailable'),
+    );
+
+    await expect(AgentMessageService.postMessage({
+      agentName: 'sprint-review', instanceId: 'default', podId: POD, content: 'a reply', replyToMessageId: '500',
+    })).rejects.toMatchObject({ name: 'ThreadRootError', code: 'reply_target_unavailable' });
+
+    expect(threadRootResolver.assertReplyTargetInPod).toHaveBeenCalledWith({ podId: POD, replyToMessageId: '500' });
+    expect(PGMessage.create).not.toHaveBeenCalled();
+    expect(emitted).toHaveLength(0);
+  });
+
   it('refuses a thread-scoped post without the PG store instead of stranding it in Mongo', async () => {
     delete process.env.PG_HOST;
 

backend/__tests__/unit/services/nativeRuntimeService.webFetch.test.js

@@ -45,6 +45,9 @@ jest.mock('../../../services/messageClaimService', () => ({
 jest.mock('../../../services/nativeOutsiderGrantPolicy', () => ({
   podHasOutsiderWritableGrant: jest.fn().mockResolvedValue(false),
 }));
+// The holder of a lost claim was addressed too (holderWasAddressed).
+const mockHolderEvent = jest.fn();
+jest.mock('../../../models/AgentEvent', () => ({ exists: (...args) => mockHolderEvent(...args) }));
 const mockSandboxRun = jest.fn();
 jest.mock('../../../services/turnSandboxService', () => ({
   isSandboxConfigured: () => true,
@@ -88,6 +91,7 @@ jest.mock('../../../services/grantBrokerProjectionService', () => ({
 const axios = require('axios').default;
 const AgentRun = require('../../../models/AgentRun');
 const { TurnSandbox } = require('../../../services/turnSandboxService');
+const MessageClaimService = require('../../../services/messageClaimService');
 const { runAgent } = require('../../../services/nativeRuntimeService');
 
 const REPO = 'https://github.com/acme/widget';
@@ -172,6 +176,31 @@ describe('native turn with commonly_fetch_url', () => {
     expect(savedRun.turns[0].toolCalls[0].result).toEqual({ url: REPO, status: 200, truncated: false, chars: 50 });
   });
 
+  // Cofounder's security read on #2269: the peer-claim notice was prepended
+  // before the person's message entered the allowed set, so the notice's own
+  // text could give an address. Unreachable today: an agent name carries no
+  // address, and the notice's leading '@' spoils a bare one. The holder here
+  // carries a space to stand in for any notice text that could.
+  test('the peer-claim notice gives no link; the person\'s message still does', async () => {
+    MessageClaimService.claim.mockResolvedValueOnce({
+      claimed: false, claimedBy: 'clerk holder.example/x', instanceId: 'default',
+    });
+    mockHolderEvent.mockResolvedValueOnce({ _id: 'evt-holder' });
+    axios.post
+      .mockResolvedValueOnce(toolCalls(
+        call('c1', 'commonly_fetch_url', { url: 'https://holder.example/x' }),
+        call('c2', 'commonly_fetch_url', { url: REPO }),
+      ))
+      .mockResolvedValueOnce(finalReply);
+
+    await runAgent(install(), mention());
+
+    expect(axios.post.mock.calls[0][1].messages[1].content).toContain('[Claim notice: @clerk holder.example/x holds');
+    expect(mockGetPage).toHaveBeenCalledTimes(1);
+    expect(mockGetPage).toHaveBeenCalledWith(REPO);
+    expect(toolMessages(1)[0].content).toMatch(/does not appear in this conversation/);
+  });
+
   test('a scheme-less link is opened over https', async () => {
     axios.post
       .mockResolvedValueOnce(toolCalls(call('c1', 'commonly_fetch_url', { url: 'github.com/acme/widget' })))

backend/controllers/messageController.ts

@@ -258,6 +258,16 @@ exports.createMessage = async (req: AuthRequest, res: Response): Promise<void> =
         // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require
         const { resolveThreadRoot } = require('../services/threadRootResolver');
         resolvedThreadRootId = await resolveThreadRoot({ podId, replyToMessageId, threadRootId });
+      } else if (replyToMessageId) {
+        // A reply's target must be in this pod (TASK-278). A store failure here
+        // is left to the write below, which answers it and re-checks in SQL.
+        // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require
+        const { assertReplyTargetInPod } = require('../services/threadRootResolver');
+        try {
+          await assertReplyTargetInPod({ podId, replyToMessageId });
+        } catch (checkErr) {
+          if ((checkErr as { name?: string })?.name === 'ThreadRootError') throw checkErr;
+        }
       }
     } catch (err) {
       const e = err as { name?: string; code?: string; message?: string };

backend/models/pg/Message.ts

@@ -148,6 +148,10 @@ class Message {
                   FROM messages parent
                  WHERE parent.id = $5::int)
              )::int
+       WHERE ($5::int IS NULL
+              OR $1::varchar IN (SELECT target.pod_id FROM messages target WHERE target.id = $5::int))
+         AND ($7::int IS NULL
+              OR $1::varchar IN (SELECT named.pod_id FROM messages named WHERE named.id = $7::int))
       RETURNING *
     `;
     try {
@@ -156,6 +160,14 @@ class Message {
         payload == null ? null : JSON.stringify(payload),
         resolvedThreadRootId ?? null,
       ]);
+      // No row: the reply target or the named root is not in this pod
+      // (services/threadRootResolver). Callers check first; this is the
+      // backstop for a path that does not, and for a message deleted in between.
+      if (!result.rows[0]) {
+        // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports
+        const { ThreadRootError } = require('../../services/threadRootResolver');
+        throw new ThreadRootError('reply_target_unavailable', 'the message being replied to is not in this pod');
+      }
       await (pool as PgPool).query(
         'UPDATE pods SET updated_at = CURRENT_TIMESTAMP WHERE id = $1',
         [podId],
@@ -212,7 +224,7 @@ class Message {
           rm.user_id as reply_user_id, ru.username as reply_username
         FROM messages m
         LEFT JOIN users u ON m.user_id = u._id
-        LEFT JOIN messages rm ON m.reply_to_message_id = rm.id
+        LEFT JOIN messages rm ON m.reply_to_message_id = rm.id AND rm.pod_id = m.pod_id
         LEFT JOIN users ru ON rm.user_id = ru._id
         WHERE m.pod_id = $1
       `;
@@ -259,7 +271,7 @@ class Message {
           rm.user_id as reply_user_id, ru.username as reply_username
         FROM messages m
         LEFT JOIN users u ON m.user_id = u._id
-        LEFT JOIN messages rm ON m.reply_to_message_id = rm.id
+        LEFT JOIN messages rm ON m.reply_to_message_id = rm.id AND rm.pod_id = m.pod_id
         LEFT JOIN users ru ON rm.user_id = ru._id
         WHERE m.id = $1
       `;

backend/server.ts

@@ -608,6 +608,22 @@ io.on('connection', (socket: any) => {
           return;
         }
 
+        // A reply's target must be in this pod (TASK-278). Refused here, before
+        // the write, so the refusal cannot fall into the Mongo branch below.
+        if (pgAvailable && replyToMessageId) {
+          try {
+            // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports
+            const { assertReplyTargetInPod } = require('./services/threadRootResolver');
+            await assertReplyTargetInPod({ podId, replyToMessageId });
+          } catch (checkErr) {
+            const e = checkErr as { name?: string; code?: string; message?: string };
+            if (e.name === 'ThreadRootError') {
+              socket.emit('error', { message: e.message, code: e.code });
+              return;
+            }
+          }
+        }
+
         // Use PostgreSQL for messages if available, otherwise fallback to MongoDB
         if (pgAvailable) {
           try {