Add per-subreddit R4R conversation modes

src/services/conversation-automation.ts

@@ -1,4 +1,5 @@
 import type { OpenRouterClient } from '../ai/openrouter.js';
+import type { SubredditConfig } from '../config/index.js';
 import type { ConversationStateRepository, SentMessageRecord } from '../database/repositories.js';
 import type { Logger } from '../utils/logger.js';
 import type { RateLimiter } from '../utils/rate-limiter.js';
@@ -9,7 +10,9 @@ import {
   buildConversationState,
   chooseConversationIntent,
   conversationMessageFingerprint,
-  MAX_AUTOMATED_CONVERSATION_TURNS,
+  createConversationAutomationState,
+  isSafeR4RReply,
+  LEGACY_AUTOMATED_CONVERSATION_TURNS,
   mergeConversationState,
   remainingAutomationTurns,
   type ConversationState,
@@ -39,8 +42,8 @@ export interface ConversationAutomationClaims {
 
 /**
  * Sends at most one reply for each newly received user message. Durable state
- * makes retries and restarts idempotent, and every conversation stops after
- * three automated turns or immediately when consent is withdrawn.
+ * makes retries and restarts idempotent. The opener subreddit selects a
+ * snapshotted legacy, autonomous R4R, or manual policy for the conversation.
  */
 export class ConversationAutomationService {
   constructor(
@@ -54,6 +57,7 @@ export class ConversationAutomationService {
     private readonly dryRun = false,
     private readonly sellerKeywords: readonly string[] = [],
     private readonly sellerDomains: readonly string[] = [],
+    private readonly configuredSubreddits: () => readonly SubredditConfig[] = () => [],
   ) {}
 
   async processUnreadConversations(signal?: AbortSignal): Promise<ConversationAutomationResult[]> {
@@ -85,14 +89,36 @@ export class ConversationAutomationService {
       return null;
     }
     const mapped = messages.map(({ sender, text, isSelf }) => ({ sender, text, isSelf }));
+    const savedState = this.states.get(conversationId);
     const state = mergeConversationState(
-      this.states.get(conversationId),
+      savedState,
       buildConversationState(mapped),
     );
+    if (!savedState) {
+      const behaviour = this.resolveConversationBehaviour(opener.subreddit);
+      const maxTurns = behaviour.mode === 'r4r'
+        ? behaviour.maxAutomatedTurns
+        : LEGACY_AUTOMATED_CONVERSATION_TURNS;
+      state.automation = createConversationAutomationState(behaviour.mode, maxTurns);
+      // Snapshot before generation so retries, restarts, dry runs, and later
+      // configuration edits cannot change the policy mid-conversation.
+      this.states.save(conversationId, state);
+    }
     const incomingKey = messageKey(latest);
     if (state.automation.lastIncomingKey === incomingKey) return null;
     if (state.automation.status !== 'active') return null;
 
+    if (state.automation.policyMode === 'manual') {
+      return this.finish(
+        conversation,
+        conversationId,
+        state,
+        incomingKey,
+        'human-required',
+        'This subreddit is configured for manual conversation handling.',
+      );
+    }
+
     const sellerSignal = findSellerSignal(latest.text, [...this.sellerKeywords], [...this.sellerDomains]);
     if (sellerSignal) {
       return this.finish(
@@ -110,15 +136,41 @@ export class ConversationAutomationService {
     if (state.boundaries.length > 0) {
       return this.finish(conversation, conversationId, state, incomingKey, 'boundary', state.boundaries.join('; '));
     }
+    if (state.automation.policyMode === 'r4r') {
+      const safetyReason = r4rSafetyHandoffReason(latest.text);
+      if (safetyReason) {
+        return this.finish(conversation, conversationId, state, incomingKey, 'human-required', safetyReason);
+      }
+    }
     if (remainingAutomationTurns(state) === 0) {
-      return this.finish(conversation, conversationId, state, incomingKey, 'human-required', 'Three automated turns have been used.');
+      return this.finish(
+        conversation,
+        conversationId,
+        state,
+        incomingKey,
+        'human-required',
+        `${state.automation.maxAutomatedTurns} automated turns have been used.`,
+      );
     }
 
-    const intent = chooseConversationIntent(mapped, state, 'media-ask');
-    if (intent.kind !== 'media-ask') {
+    const isLegacy = state.automation.policyMode === 'legacy';
+    const intent = chooseConversationIntent(mapped, state, isLegacy ? 'media-ask' : undefined);
+    if (isLegacy && intent.kind !== 'media-ask') {
       return this.finish(conversation, conversationId, state, incomingKey, 'human-required', intent.reason);
     }
-    const reply = await this.ai.generateMediaAsk(mapped, conversation.username, state, intent);
+    const reply = isLegacy
+      ? await this.ai.generateMediaAsk(mapped, conversation.username, state, intent)
+      : await this.ai.generateReply(mapped, conversation.username, state, intent);
+    if (!isLegacy && !isSafeR4RReply(reply)) {
+      return this.finish(
+        conversation,
+        conversationId,
+        state,
+        incomingKey,
+        'human-required',
+        'The generated R4R reply failed autonomous safety checks.',
+      );
+    }
     if (this.dryRun) {
       this.logger.info(`[automation] Dry run prepared turn ${state.automation.turnsUsed + 1} for u/${conversation.username}`);
       return null;
@@ -129,14 +181,14 @@ export class ConversationAutomationService {
     // 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;
+    if (isLegacy) state.askedForMedia = true;
     state.automation.turnsUsed += 1;
     state.automation.lastIncomingKey = incomingKey;
     state.automation.aiMessageFingerprints.push(conversationMessageFingerprint(reply));
-    const exhausted = state.automation.turnsUsed >= MAX_AUTOMATED_CONVERSATION_TURNS;
+    const exhausted = state.automation.turnsUsed >= state.automation.maxAutomatedTurns;
     if (exhausted) {
       state.automation.status = 'human-required';
-      state.automation.handoffReason = 'Three automated turns have been used.';
+      state.automation.handoffReason = `${state.automation.maxAutomatedTurns} automated turns have been used.`;
     }
     this.states.save(conversationId, state);
 
@@ -148,7 +200,7 @@ export class ConversationAutomationService {
     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}`,
+      `[automation] Sent ${state.automation.policyMode} turn ${state.automation.turnsUsed}/${state.automation.maxAutomatedTurns} to u/${conversation.username}`,
     );
     return {
       conversationId,
@@ -158,6 +210,13 @@ export class ConversationAutomationService {
     };
   }
 
+  private resolveConversationBehaviour(subreddit: string | undefined): SubredditConfig['conversation'] {
+    const configured = subreddit
+      ? this.configuredSubreddits().find((entry) => entry.name.toLowerCase() === subreddit.toLowerCase())
+      : undefined;
+    return configured?.conversation ?? { mode: 'legacy', maxAutomatedTurns: LEGACY_AUTOMATED_CONVERSATION_TURNS };
+  }
+
   private finish(
     conversation: ActiveChat,
     conversationId: string,
@@ -186,3 +245,15 @@ function hasReceivedImage(messages: ChatMessage[]): boolean {
 function messageKey(message: ChatMessage): string {
   return message.eventId ?? `${message.timestamp}:${message.sender}:${message.text}`;
 }
+
+function r4rSafetyHandoffReason(message: string): string | null {
+  if (/\b(?:i(?:'m| am)|im|age(?:d)?|turning)\s*(?:1[0-7]|[0-9])\b/i.test(message)
+    || /\b(?:1[0-7]|[0-9])\s*[fm]\b/i.test(message)
+    || /\b(?:minor|underage)\b/i.test(message)) {
+    return 'The participant may be under 18.';
+  }
+  if (/\b(?:meet|meetup|meet up|come over|address|phone number|whatsapp|telegram|snapchat)\b/i.test(message)) {
+    return 'The conversation reached contact details or an in-person meeting.';
+  }
+  return null;
+}