diff --git a/src/database/client.ts b/src/database/client.ts index 9bb8c9f..aeb5770 100644 --- a/src/database/client.ts +++ b/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; diff --git a/src/database/repositories.ts b/src/database/repositories.ts index c626cb2..7f9b228 100644 --- a/src/database/repositories.ts +++ b/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( diff --git a/src/main-loop.ts b/src/main-loop.ts index a722be9..4a67924 100644 --- a/src/main-loop.ts +++ b/src/main-loop.ts @@ -349,6 +349,7 @@ export async function runMainLoop(app: AppContainer): Promise { selectionReason: processed.selectionReason, sourceType: processed.sourceType, }); + recipientDeduplicator.commit(post.author); const currentCount = currentHourlySendCount(rateLimiter); logger.info( diff --git a/src/services/conversation-automation.ts b/src/services/conversation-automation.ts index 40cefe5..6a56072 100644 --- a/src/services/conversation-automation.ts +++ b/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( - () => this.chat.sendChatMessage(conversation.url!, reply), - (success) => success, - signal, - ); - if (!sent) return null; + if (!this.claims.tryClaim(conversationId, incomingKey)) 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}`, ); diff --git a/src/services/recipient-deduplicator.ts b/src/services/recipient-deduplicator.ts index db52a05..276bf0c 100644 --- a/src/services/recipient-deduplicator.ts +++ b/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); } } diff --git a/src/services/reddit-account-runtime-builder.ts b/src/services/reddit-account-runtime-builder.ts index 3ca7dc9..5046abb 100644 --- a/src/services/reddit-account-runtime-builder.ts +++ b/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), diff --git a/tests/conversation-automation.test.ts b/tests/conversation-automation.test.ts index 4d83df0..20b7f2f 100644 --- a/tests/conversation-automation.test.ts +++ b/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) => 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(); + }); }); diff --git a/tests/database.test.ts b/tests/database.test.ts index 1a86b87..55ae259 100644 --- a/tests/database.test.ts +++ b/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 }[]; diff --git a/tests/repository-account-isolation.test.ts b/tests/repository-account-isolation.test.ts index e92d340..2c17dfb 100644 --- a/tests/repository-account-isolation.test.ts +++ b/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);