fix(attention): refresh #1577 replied-stamp audit onto main (TASK-387)

backend/__tests__/unit/scripts/audit-replied-attention-stamps.test.js

@@ -0,0 +1,53 @@
+// @ts-nocheck
+
+// Only the argv parsing is under test here; the audit itself is covered in
+// __tests__/unit/services/attentionItemService.test.js. Stub the modules the
+// script pulls in at import time so this needs no database.
+jest.mock('mongoose', () => ({ connect: jest.fn(), disconnect: jest.fn() }));
+jest.mock('../../../services/attentionItemService', () => ({ auditRepliedMentionAttention: jest.fn() }));
+
+const { parseArgs } = require('../../../scripts/audit-replied-attention-stamps');
+
+describe('audit-replied-attention-stamps argv', () => {
+  it('reads --apply and an ISO bound', () => {
+    const { apply, resolvedBefore } = parseArgs(['node', 'script', '--apply', '--resolved-before=2026-09-06T12:42:00Z']);
+    expect(apply).toBe(true);
+    expect(resolvedBefore.toISOString()).toBe('2026-09-06T12:42:00.000Z');
+  });
+
+  it('defaults to a dry run over every replied stamp when no flags are given', () => {
+    expect(parseArgs(['node', 'script'])).toEqual({ apply: false });
+  });
+
+  // The whole point of the flag is to NARROW the scan to the cutover window.
+  // Falling through as `undefined` on a bad value widens it to every replied
+  // stamp ever written — under --apply, an unbounded reopen from a typo.
+  it.each(['--resolved-before=garbage', '--resolved-before=', '--resolved-before=2026-13-45'])(
+    'throws rather than silently widening the scan: %s',
+    (flag) => {
+      expect(() => parseArgs(['node', 'script', flag])).toThrow(/not a date/);
+    },
+  );
+
+  // The space form is the one a shell user types. It used to parse as an
+  // unbounded run under --apply: the bound was never read, so the scan widened.
+  it('reads the space form and keeps --apply after it', () => {
+    const { apply, resolvedBefore } = parseArgs(['node', 'script', '--resolved-before', '2026-09-06', '--apply']);
+    expect(apply).toBe(true);
+    expect(resolvedBefore.toISOString()).toBe('2026-09-06T00:00:00.000Z');
+  });
+
+  it.each([
+    ['--resolved-before', '--apply'],
+    ['--resolved-before'],
+  ])('throws when the bound has no value: %s', (...argv) => {
+    expect(() => parseArgs(['node', 'script', ...argv])).toThrow(/not a date/);
+  });
+
+  it.each([['--aplly'], ['--apply=yes'], ['--resolved_before=2026-09-06']])(
+    'throws on an unknown or misspelled flag: %s',
+    (flag) => {
+      expect(() => parseArgs(['node', 'script', flag])).toThrow(/Unknown argument/);
+    },
+  );
+});

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

@@ -693,6 +693,92 @@ describe('attentionItemService', () => {
     expect(sourcePattern.test('task.1:update-1')).toBe(true);
     expect(sourcePattern.test('taskx1:update-1')).toBe(false);
   });
+
+  it('reopens a replied stamp the source contradicts, and keeps the one it corroborates', async () => {
+    const rows = [
+      {
+        _id: 'attention-false', recipientUserId: 'sam', podId: 'pod-1', kind: 'mention',
+        source: { type: 'message', id: '42' }, messageId: '42', threadRootId: '40',
+        sourceCreatedAt: new Date('2026-09-05T09:00:00.000Z'),
+      },
+      {
+        _id: 'attention-true', recipientUserId: 'sam', podId: 'pod-1', kind: 'mention',
+        source: { type: 'message', id: '43' }, messageId: '43', threadRootId: '43',
+        sourceCreatedAt: new Date('2026-09-05T09:30:00.000Z'),
+      },
+    ];
+    mockFind.mockReturnValue({ sort: () => ({ lean: async () => rows }) });
+    mockPgMessageHasReplyByUserAfter.mockImplementation(async (podId, userId, after, target) => target.messageId === '43');
+
+    const result = await AttentionItemService.auditRepliedMentionAttention({ apply: true });
+
+    expect(result).toEqual({ scanned: 2, kept: 1, contradicted: 1, reopened: 1, undecidable: 0, unavailable: 0 });
+    expect(mockUpdateOne).toHaveBeenCalledTimes(1);
+    expect(mockUpdateOne).toHaveBeenCalledWith(
+      { _id: 'attention-false', kind: 'mention', status: 'resolved', resolvedBy: 'replied' },
+      { $set: { status: 'open' }, $unset: { resolvedAt: '', resolvedBy: '' } },
+    );
+  });
+
+  it('changes nothing without --apply, and still reports what it would reopen', async () => {
+    mockFind.mockReturnValue({ sort: () => ({ lean: async () => [{
+      _id: 'attention-false', recipientUserId: 'sam', podId: 'pod-1', kind: 'mention',
+      source: { type: 'message', id: '42' }, messageId: '42',
+      sourceCreatedAt: new Date('2026-09-05T09:00:00.000Z'),
+    }] }) });
+    mockPgMessageHasReplyByUserAfter.mockResolvedValue(false);
+
+    const result = await AttentionItemService.auditRepliedMentionAttention();
+
+    expect(result).toMatchObject({ scanned: 1, contradicted: 1, reopened: 0 });
+    expect(mockUpdateOne).not.toHaveBeenCalled();
+  });
+
+  it('counts a Mongo-sourced stamp as undecidable rather than reopening it', async () => {
+    mockFind.mockReturnValue({ sort: () => ({ lean: async () => [{
+      _id: 'attention-mongo', recipientUserId: 'sam', podId: 'pod-1', kind: 'mention',
+      source: { type: 'message', id: '507f191e810c19729de860ea' },
+      sourceCreatedAt: new Date('2026-09-05T09:00:00.000Z'),
+    }] }) });
+
+    const result = await AttentionItemService.auditRepliedMentionAttention({ apply: true });
+
+    expect(result).toMatchObject({ scanned: 1, kept: 0, contradicted: 0, reopened: 0, undecidable: 1 });
+    expect(mockUpdateOne).not.toHaveBeenCalled();
+    expect(mockPgMessageHasReplyByUserAfter).not.toHaveBeenCalled();
+  });
+
+  it('counts a source-less stamp as undecidable and never reopens it', async () => {
+    mockFind.mockReturnValue({ sort: () => ({ lean: async () => [{
+      _id: 'attention-sourceless', recipientUserId: 'sam', podId: 'pod-1', kind: 'mention',
+      sourceCreatedAt: new Date('2026-09-05T09:00:00.000Z'),
+    }] }) });
+
+    const result = await AttentionItemService.auditRepliedMentionAttention({ apply: true });
+
+    expect(result).toMatchObject({ scanned: 1, kept: 0, contradicted: 0, reopened: 0, undecidable: 1 });
+    expect(mockUpdateOne).not.toHaveBeenCalled();
+    expect(mockPgMessageHasReplyByUserAfter).not.toHaveBeenCalled();
+  });
+
+  it('reads only replied stamps, and honours the resolvedBefore bound', async () => {
+    mockFind.mockReturnValue({ sort: () => ({ lean: async () => [] }) });
+    const before = new Date('2026-09-06T00:00:00.000Z');
+
+    await AttentionItemService.auditRepliedMentionAttention({ resolvedBefore: before });
+
+    expect(mockFind).toHaveBeenCalledWith({
+      kind: 'mention', status: 'resolved', resolvedBy: 'replied', resolvedAt: { $lt: before },
+    });
+  });
+
+  it('omits the resolvedAt bound entirely when none is given', async () => {
+    mockFind.mockReturnValue({ sort: () => ({ lean: async () => [] }) });
+
+    await AttentionItemService.auditRepliedMentionAttention();
+
+    expect(mockFind).toHaveBeenCalledWith({ kind: 'mention', status: 'resolved', resolvedBy: 'replied' });
+  });
 });
 
 // Stickiness Phase 1b: an agent's answer to the

backend/package.json

@@ -29,6 +29,7 @@
     "migrate:task-priority": "ts-node scripts/migrate-task-priority.ts",
     "migrate:task-source-ref-identity": "ts-node scripts/migrate-task-source-ref-identity.ts",
     "sweep:resolved-mention-attention": "ts-node scripts/sweep-resolved-mention-attention.ts",
+    "audit:replied-attention-stamps": "ts-node scripts/audit-replied-attention-stamps.ts",
     "sweep:done-task-handoffs": "ts-node scripts/sweep-done-task-handoffs.ts",
     "sweep:leftover-runtime-tokens": "ts-node scripts/sweep-leftover-runtime-tokens.ts",
     "sweep:stale-digest": "ts-node scripts/sweep-stale-digest.ts",

backend/scripts/audit-replied-attention-stamps.ts

@@ -0,0 +1,84 @@
+/**
+ * Audit mention AttentionItems stamped `resolvedBy: 'replied'`. The broad
+ * pre-#1573 resolver closed every open mention a recipient held in a pod as
+ * soon as they posted anything there, so some of those stamps assert a reply
+ * nobody wrote. This re-reads each row's source and reopens only the rows the
+ * source contradicts; rows whose source cannot answer the question are
+ * counted and left alone.
+ *
+ * Dry run by default:
+ *   npm run audit:replied-attention-stamps
+ * Bound the population to stamps written before the narrow resolver deployed:
+ *   npm run audit:replied-attention-stamps -- --resolved-before=2026-09-06T00:00:00Z
+ * Apply the source-backed reopens:
+ *   npm run audit:replied-attention-stamps -- --apply
+ */
+/* eslint-disable no-console */
+const mongoose = require('mongoose');
+const { auditRepliedMentionAttention } = require('../services/attentionItemService');
+
+type PgPool = { end: () => Promise<void> };
+
+/**
+ * Read the flags out of argv. Exported so the throws below are testable without
+ * a database. Three things must fail loudly, never fall through: a bound that
+ * is missing or unparsable (it would widen the audit from the cutover window
+ * to every replied stamp ever written), a bound whose value is swallowed by a
+ * neighbouring flag, and an unknown flag such as a misspelled `--apply`. Under
+ * `--apply` each of those is an unbounded or unintended reopen.
+ */
+export const parseArgs = (argv: string[]): { apply: boolean; resolvedBefore?: Date } => {
+  const args = argv.slice(2);
+  const BOUND = '--resolved-before';
+  let apply = false;
+  let sawBound = false;
+  let raw = '';
+  for (let i = 0; i < args.length; i += 1) {
+    const arg = args[i];
+    if (arg === '--apply') {
+      apply = true;
+    } else if (arg === BOUND || arg.startsWith(`${BOUND}=`)) {
+      sawBound = true;
+      if (arg === BOUND) {
+        raw = args[i + 1] ?? '';
+        i += 1;
+      } else {
+        raw = arg.slice(BOUND.length + 1);
+      }
+    } else {
+      throw new Error(`Unknown argument: ${arg}`);
+    }
+  }
+  if (!sawBound) return { apply };
+  const parsed = new Date(raw);
+  if (!raw.trim() || raw.startsWith('--') || Number.isNaN(parsed.getTime())) {
+    throw new Error(`--resolved-before is not a date: ${raw || '(missing)'}`);
+  }
+  return { apply, resolvedBefore: parsed };
+};
+
+export const main = async (): Promise<void> => {
+  if (!process.env.MONGO_URI) throw new Error('MONGO_URI is required');
+  const { apply: APPLY, resolvedBefore } = parseArgs(process.argv);
+  // The audit reads legacy PostgreSQL mention sources as well as Mongo ones.
+  // Close that pool like the sweep does so the one-shot process exits after
+  // printing its measured result.
+  // eslint-disable-next-line global-require
+  const { pool } = require('../config/db-pg') as { pool: PgPool | null };
+  await mongoose.connect(process.env.MONGO_URI);
+  try {
+    const result = await auditRepliedMentionAttention({ apply: APPLY, resolvedBefore });
+    console.log(JSON.stringify({ ...result, apply: APPLY, resolvedBefore: resolvedBefore?.toISOString() || null }));
+    if (!APPLY) console.log('DRY RUN — no AttentionItems changed. Re-run with --apply after review.');
+  } finally {
+    await mongoose.disconnect();
+    if (pool) await pool.end();
+  }
+};
+
+if (require.main === module) {
+  main().catch((error) => {
+    console.error('replied-stamp audit failed:', error);
+    process.exit(1);
+  });
+}

backend/services/attentionItemService.ts

@@ -401,6 +401,59 @@ export const sweepMissedAgentMentions = async (
   return { scanned: rows.length, eligible: ids.length, missed: Number(result.modifiedCount || 0) };
 };
 
+/**
+ * Audit rows already stamped `resolvedBy: 'replied'`. Before the narrow
+ * resolver landed, any post by the recipient in the pod closed every open
+ * mention they held there, so a row can assert a reply that was never
+ * written. This re-reads each row's source with the same evidence the sweep
+ * uses and reopens only the ones the source contradicts.
+ *
+ * Three outcomes, not two. `undecidable` exists because Mongo-fallback
+ * sources persist no reply or thread edges: absence of evidence there is not
+ * evidence the recipient stayed silent, and reopening those would re-raise
+ * attention a user has already dealt with. They are counted and left alone.
+ */
+export const auditRepliedMentionAttention = async (
+  { apply = false, resolvedBefore }: { apply?: boolean; resolvedBefore?: Date } = {},
+) => {
+  const before = validDate(resolvedBefore);
+  const rows = await AttentionItem.find({
+    kind: 'mention',
+    status: 'resolved',
+    resolvedBy: 'replied',
+    ...(before ? { resolvedAt: { $lt: before } } : {}),
+  }).sort({ createdAt: 1 }).lean();
+  let kept = 0;
+  let contradicted = 0;
+  let reopened = 0;
+  let undecidable = 0;
+  let unavailable = 0;
+  for (const row of rows) {
+    try {
+      const sourceId = String(row?.source?.id || '');
+      // Only a PostgreSQL source carries the reply and thread edges this
+      // audit needs; see recipientRepliedAfterMention.
+      if (!/^\d+$/.test(sourceId)) { undecidable += 1; continue; }
+      const sourceCreatedAt = await sourceTimeForMention(row);
+      if (!sourceCreatedAt) { undecidable += 1; continue; }
+      if (await recipientRepliedAfterMention(row, sourceCreatedAt)) { kept += 1; continue; }
+      contradicted += 1;
+      if (!apply) continue;
+      // Repeat the stamp in the filter so a row an operator acknowledged
+      // between the read and the write is not reopened underneath them.
+      const result = await AttentionItem.updateOne(
+        { _id: row._id, kind: 'mention', status: 'resolved', resolvedBy: 'replied' },
+        { $set: { status: 'open' }, $unset: { resolvedAt: '', resolvedBy: '' } },
+      );
+      reopened += Number(result.modifiedCount || 0);
+    } catch (error) {
+      unavailable += 1;
+      console.warn('[attention] replied-stamp audit skipped an unreadable source:', (error as Error).message);
+    }
+  }
+  return { scanned: rows.length, kept, contradicted, reopened, undecidable, unavailable };
+};
+
 export const recordApproval = async (approval: any, options: RecordOptions = {}): Promise<void> => {
   try {
     const podId = approval?.podId;
@@ -962,6 +1015,6 @@ export const hasEverHadAttention = async (recipientUserId: unknown): Promise<boo
 // excluding true decisions and approvals.
 export const acknowledgeMention = acknowledgeAttention;
 
-export default { recordMentionedUsers, resolveMentionAttentionForReply, sweepResolvedMentionAttention, sweepMissedAgentMentions, sweepDoneTaskHandoffs, recordApproval, recordActionApproval, recordDecision, recordTaskAttention, resolveTaskAttention, recordInitiativeStall, resolveInitiativeStalls, initiativeStallSourceId, resolve, resolveActioned, resolveMany, getOpenQueue, hasEverHadAttention, acknowledgeAttention, acknowledgeMention };
+export default { recordMentionedUsers, resolveMentionAttentionForReply, sweepResolvedMentionAttention, auditRepliedMentionAttention, sweepMissedAgentMentions, sweepDoneTaskHandoffs, recordApproval, recordActionApproval, recordDecision, recordTaskAttention, resolveTaskAttention, recordInitiativeStall, resolveInitiativeStalls, initiativeStallSourceId, resolve, resolveActioned, resolveMany, getOpenQueue, hasEverHadAttention, acknowledgeAttention, acknowledgeMention };
 // eslint-disable-next-line @typescript-eslint/no-require-imports
-module.exports = { recordMentionedUsers, resolveMentionAttentionForReply, sweepResolvedMentionAttention, sweepMissedAgentMentions, AGENT_MENTION_MISSED_DAYS, sweepDoneTaskHandoffs, recordApproval, recordActionApproval, recordDecision, recordTaskAttention, resolveTaskAttention, recordInitiativeStall, resolveInitiativeStalls, initiativeStallSourceId, resolve, resolveActioned, resolveMany, getOpenQueue, hasEverHadAttention, acknowledgeAttention, acknowledgeMention, TASK_HANDOFF_CANDIDATE_RE, isHandoffRequest, handoffRecipients };
+module.exports = { recordMentionedUsers, resolveMentionAttentionForReply, sweepResolvedMentionAttention, auditRepliedMentionAttention, sweepMissedAgentMentions, AGENT_MENTION_MISSED_DAYS, sweepDoneTaskHandoffs, recordApproval, recordActionApproval, recordDecision, recordTaskAttention, resolveTaskAttention, recordInitiativeStall, resolveInitiativeStalls, initiativeStallSourceId, resolve, resolveActioned, resolveMany, getOpenQueue, hasEverHadAttention, acknowledgeAttention, acknowledgeMention, TASK_HANDOFF_CANDIDATE_RE, isHandoffRequest, handoffRecipients };