Prioritize conversation replies

src/services/conversation-automation.ts

@@ -6,6 +6,7 @@ import type { RateLimiter } from '../utils/rate-limiter.js';
 import { findSellerSignal } from '../reddit/api.js';
 import type { ActiveChat, ChatMessage } from './chat/types.js';
 import type { ChatService } from './chat/service.js';
+import type { ConversationAutomationClaimRecord } from '../database/repositories.js';
 import {
   buildConversationState,
   chooseConversationIntent,
@@ -37,16 +38,40 @@ export interface ConversationOpenerHistory {
 }
 
 export interface ConversationAutomationClaims {
-  tryClaim(conversationId: string, incomingKey: string): boolean;
+  tryClaim(conversationId: string, incomingKey: string, replyText?: string, incomingAt?: number): boolean;
+  getClaim?(conversationId: string, incomingKey: string): ConversationAutomationClaimRecord | undefined;
+  listPending?(): ConversationAutomationClaimRecord[];
+  markAttempt?(conversationId: string, incomingKey: string): boolean;
+  releasePending?(conversationId: string, incomingKey: string): boolean;
   markSent(conversationId: string, incomingKey: string): boolean;
 }
 
+export interface ReplyPriorityStatus {
+  processing: boolean;
+  actionableBacklog: number;
+  oldestPendingAt: number | null;
+  lastScanAt: number | null;
+  lastReplySentAt: number | null;
+  lastReplyLatencyMs: number | null;
+}
+
+const REPLY_REFRESH_LIMIT = 25;
+const REPLY_SEND_PRIORITY = 100;
+const PENDING_REPLY_RETRY_DELAY_MS = 2 * 60_000;
+
 /**
  * Sends at most one reply for each newly received user message. Durable state
  * makes retries and restarts idempotent. The opener subreddit selects a
  * snapshotted legacy, autonomous R4R, hybrid R4R-to-media, or manual policy.
  */
 export class ConversationAutomationService {
+  private processing = false;
+  private actionableBacklog = 0;
+  private oldestPendingAt: number | null = null;
+  private lastScanAt: number | null = null;
+  private lastReplySentAt: number | null = null;
+  private lastReplyLatencyMs: number | null = null;
+
   constructor(
     private readonly chat: ChatService,
     private readonly ai: OpenRouterClient,
@@ -59,17 +84,60 @@ export class ConversationAutomationService {
     private readonly sellerKeywords: readonly string[] = [],
     private readonly sellerDomains: readonly string[] = [],
     private readonly configuredSubreddits: () => readonly SubredditConfig[] = () => [],
+    private readonly now: () => number = Date.now,
+    private readonly pendingReplyRetryDelayMs = PENDING_REPLY_RETRY_DELAY_MS,
   ) {}
 
   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);
+    this.processing = true;
+    let conversations: ActiveChat[] = [];
+    try {
+      await this.chat.refreshTopChats(REPLY_REFRESH_LIMIT);
+      conversations = await this.chat.getCachedConversations();
+      this.reconcilePendingClaims(conversations);
+      const actionable = this.actionableConversations(conversations);
+      this.actionableBacklog = actionable.length;
+      this.oldestPendingAt = actionable[0]?.messages?.at(-1)?.timestamp ?? null;
+      this.lastScanAt = this.now();
+      if (actionable.length > 0) {
+        const oldestAgeSeconds = this.oldestPendingAt === null
+          ? 0
+          : Math.max(0, Math.round((this.now() - this.oldestPendingAt) / 1_000));
+        this.logger.info(
+          `[automation] Reply-first queue has ${actionable.length} actionable conversation(s); oldest ${oldestAgeSeconds}s`,
+        );
+      }
+
+      const results: ConversationAutomationResult[] = [];
+      for (const conversation of actionable) {
+        const result = await this.processConversation(conversation, signal);
+        this.actionableBacklog = Math.max(0, this.actionableBacklog - 1);
+        this.oldestPendingAt = actionable[actionable.length - this.actionableBacklog]?.messages?.at(-1)?.timestamp ?? null;
+        if (result) results.push(result);
+      }
+      return results;
+    } finally {
+      this.processing = false;
+      const remaining = this.actionableConversations(conversations);
+      this.actionableBacklog = remaining.length;
+      this.oldestPendingAt = remaining[0]?.messages?.at(-1)?.timestamp ?? null;
     }
-    return results;
+  }
+
+  async hasActionableReplies(): Promise<boolean> {
+    if (this.processing || this.actionableBacklog > 0) return true;
+    return this.actionableConversations(await this.chat.getCachedConversations()).length > 0;
+  }
+
+  getPriorityStatus(): ReplyPriorityStatus {
+    return {
+      processing: this.processing,
+      actionableBacklog: this.actionableBacklog,
+      oldestPendingAt: this.oldestPendingAt,
+      lastScanAt: this.lastScanAt,
+      lastReplySentAt: this.lastReplySentAt,
+      lastReplyLatencyMs: this.lastReplyLatencyMs,
+    };
   }
 
   async processConversation(
@@ -110,6 +178,46 @@ export class ConversationAutomationService {
       this.states.save(conversationId, state);
     }
     const incomingKey = messageKey(latest);
+    const existingClaim = this.claims.getClaim?.(conversationId, incomingKey);
+    if (existingClaim?.status === 'pending') {
+      const observedOutbound = messages.some((message) => message.isSelf && message.timestamp > latest.timestamp);
+      if (observedOutbound) {
+        this.claims.markSent(conversationId, incomingKey);
+        this.logger.info(`[automation] Reconciled interrupted reply for u/${conversation.username} from Matrix history`);
+        return null;
+      }
+      if (!existingClaim.replyText) {
+        this.claims.releasePending?.(conversationId, incomingKey);
+        return this.finish(
+          conversation,
+          conversationId,
+          state,
+          incomingKey,
+          'human-required',
+          'An interrupted reply has no recoverable message body.',
+        );
+      }
+      if (this.now() - existingClaim.claimedAt < this.pendingReplyRetryDelayMs) return null;
+      const retryPlan = replyPlan(state);
+      const exhausted = this.advanceStateForReply(
+        state,
+        conversationId,
+        incomingKey,
+        existingClaim.replyText,
+        retryPlan.isLegacy,
+        retryPlan.isHybridMediaAsk,
+      );
+      return this.sendClaimedReply(
+        conversation,
+        state,
+        incomingKey,
+        existingClaim.replyText,
+        latest.timestamp,
+        exhausted,
+        signal,
+      );
+    }
+    if (existingClaim?.status === 'sent') return null;
     if (state.automation.lastIncomingKey === incomingKey) return null;
     if (state.automation.status !== 'active') return null;
 
@@ -159,11 +267,7 @@ export class ConversationAutomationService {
       );
     }
 
-    const isLegacy = state.automation.policyMode === 'legacy';
-    const isHybrid = state.automation.policyMode === 'r4r-media';
-    const isHybridMediaAsk = isHybrid
-      && !state.askedForMedia
-      && state.automation.turnsUsed >= state.automation.rapportTurnsBeforeMediaAsk;
+    const { isLegacy, isHybridMediaAsk } = replyPlan(state);
     if (isHybridMediaAsk) {
       state.automation.hybridPhase = 'media-ask';
       this.states.save(conversationId, state);
@@ -194,16 +298,84 @@ export class ConversationAutomationService {
       return null;
     }
 
-    if (!this.claims.tryClaim(conversationId, incomingKey)) return null;
+    if (!this.claims.tryClaim(conversationId, incomingKey, reply, latest.timestamp)) 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.
-    if (isLegacy || isHybridMediaAsk) state.askedForMedia = true;
-    if (isHybridMediaAsk) state.automation.hybridPhase = 'post-media-ask';
-    state.automation.turnsUsed += 1;
-    state.automation.lastIncomingKey = incomingKey;
-    state.automation.aiMessageFingerprints.push(conversationMessageFingerprint(reply));
+    const exhausted = this.advanceStateForReply(
+      state,
+      conversationId,
+      incomingKey,
+      reply,
+      isLegacy,
+      isHybridMediaAsk,
+    );
+    return this.sendClaimedReply(
+      conversation,
+      state,
+      incomingKey,
+      reply,
+      latest.timestamp,
+      exhausted,
+      signal,
+    );
+  }
+
+  private actionableConversations(conversations: ActiveChat[]): ActiveChat[] {
+    return conversations
+      .filter((conversation) => this.isActionableConversation(conversation))
+      .sort((left, right) => {
+        const leftAt = left.messages?.at(-1)?.timestamp ?? left.lastMessageTime;
+        const rightAt = right.messages?.at(-1)?.timestamp ?? right.lastMessageTime;
+        return leftAt - rightAt;
+      });
+  }
+
+  private reconcilePendingClaims(conversations: ActiveChat[]): void {
+    for (const claim of this.claims.listPending?.() ?? []) {
+      const conversation = conversations.find((item) => item.roomId === claim.conversationId);
+      if (!conversation) continue;
+      const messages = conversation.messages ?? [];
+      const incomingIndex = messages.findIndex((message) => messageKey(message) === claim.incomingKey);
+      if (incomingIndex < 0) continue;
+      if (!messages.slice(incomingIndex + 1).some((message) => message.isSelf)) continue;
+      if (this.claims.markSent(claim.conversationId, claim.incomingKey)) {
+        this.logger.info(`[automation] Reconciled interrupted reply for u/${conversation.username} from Matrix history`);
+      }
+    }
+  }
+
+  private isActionableConversation(conversation: ActiveChat): boolean {
+    if (!conversation.unread || conversation.isFromUs || !conversation.roomId || !conversation.url) return false;
+    const latest = conversation.messages?.at(-1);
+    if (!latest || latest.isSelf) return false;
+    const opener = this.openers.getByChatRoomId(conversation.roomId);
+    if (!opener || latest.timestamp <= opener.sentAt) return false;
+    const incomingKey = messageKey(latest);
+    const claim = this.claims.getClaim?.(conversation.roomId, incomingKey);
+    if (claim?.status === 'pending') return true;
+    if (claim?.status === 'sent') return false;
+    const state = this.states.get(conversation.roomId);
+    if (!state) return true;
+    return state.automation.status === 'active' && state.automation.lastIncomingKey !== incomingKey;
+  }
+
+  private advanceStateForReply(
+    state: ConversationState,
+    conversationId: string,
+    incomingKey: string,
+    reply: string,
+    isLegacy: boolean,
+    isHybridMediaAsk: boolean,
+  ): boolean {
+    if (state.automation.lastIncomingKey !== incomingKey) {
+      if (isLegacy || isHybridMediaAsk) state.askedForMedia = true;
+      if (isHybridMediaAsk) state.automation.hybridPhase = 'post-media-ask';
+      state.automation.turnsUsed += 1;
+      state.automation.lastIncomingKey = incomingKey;
+      state.automation.aiMessageFingerprints.push(conversationMessageFingerprint(reply));
+    }
     // Keep the hybrid conversation active after its one media ask even when
     // that send consumes the final turn. The next inbound event still needs
     // to be classified as an image, a decline, or a normal turn-limit handoff.
@@ -214,19 +386,36 @@ export class ConversationAutomationService {
       state.automation.handoffReason = `${state.automation.maxAutomatedTurns} automated turns have been used.`;
     }
     this.states.save(conversationId, state);
+    return exhausted;
+  }
 
+  private async sendClaimedReply(
+    conversation: ActiveChat,
+    state: ConversationState,
+    incomingKey: string,
+    reply: string,
+    incomingAt: number,
+    exhausted: boolean,
+    signal?: AbortSignal,
+  ): Promise<ConversationAutomationResult | null> {
+    this.claims.markAttempt?.(conversation.roomId!, incomingKey);
     const sent = await this.rateLimiter.throttle(
       () => this.chat.sendChatMessage(conversation.url!, reply),
       (success) => success,
       signal,
+      REPLY_SEND_PRIORITY,
     );
     if (!sent) return null;
-    this.claims.markSent(conversationId, incomingKey);
+    const sentAt = this.now();
+    this.claims.markSent(conversation.roomId!, incomingKey);
+    this.lastReplySentAt = sentAt;
+    this.lastReplyLatencyMs = Math.max(0, sentAt - incomingAt);
     this.logger.info(
-      `[automation] Sent ${state.automation.policyMode} turn ${state.automation.turnsUsed}/${state.automation.maxAutomatedTurns} to u/${conversation.username}`,
+      `[automation] Sent ${state.automation.policyMode} turn ${state.automation.turnsUsed}/${state.automation.maxAutomatedTurns} `
+      + `to u/${conversation.username}; reply latency ${Math.round(this.lastReplyLatencyMs / 1_000)}s`,
     );
     return {
-      conversationId,
+      conversationId: conversation.roomId!,
       username: conversation.username,
       kind: exhausted ? 'human-required' : 'sent',
       remainingTurns: remainingAutomationTurns(state),
@@ -273,6 +462,14 @@ function messageKey(message: ChatMessage): string {
   return message.eventId ?? `${message.timestamp}:${message.sender}:${message.text}`;
 }
 
+function replyPlan(state: ConversationState): { isLegacy: boolean; isHybridMediaAsk: boolean } {
+  const isLegacy = state.automation.policyMode === 'legacy';
+  const isHybridMediaAsk = state.automation.policyMode === 'r4r-media'
+    && !state.askedForMedia
+    && state.automation.turnsUsed >= state.automation.rapportTurnsBeforeMediaAsk;
+  return { isLegacy, isHybridMediaAsk };
+}
+
 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)