feat: automate bounded image-sharing conversations

src/services/conversation-automation.ts

@@ -0,0 +1,148 @@
+import type { OpenRouterClient } from '../ai/openrouter.js';
+import type { ConversationStateRepository } from '../database/repositories.js';
+import type { Logger } from '../utils/logger.js';
+import type { RateLimiter } from '../utils/rate-limiter.js';
+import type { ActiveChat, ChatMessage } from './chat/types.js';
+import type { ChatService } from './chat/service.js';
+import {
+  buildConversationState,
+  chooseConversationIntent,
+  MAX_AUTOMATED_CONVERSATION_TURNS,
+  mergeConversationState,
+  remainingAutomationTurns,
+  type ConversationState,
+} from './conversation-state.js';
+
+export type ConversationAutomationResultKind =
+  | 'sent'
+  | 'human-required'
+  | 'boundary'
+  | 'image-received';
+
+export interface ConversationAutomationResult {
+  conversationId: string;
+  username: string;
+  kind: ConversationAutomationResultKind;
+  remainingTurns: number;
+}
+
+/**
+ * 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.
+ */
+export class ConversationAutomationService {
+  constructor(
+    private readonly chat: ChatService,
+    private readonly ai: OpenRouterClient,
+    private readonly states: ConversationStateRepository,
+    private readonly rateLimiter: RateLimiter,
+    private readonly logger: Logger,
+    private readonly dryRun = false,
+  ) {}
+
+  async processUnreadConversations(signal?: AbortSignal): Promise<ConversationAutomationResult[]> {
+    await this.chat.refreshTopChats(5);
+    const conversations = await this.chat.getCachedConversations();
+    const results: ConversationAutomationResult[] = [];
+    for (const conversation of conversations) {
+      const result = await this.processConversation(conversation, signal);
+      if (result) results.push(result);
+    }
+    return results;
+  }
+
+  async processConversation(
+    conversation: ActiveChat,
+    signal?: AbortSignal,
+  ): Promise<ConversationAutomationResult | null> {
+    if (!conversation.unread || conversation.isFromUs || !conversation.roomId || !conversation.url) return null;
+    const messages = conversation.messages ?? [];
+    const latest = messages.at(-1);
+    if (!latest || latest.isSelf) return null;
+
+    const conversationId = conversation.roomId;
+    const mapped = messages.map(({ sender, text, isSelf }) => ({ sender, text, isSelf }));
+    const state = mergeConversationState(
+      this.states.get(conversationId),
+      buildConversationState(mapped),
+    );
+    const incomingKey = messageKey(latest);
+    if (state.automation.lastIncomingKey === incomingKey) return null;
+    if (state.automation.status !== 'active') return null;
+
+    if (hasReceivedImage(messages)) {
+      return this.finish(conversation, conversationId, state, incomingKey, 'image-received', 'The user shared an image.');
+    }
+    if (state.boundaries.length > 0) {
+      return this.finish(conversation, conversationId, state, incomingKey, 'boundary', state.boundaries.join('; '));
+    }
+    if (remainingAutomationTurns(state) === 0) {
+      return this.finish(conversation, conversationId, state, incomingKey, 'human-required', 'Three automated turns have been used.');
+    }
+
+    const intent = chooseConversationIntent(mapped, state, 'media-ask');
+    if (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);
+    if (this.dryRun) {
+      this.logger.info(`[automation] Dry run prepared turn ${state.automation.turnsUsed + 1} for u/${conversation.username}`);
+      return null;
+    }
+
+    const sent = await this.rateLimiter.throttle(
+      () => this.chat.sendChatMessage(conversation.url!, reply),
+      (success) => success,
+      signal,
+    );
+    if (!sent) return null;
+
+    state.askedForMedia = true;
+    state.automation.turnsUsed += 1;
+    state.automation.lastIncomingKey = incomingKey;
+    const exhausted = state.automation.turnsUsed >= MAX_AUTOMATED_CONVERSATION_TURNS;
+    if (exhausted) {
+      state.automation.status = 'human-required';
+      state.automation.handoffReason = 'Three automated turns have been used.';
+    }
+    this.states.save(conversationId, state);
+    this.logger.info(
+      `[automation] Sent turn ${state.automation.turnsUsed}/${MAX_AUTOMATED_CONVERSATION_TURNS} to u/${conversation.username}`,
+    );
+    return {
+      conversationId,
+      username: conversation.username,
+      kind: exhausted ? 'human-required' : 'sent',
+      remainingTurns: remainingAutomationTurns(state),
+    };
+  }
+
+  private finish(
+    conversation: ActiveChat,
+    conversationId: string,
+    state: ConversationState,
+    incomingKey: string,
+    status: 'image-received' | 'human-required' | 'boundary',
+    reason: string,
+  ): ConversationAutomationResult {
+    state.automation.status = status;
+    state.automation.lastIncomingKey = incomingKey;
+    state.automation.handoffReason = reason;
+    this.states.save(conversationId, state);
+    return {
+      conversationId,
+      username: conversation.username,
+      kind: status,
+      remainingTurns: remainingAutomationTurns(state),
+    };
+  }
+}
+
+function hasReceivedImage(messages: ChatMessage[]): boolean {
+  return messages.some((message) => !message.isSelf && (message.media?.length ?? 0) > 0);
+}
+
+function messageKey(message: ChatMessage): string {
+  return message.eventId ?? `${message.timestamp}:${message.sender}:${message.text}`;
+}

src/services/conversation-state.ts

@@ -25,8 +25,40 @@ export interface ConversationPersona {
 export interface ConversationState {
   persona: ConversationPersona;
   boundaries: string[];
+  /** Kept for backwards compatibility with state written before automation. */
   askedForMedia: boolean;
   topics: string[];
+  automation: ConversationAutomationState;
+}
+
+export const MAX_AUTOMATED_CONVERSATION_TURNS = 3;
+
+export type ConversationAutomationStatus =
+  | 'active'
+  | 'image-received'
+  | 'human-required'
+  | 'boundary'
+  | 'user-overridden';
+
+export interface ConversationAutomationState {
+  turnsUsed: number;
+  status: ConversationAutomationStatus;
+  lastIncomingKey: string | null;
+  handoffReason: string | null;
+}
+
+export function createConversationAutomationState(): ConversationAutomationState {
+  return {
+    turnsUsed: 0,
+    status: 'active',
+    lastIncomingKey: null,
+    handoffReason: null,
+  };
+}
+
+export function remainingAutomationTurns(state: ConversationState): number {
+  if (state.automation.status !== 'active') return 0;
+  return Math.max(0, MAX_AUTOMATED_CONVERSATION_TURNS - state.automation.turnsUsed);
 }
 
 const EMPTY_PERSONA: ConversationPersona = {
@@ -67,6 +99,7 @@ export function buildConversationState(messages: ConversationMessage[]): Convers
     boundaries,
     askedForMedia: /\b(send|show|share)\b[^.\n]{0,40}\b(pic|pics|photo|photos|selfie|selfies|video|videos|nudes?)\b/i.test(ownText),
     topics,
+    automation: createConversationAutomationState(),
   };
 }
 
@@ -79,10 +112,13 @@ export function chooseConversationIntent(
     if (state.boundaries.includes('does not want to send photos or videos')) {
       return { kind: 'respect-boundary', reason: 'they have declined or set a boundary around photos or videos' };
     }
-    if (state.askedForMedia) {
-      return { kind: 'follow-up', reason: 'a media ask was already made, so do not repeat it' };
+    if (state.automation.status !== 'active' || remainingAutomationTurns(state) === 0) {
+      return { kind: 'follow-up', reason: 'the automated conversation has reached human handoff' };
     }
-    return { kind: 'media-ask', reason: 'a media ask was explicitly requested and no boundary is recorded' };
+    return {
+      kind: 'media-ask',
+      reason: `image-sharing turn ${state.automation.turnsUsed + 1} of ${MAX_AUTOMATED_CONVERSATION_TURNS}`,
+    };
   }
 
   if (state.boundaries.length > 0) {
@@ -116,6 +152,7 @@ export function mergeConversationState(
     boundaries: [...new Set([...saved.boundaries, ...observed.boundaries])],
     askedForMedia: saved.askedForMedia || observed.askedForMedia,
     topics: [...new Set([...saved.topics, ...observed.topics])],
+    automation: saved.automation,
   };
 }
 
@@ -152,6 +189,7 @@ export function parseConversationState(value: string | null | undefined): Conver
     if (!parsed.persona || !Array.isArray(parsed.boundaries) || !Array.isArray(parsed.topics) || typeof parsed.askedForMedia !== 'boolean') {
       return null;
     }
+    const automation = parseAutomationState(parsed.automation);
     return {
       persona: {
         age: typeof parsed.persona.age === 'number' ? parsed.persona.age : null,
@@ -163,12 +201,30 @@ export function parseConversationState(value: string | null | undefined): Conver
       boundaries: parsed.boundaries.filter((value): value is string => typeof value === 'string'),
       askedForMedia: parsed.askedForMedia,
       topics: parsed.topics.filter((value): value is string => typeof value === 'string'),
+      automation,
     };
   } catch {
     return null;
   }
 }
 
+function parseAutomationState(value: unknown): ConversationAutomationState {
+  if (!value || typeof value !== 'object') return createConversationAutomationState();
+  const parsed = value as Partial<ConversationAutomationState>;
+  const status: ConversationAutomationStatus =
+    parsed.status === 'image-received' || parsed.status === 'human-required' || parsed.status === 'boundary' || parsed.status === 'user-overridden'
+      ? parsed.status
+      : 'active';
+  return {
+    turnsUsed: typeof parsed.turnsUsed === 'number'
+      ? Math.max(0, Math.min(MAX_AUTOMATED_CONVERSATION_TURNS, Math.trunc(parsed.turnsUsed)))
+      : 0,
+    status,
+    lastIncomingKey: typeof parsed.lastIncomingKey === 'string' ? parsed.lastIncomingKey : null,
+    handoffReason: typeof parsed.handoffReason === 'string' ? parsed.handoffReason : null,
+  };
+}
+
 function extractTopics(text: string): string[] {
   const topics: string[] = [];
   for (const topic of ['music', 'films', 'movies', 'gaming', 'work', 'travel', 'fitness']) {

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

@@ -41,6 +41,7 @@ import {
 import { RecipientDeduplicator } from './recipient-deduplicator.js';
 import { ConfigStore } from '../config/persistence.js';
 import { createConfigBackedSubredditPollingControl } from './subreddit-polling-control.js';
+import { ConversationAutomationService } from './conversation-automation.js';
 
 export interface ConcreteRuntimeFactoryOptions {
   config: Config;
@@ -153,13 +154,21 @@ async function buildRuntime(
       sentMessageHistory: async (since) => messagesRepo.getSentTimestampsSince(since),
     });
     await rateLimiter.initialize();
+    const conversationAutomation = new ConversationAutomationService(
+      chatService,
+      aiClient,
+      conversationStateRepo,
+      rateLimiter,
+      logger,
+      accountConfig.browser.dryRun,
+    );
 
     const resources: RedditAccountRuntimeResources = {
       browserProfileDir, browserClient, accountAuth, matrixClient, chatCache, redditApi,
       postsRepo, messagesRepo, successfulMessagesRepo, deliveryOutboxRepo,
       conversationStateRepo, conversationOutcomesRepo, redditUsersRepo, accountContentRepo,
       activityLogRepo,
-      monitorService, processorService, responseChecker, chatService, mediaService,
+      monitorService, processorService, responseChecker, conversationAutomation, chatService, mediaService,
       mediaSyncService, rateLimiter,
     };
     const lifecycle = new AccountRuntimeLoop(
@@ -232,6 +241,20 @@ class AccountRuntimeLoop {
         if (Date.now() >= nextResponseCheckAt) {
           nextResponseCheckAt = Date.now() + responseInterval;
           await this.resources.responseChecker.checkForResponses();
+          const automationResults = await this.resources.conversationAutomation.processUnreadConversations(signal);
+          for (const result of automationResults) {
+            if (result.kind === 'sent') continue;
+            this.resources.activityLogRepo.insert({
+              timestamp: Date.now(),
+              type: 'response',
+              message: result.kind === 'human-required'
+                ? `Human follow-up required for u/${result.username}`
+                : result.kind === 'image-received'
+                  ? `Image received from u/${result.username}`
+                  : `Automation stopped for u/${result.username} after a boundary`,
+              recipient: result.username,
+            });
+          }
         }
         await delay(this.config.reddit.pollIntervalMs, signal);
       } catch (error) {

src/services/reddit-account-runtime.ts

@@ -30,6 +30,7 @@ import type { MonitorService } from './monitor.js';
 import type { ProcessorService } from './processor.js';
 import type { ResponseCheckerService } from './response-checker.js';
 import type { RedditPost } from '../reddit/api.js';
+import type { ConversationAutomationService } from './conversation-automation.js';
 
 /** Credentials passed to a factory. They must never be exposed by list/snapshot APIs. */
 export interface VerifiedAccountSession {
@@ -70,6 +71,7 @@ export interface RedditAccountRuntimeResources {
   monitorService: MonitorService;
   processorService: ProcessorService;
   responseChecker: ResponseCheckerService;
+  conversationAutomation: ConversationAutomationService;
   chatService: ChatService;
   mediaService: MediaService;
   mediaSyncService: MediaSyncService;