fix: prevent duplicate outreach and follow-ups

src/database/client.ts

@@ -217,6 +217,40 @@ export function runMigrations(db: Database.Database): void {
         );
       `);
     },
+    () => db.exec(`
+      CREATE INDEX IF NOT EXISTS idx_reddit_conversations_global_username
+        ON reddit_conversations(username COLLATE NOCASE);
+      CREATE INDEX IF NOT EXISTS idx_sent_messages_global_recipient
+        ON sent_messages(recipient COLLATE NOCASE);
+
+      INSERT OR IGNORE INTO recipient_suppressions (recipient, reason, created_at)
+      SELECT recipient, 'sent-message-history', MIN(sent_at)
+      FROM sent_messages
+      WHERE trim(recipient) <> ''
+      GROUP BY recipient COLLATE NOCASE;
+
+      INSERT OR IGNORE INTO recipient_suppressions (recipient, reason, created_at)
+      SELECT rc.username, 'historical-conversation', MIN(rc.updated_at)
+      FROM reddit_conversations rc
+      WHERE trim(rc.username) <> ''
+        AND lower(trim(rc.username)) NOT IN ('unknown', '[deleted]')
+        AND lower(trim(rc.username)) NOT LIKE 't2\\_%' ESCAPE '\\'
+        AND trim(rc.username) NOT LIKE '@%'
+      GROUP BY rc.username COLLATE NOCASE;
+
+      CREATE TABLE conversation_automation_claims (
+        account_id TEXT NOT NULL,
+        conversation_id TEXT NOT NULL,
+        incoming_key TEXT NOT NULL,
+        status TEXT NOT NULL CHECK(status IN ('pending', 'sent')),
+        claimed_at INTEGER NOT NULL,
+        sent_at INTEGER,
+        PRIMARY KEY (account_id, conversation_id, incoming_key),
+        FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id) ON DELETE CASCADE
+      );
+      CREATE INDEX idx_conversation_automation_claims_status
+        ON conversation_automation_claims(account_id, status, claimed_at);
+    `),
   ];
 
   let currentVersion = db.pragma('user_version', { simple: true }) as number;

src/database/repositories.ts

@@ -120,9 +120,12 @@ export interface DeliveryOutboxRecord {
   sentMessageId: number | null;
 }
 
-/** Read-only recipient history shared by every Reddit account runtime. */
+/** Global recipient history and durable pre-send claims shared by all account runtimes. */
 export class RecipientHistoryRepository {
   private readonly hasSentToStmt: Database.Statement;
+  private readonly tryClaimStmt: Database.Statement;
+  private readonly commitClaimStmt: Database.Statement;
+  private readonly releaseClaimStmt: Database.Statement;
 
   constructor(db: Database.Database) {
     this.hasSentToStmt = db.prepare(`
@@ -130,13 +133,49 @@ export class RecipientHistoryRepository {
         SELECT recipient FROM sent_messages WHERE recipient = ? COLLATE NOCASE
         UNION ALL
         SELECT recipient FROM recipient_suppressions WHERE recipient = ? COLLATE NOCASE
+        UNION ALL
+        SELECT username AS recipient
+        FROM reddit_conversations
+        WHERE username = ? COLLATE NOCASE
       )
       LIMIT 1
     `);
+    this.tryClaimStmt = db.prepare(`
+      INSERT INTO recipient_suppressions (recipient, reason, created_at)
+      SELECT ?, 'pending-send', ?
+      WHERE NOT EXISTS (
+        SELECT 1 FROM sent_messages WHERE recipient = ? COLLATE NOCASE
+      )
+        AND NOT EXISTS (
+          SELECT 1 FROM reddit_conversations
+          WHERE username = ? COLLATE NOCASE
+        )
+      ON CONFLICT(recipient) DO NOTHING
+    `);
+    this.commitClaimStmt = db.prepare(`
+      UPDATE recipient_suppressions SET reason = 'sent'
+      WHERE recipient = ? COLLATE NOCASE AND reason = 'pending-send'
+    `);
+    this.releaseClaimStmt = db.prepare(`
+      DELETE FROM recipient_suppressions
+      WHERE recipient = ? COLLATE NOCASE AND reason = 'pending-send'
+    `);
   }
 
   hasSentTo(recipient: string): boolean {
-    return this.hasSentToStmt.get(recipient, recipient) !== undefined;
+    return this.hasSentToStmt.get(recipient, recipient, recipient) !== undefined;
+  }
+
+  tryClaim(recipient: string, now: number = Date.now()): boolean {
+    return this.tryClaimStmt.run(recipient, now, recipient, recipient).changes === 1;
+  }
+
+  commitClaim(recipient: string): void {
+    this.commitClaimStmt.run(recipient);
+  }
+
+  releaseClaim(recipient: string): void {
+    this.releaseClaimStmt.run(recipient);
   }
 }
 
@@ -1022,6 +1061,31 @@ export class ConversationStateRepository {
   }
 }
 
+/** Durable idempotency claims for AI replies to individual inbound events. */
+export class ConversationAutomationClaimRepository {
+  constructor(
+    private readonly db: Database.Database,
+    private readonly accountId: string = LEGACY_UNASSIGNED_ACCOUNT_ID,
+  ) {}
+
+  tryClaim(conversationId: string, incomingKey: string, now: number = Date.now()): boolean {
+    return this.db.prepare(`
+      INSERT INTO conversation_automation_claims (
+        account_id, conversation_id, incoming_key, status, claimed_at
+      ) VALUES (?, ?, ?, 'pending', ?)
+      ON CONFLICT(account_id, conversation_id, incoming_key) DO NOTHING
+    `).run(this.accountId, conversationId, incomingKey, now).changes === 1;
+  }
+
+  markSent(conversationId: string, incomingKey: string, now: number = Date.now()): boolean {
+    return this.db.prepare(`
+      UPDATE conversation_automation_claims
+      SET status = 'sent', sent_at = ?
+      WHERE account_id = ? AND conversation_id = ? AND incoming_key = ? AND status = 'pending'
+    `).run(now, this.accountId, conversationId, incomingKey).changes === 1;
+  }
+}
+
 /** Outcome labels are recorded independently from a raw inbox response. */
 export class ConversationOutcomesRepository {
   constructor(

src/main-loop.ts

@@ -349,6 +349,7 @@ export async function runMainLoop(app: AppContainer): Promise<void> {
                   selectionReason: processed.selectionReason,
                   sourceType: processed.sourceType,
                 });
+                recipientDeduplicator.commit(post.author);
 
                 const currentCount = currentHourlySendCount(rateLimiter);
                 logger.info(

src/services/conversation-automation.ts

@@ -1,5 +1,5 @@
 import type { OpenRouterClient } from '../ai/openrouter.js';
-import type { ConversationStateRepository } from '../database/repositories.js';
+import type { ConversationStateRepository, SentMessageRecord } from '../database/repositories.js';
 import type { Logger } from '../utils/logger.js';
 import type { RateLimiter } from '../utils/rate-limiter.js';
 import { findSellerSignal } from '../reddit/api.js';
@@ -28,6 +28,15 @@ export interface ConversationAutomationResult {
   remainingTurns: number;
 }
 
+export interface ConversationOpenerHistory {
+  getByChatRoomId(chatRoomId: string): SentMessageRecord | undefined;
+}
+
+export interface ConversationAutomationClaims {
+  tryClaim(conversationId: string, incomingKey: string): boolean;
+  markSent(conversationId: string, incomingKey: string): boolean;
+}
+
 /**
  * Sends at most one reply for each newly received user message. Durable state
  * makes retries and restarts idempotent, and every conversation stops after
@@ -38,6 +47,8 @@ export class ConversationAutomationService {
     private readonly chat: ChatService,
     private readonly ai: OpenRouterClient,
     private readonly states: ConversationStateRepository,
+    private readonly openers: ConversationOpenerHistory,
+    private readonly claims: ConversationAutomationClaims,
     private readonly rateLimiter: RateLimiter,
     private readonly logger: Logger,
     private readonly dryRun = false,
@@ -66,6 +77,13 @@ export class ConversationAutomationService {
     if (!latest || latest.isSelf) return null;
 
     const conversationId = conversation.roomId;
+    const opener = this.openers.getByChatRoomId(conversationId);
+    if (!opener || latest.timestamp <= opener.sentAt) {
+      this.logger.debug?.(
+        `[automation] Ignoring unverified inbound event for u/${conversation.username}: no confirmed post-opener response in this room`,
+      );
+      return null;
+    }
     const mapped = messages.map(({ sender, text, isSelf }) => ({ sender, text, isSelf }));
     const state = mergeConversationState(
       this.states.get(conversationId),
@@ -106,13 +124,11 @@ export class ConversationAutomationService {
       return null;
     }
 
-    const sent = await this.rateLimiter.throttle(
+    if (!this.claims.tryClaim(conversationId, incomingKey)) return null;
-      () => this.chat.sendChatMessage(conversation.url!, reply),
-      (success) => success,
-      signal,
-    );
-    if (!sent) return null;
 
+    // Persist the full intended state before the remote send. If the process
+    // dies after Matrix accepts the message, both this state and the durable
+    // event claim make the retry fail closed instead of sending twice.
     state.askedForMedia = true;
     state.automation.turnsUsed += 1;
     state.automation.lastIncomingKey = incomingKey;
@@ -123,6 +139,14 @@ export class ConversationAutomationService {
       state.automation.handoffReason = 'Three automated turns have been used.';
     }
     this.states.save(conversationId, state);
+
+    const sent = await this.rateLimiter.throttle(
+      () => this.chat.sendChatMessage(conversation.url!, reply),
+      (success) => success,
+      signal,
+    );
+    if (!sent) return null;
+    this.claims.markSent(conversationId, incomingKey);
     this.logger.info(
       `[automation] Sent turn ${state.automation.turnsUsed}/${MAX_AUTOMATED_CONVERSATION_TURNS} to u/${conversation.username}`,
     );

src/services/recipient-deduplicator.ts

@@ -1,5 +1,8 @@
 export interface RecipientHistory {
   hasSentTo(recipient: string): boolean;
+  tryClaim?(recipient: string): boolean;
+  commitClaim?(recipient: string): void;
+  releaseClaim?(recipient: string): void;
 }
 
 /**
@@ -24,13 +27,20 @@ export class RecipientDeduplicator {
     if (!key || this.reservations.has(key) || this.history.hasSentTo(recipient)) {
       return false;
     }
+    if (this.history.tryClaim && !this.history.tryClaim(recipient)) return false;
     this.reservations.add(key);
     return true;
   }
 
+  commit(recipient: string): void {
+    this.history.commitClaim?.(recipient);
+  }
+
   release(recipient: string): void {
     const key = normalizeRecipient(recipient);
-    if (key) this.reservations.delete(key);
+    if (!key) return;
+    this.reservations.delete(key);
+    this.history.releaseClaim?.(recipient);
   }
 }
 

src/services/reddit-account-runtime-builder.ts

@@ -7,6 +7,7 @@ import { BrowserClient } from '../browser/client.js';
 import {
   PostsRepository, MessagesRepository, SuccessfulMessagesRepository,
   DeliveryOutboxRepository, ConversationStateRepository,
+  ConversationAutomationClaimRepository,
   ConversationOutcomesRepository, RedditUsersRepository,
   ActivityLogRepository,
   RecipientHistoryRepository,
@@ -108,6 +109,7 @@ async function buildRuntime(
     const successfulMessagesRepo = new SuccessfulMessagesRepository(db, session.accountId);
     const deliveryOutboxRepo = new DeliveryOutboxRepository(db, session.accountId);
     const conversationStateRepo = new ConversationStateRepository(db, session.accountId);
+    const conversationAutomationClaimRepo = new ConversationAutomationClaimRepository(db, session.accountId);
     const conversationOutcomesRepo = new ConversationOutcomesRepository(db, session.accountId);
     const redditUsersRepo = new RedditUsersRepository(db);
     const accountContentRepo = new AccountContentRepository(db);
@@ -158,6 +160,8 @@ async function buildRuntime(
       chatService,
       aiClient,
       conversationStateRepo,
+      messagesRepo,
+      conversationAutomationClaimRepo,
       rateLimiter,
       logger,
       accountConfig.browser.dryRun,
@@ -331,6 +335,7 @@ class AccountRuntimeLoop {
             modelId: processed.modelId, selectionReason: processed.selectionReason,
             sourceType: processed.sourceType,
           });
+          this.recipientDeduplicator.commit(post.author);
           this.resources.activityLogRepo.insert({
             timestamp: Date.now(), type: 'sent', message: `Sent message to u/${post.author} (r/${post.subreddit})`,
             recipient: post.author, imageUrl: imageUrl ?? undefined, postTitle: post.title, postUrl: redditApi.getPostUrl(post),

tests/conversation-automation.test.ts

@@ -2,7 +2,11 @@ import { describe, expect, it, vi } from 'vitest';
 import { ConversationAutomationService } from '../src/services/conversation-automation.js';
 import type { ConversationState } from '../src/services/conversation-state.js';
 
-function harness(sellerKeywords: string[] = [], sellerDomains: string[] = []) {
+function harness(
+  sellerKeywords: string[] = [],
+  sellerDomains: string[] = [],
+  openerSentAt: number | null = 0,
+) {
   let saved: ConversationState | undefined;
   const chat = {
     refreshTopChats: vi.fn().mockResolvedValue(undefined),
@@ -14,15 +18,24 @@ function harness(sellerKeywords: string[] = [], sellerDomains: string[] = []) {
     get: vi.fn(() => saved),
     save: vi.fn((_id: string, state: ConversationState) => { saved = structuredClone(state); }),
   };
+  const openers = {
+    getByChatRoomId: vi.fn(() => openerSentAt === null
+      ? undefined
+      : { sentAt: openerSentAt }),
+  };
+  const claims = {
+    tryClaim: vi.fn(() => true),
+    markSent: vi.fn(() => true),
+  };
   const rateLimiter = {
     throttle: vi.fn(async (operation: () => Promise<boolean>) => operation()),
   };
   const logger = { info: vi.fn() };
   const service = new ConversationAutomationService(
-    chat as never, ai as never, states as never, rateLimiter as never, logger as never,
+    chat as never, ai as never, states as never, openers as never, claims, rateLimiter as never, logger as never,
     false, sellerKeywords, sellerDomains,
   );
-  return { service, chat, ai, states, getSaved: () => saved };
+  return { service, chat, ai, states, openers, claims, getSaved: () => saved };
 }
 
 function incoming(turn: number, text = `reply ${turn}`, media?: Array<{ originalUrl: string; localPath: string; fileName: string }>) {
@@ -98,4 +111,41 @@ describe('conversation automation', () => {
     expect(chat.sendChatMessage).toHaveBeenCalledTimes(1);
     expect(ai.generateMediaAsk).toHaveBeenCalledTimes(1);
   });
+
+  it('does not reply when the room has no recorded opener', async () => {
+    const { service, chat, ai, claims } = harness([], [], null);
+
+    expect(await service.processConversation(incoming(1))).toBeNull();
+    expect(ai.generateMediaAsk).not.toHaveBeenCalled();
+    expect(chat.sendChatMessage).not.toHaveBeenCalled();
+    expect(claims.tryClaim).not.toHaveBeenCalled();
+  });
+
+  it('does not reply to an inbound event older than the recorded opener', async () => {
+    const { service, chat, ai, claims } = harness([], [], 4);
+
+    expect(await service.processConversation(incoming(1))).toBeNull();
+    expect(ai.generateMediaAsk).not.toHaveBeenCalled();
+    expect(chat.sendChatMessage).not.toHaveBeenCalled();
+    expect(claims.tryClaim).not.toHaveBeenCalled();
+  });
+
+  it('claims an inbound event durably before sending and marks it sent afterwards', async () => {
+    const { service, chat, claims } = harness();
+
+    await service.processConversation(incoming(1));
+
+    expect(claims.tryClaim).toHaveBeenCalledWith('room-alice', 'in-1');
+    expect(claims.tryClaim.mock.invocationCallOrder[0])
+      .toBeLessThan(chat.sendChatMessage.mock.invocationCallOrder[0]);
+    expect(claims.markSent).toHaveBeenCalledWith('room-alice', 'in-1');
+  });
+
+  it('does not send when another worker already claimed the inbound event', async () => {
+    const { service, chat, claims } = harness();
+    claims.tryClaim.mockReturnValue(false);
+
+    expect(await service.processConversation(incoming(1))).toBeNull();
+    expect(chat.sendChatMessage).not.toHaveBeenCalled();
+  });
 });

tests/database.test.ts

@@ -107,7 +107,7 @@ describe('Database Client', () => {
         .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")
         .all() as { name: string }[];
       // Existing domain/config tables plus account, conversation, sync and media ownership stores.
-      expect(tables.length).toBe(14);
+      expect(tables.length).toBe(15);
     });
 
     it('tracks schema version and migrates a legacy sent_messages table transactionally', () => {
@@ -133,6 +133,38 @@ describe('Database Client', () => {
       });
     });
 
+    it('backfills global recipient suppression from sent messages and known conversations', () => {
+      runMigrations(db);
+      db.prepare(`
+        INSERT INTO reddit_conversations (
+          account_id, conversation_id, username, room_name, payload_json, updated_at
+        ) VALUES (?, 'old-room', 'Old_Contact', 'Old Contact', '{}', 20)
+      `).run('__legacy_unassigned__');
+      db.prepare(`
+        INSERT INTO processed_posts (account_id, id, subreddit, author, title, processed_at)
+        VALUES (?, 'old-post', 'test', 'sent-contact', 'title', 10)
+      `).run('__legacy_unassigned__');
+      db.prepare(`
+        INSERT INTO sent_messages (
+          account_id, post_id, recipient, subject, message, sent_at
+        ) VALUES (?, 'old-post', 'Sent_Contact', 'subject', 'message', 11)
+      `).run('__legacy_unassigned__');
+
+      const latestVersion = db.pragma('user_version', { simple: true }) as number;
+      db.exec(`
+        DELETE FROM recipient_suppressions;
+        DROP TABLE conversation_automation_claims;
+      `);
+      db.pragma(`user_version = ${latestVersion - 1}`);
+      runMigrations(db);
+
+      expect(db.prepare('SELECT recipient, reason FROM recipient_suppressions ORDER BY recipient COLLATE NOCASE').all())
+        .toEqual([
+          { recipient: 'Old_Contact', reason: 'historical-conversation' },
+          { recipient: 'Sent_Contact', reason: 'sent-message-history' },
+        ]);
+    });
+
     it('processed_posts has correct schema', () => {
       runMigrations(db);
       const columns = db.prepare('PRAGMA table_info(processed_posts)').all() as { name: string; notnull: number }[];

tests/repository-account-isolation.test.ts

@@ -5,6 +5,7 @@ import { runMigrations } from '../src/database/client.js';
 import {
   ActivityLogRepository,
   ConversationOutcomesRepository,
+  ConversationAutomationClaimRepository,
   ConversationStateRepository,
   DeliveryOutboxRepository,
   MessagesRepository,
@@ -101,6 +102,45 @@ describe('account-bound domain repositories', () => {
     expect(statesB.get('same-room')?.boundaries).toContain('does not want further sexual escalation');
   });
 
+  it('atomically claims each inbound automation event once per account', () => {
+    const claimsA = new ConversationAutomationClaimRepository(db, ACCOUNT_A);
+    const claimsB = new ConversationAutomationClaimRepository(db, ACCOUNT_B);
+
+    expect(claimsA.tryClaim('same-room', 'same-event', 10)).toBe(true);
+    expect(claimsA.tryClaim('same-room', 'same-event', 11)).toBe(false);
+    expect(claimsB.tryClaim('same-room', 'same-event', 12)).toBe(true);
+    expect(claimsA.markSent('same-room', 'same-event', 13)).toBe(true);
+    expect(claimsA.markSent('same-room', 'same-event', 14)).toBe(false);
+  });
+
+  it('claims recipients globally and blocks usernames with existing chat history', () => {
+    db.prepare(`
+      INSERT INTO reddit_conversations (
+        account_id, conversation_id, username, room_name, payload_json, updated_at
+      ) VALUES (?, ?, ?, ?, ?, ?)
+    `).run(
+      ACCOUNT_A,
+      'historical-room',
+      'Historical_User',
+      'Historical User',
+      JSON.stringify({
+        messages: [{ sender: 'Me', text: 'prior opener', timestamp: 1, isSelf: true }],
+      }),
+      2,
+    );
+    const history = new RecipientHistoryRepository(db);
+
+    expect(history.hasSentTo('historical_user')).toBe(true);
+    expect(history.tryClaim('HISTORICAL_USER', 3)).toBe(false);
+    expect(history.tryClaim('new_user', 4)).toBe(true);
+    expect(history.tryClaim('NEW_USER', 5)).toBe(false);
+    history.releaseClaim('new_user');
+    expect(history.tryClaim('new_user', 6)).toBe(true);
+    history.commitClaim('new_user');
+    history.releaseClaim('new_user');
+    expect(history.hasSentTo('NEW_USER')).toBe(true);
+  });
+
   it('isolates outbox claims, successful examples, activity and outcomes', () => {
     const postsA = new PostsRepository(db, ACCOUNT_A);
     const postsB = new PostsRepository(db, ACCOUNT_B);