Fix missed conversation replies

docs/data-flow.md

@@ -62,7 +62,7 @@ Do not add local retries around this layer without preserving the shared schedul
 
 ## Reply-first conversation automation
 
-Each account runtime refreshes recent chats and processes actionable unread conversations oldest-first before it can accept new outreach.
+Each account runtime refreshes recent chats and processes actionable inbound conversations oldest-first before it can accept new outreach. After the initial conservative startup scan, the persisted latest Matrix message event plus the worker's prior-scan watermark is authoritative: a newly observed inbound event newer than the confirmed opener remains actionable even if Matrix's presentation-only unread notification count has already returned to zero. On startup, read-marked rooms without a pending reply claim are not revived. Durable event claims, conversation policy state, and a later outbound event prevent duplicate replies.
 
 ```mermaid
 flowchart TD

src/services/conversation-automation.ts

@@ -88,14 +88,15 @@ export class ConversationAutomationService {
     private readonly pendingReplyRetryDelayMs = PENDING_REPLY_RETRY_DELAY_MS,
   ) {}
 
-  async processUnreadConversations(signal?: AbortSignal): Promise<ConversationAutomationResult[]> {
+  async processPendingConversations(signal?: AbortSignal): Promise<ConversationAutomationResult[]> {
     this.processing = true;
     let conversations: ActiveChat[] = [];
     try {
+      const previousScanAt = this.lastScanAt;
       await this.chat.refreshTopChats(REPLY_REFRESH_LIMIT);
       conversations = await this.chat.getCachedConversations();
       this.reconcilePendingClaims(conversations);
-      const actionable = this.actionableConversations(conversations);
+      const actionable = this.actionableConversations(conversations, previousScanAt);
       this.actionableBacklog = actionable.length;
       this.oldestPendingAt = actionable[0]?.messages?.at(-1)?.timestamp ?? null;
       this.lastScanAt = this.now();
@@ -144,10 +145,10 @@ export class ConversationAutomationService {
     conversation: ActiveChat,
     signal?: AbortSignal,
   ): Promise<ConversationAutomationResult | null> {
-    if (!conversation.unread || conversation.isFromUs || !conversation.roomId || !conversation.url) return null;
+    if (!conversation.roomId || !conversation.url) return null;
     const messages = conversation.messages ?? [];
-    const latest = messages.at(-1);
-    if (!latest || latest.isSelf) return null;
+    const latest = latestInboundMessage(messages);
+    if (!latest) return null;
 
     const conversationId = conversation.roomId;
     const opener = this.openers.getByChatRoomId(conversationId);
@@ -322,9 +323,12 @@ export class ConversationAutomationService {
     );
   }
 
-  private actionableConversations(conversations: ActiveChat[]): ActiveChat[] {
+  private actionableConversations(
+    conversations: ActiveChat[],
+    observedAfter: number | null = this.lastScanAt,
+  ): ActiveChat[] {
     return conversations
-      .filter((conversation) => this.isActionableConversation(conversation))
+      .filter((conversation) => this.isActionableConversation(conversation, observedAfter))
       .sort((left, right) => {
         const leftAt = left.messages?.at(-1)?.timestamp ?? left.lastMessageTime;
         const rightAt = right.messages?.at(-1)?.timestamp ?? right.lastMessageTime;
@@ -346,20 +350,26 @@ export class ConversationAutomationService {
     }
   }
 
-  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;
+  private isActionableConversation(conversation: ActiveChat, observedAfter: number | null): boolean {
+    if (!conversation.roomId || !conversation.url) return false;
+    const latest = latestInboundMessage(conversation.messages ?? []);
+    if (!latest) 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;
+    // On the first scan after startup, retain Matrix unread as the conservative
+    // bootstrap signal so a deployment cannot revive historical read replies.
+    // Later scans can recover fresh inbound events even if that flag cleared.
+    if (!conversation.unread && (observedAfter === null || latest.timestamp <= observedAfter)) return false;
     const state = this.states.get(conversation.roomId);
-    if (!state) return true;
+    if (state) {
       return state.automation.status === 'active' && state.automation.lastIncomingKey !== incomingKey;
     }
+    return true;
+  }
 
   private advanceStateForReply(
     state: ConversationState,
@@ -454,6 +464,17 @@ export class ConversationAutomationService {
   }
 }
 
+/**
+ * Matrix unread counts are presentation state and can return to zero before
+ * the automation worker observes an inbound event. The durable message list,
+ * opener timestamp, scan watermark, conversation state, and event claim decide
+ * whether work remains; only a latest inbound event can enter that flow.
+ */
+function latestInboundMessage(messages: ChatMessage[]): ChatMessage | null {
+  const latest = messages.at(-1);
+  return latest && !latest.isSelf ? latest : null;
+}
+
 function hasReceivedImage(messages: ChatMessage[]): boolean {
   return messages.some((message) => !message.isSelf && (message.media?.length ?? 0) > 0);
 }

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

@@ -286,7 +286,7 @@ class AccountRuntimeLoop {
         }
         if (Date.now() >= nextReplyCheckAt) {
           nextReplyCheckAt = Date.now() + replyInterval;
-          const automationResults = await this.resources.conversationAutomation.processUnreadConversations(signal);
+          const automationResults = await this.resources.conversationAutomation.processPendingConversations(signal);
           for (const result of automationResults) {
             if (result.kind === 'sent') continue;
             this.resources.activityLogRepo.insert({

tests/conversation-automation.test.ts

@@ -179,7 +179,7 @@ describe('conversation automation', () => {
     const older = { ...incoming(1), username: 'older', roomId: 'room-older', url: '/room/older' };
     chat.getCachedConversations.mockResolvedValue([newer, older]);
 
-    await service.processUnreadConversations();
+    await service.processPendingConversations();
 
     expect(chat.refreshTopChats).toHaveBeenCalledWith(25);
     expect(chat.sendChatMessage.mock.calls.map(([url]) => url)).toEqual(['/room/older', '/room/newer']);
@@ -190,7 +190,7 @@ describe('conversation automation', () => {
     chat.getCachedConversations.mockResolvedValue([incoming(1)]);
 
     expect(await service.hasActionableReplies()).toBe(true);
-    await service.processUnreadConversations();
+    await service.processPendingConversations();
 
     expect(await service.hasActionableReplies()).toBe(false);
     expect(service.getPriorityStatus()).toMatchObject({
@@ -215,6 +215,35 @@ describe('conversation automation', () => {
     );
   });
 
+  it('processes a durable inbound event even when Matrix reports the room as read', async () => {
+    const { service, chat, claims } = harness();
+    const conversation = { ...incoming(501), unread: false };
+    chat.getCachedConversations.mockResolvedValue([]);
+    await service.processPendingConversations();
+    chat.getCachedConversations.mockResolvedValue([conversation]);
+
+    expect(await service.hasActionableReplies()).toBe(true);
+    await service.processPendingConversations();
+
+    expect(claims.tryClaim).toHaveBeenCalledWith(
+      'room-alice',
+      'in-501',
+      'Would you be comfortable sharing a selfie?',
+      1_003,
+    );
+    expect(chat.sendChatMessage).toHaveBeenCalledOnce();
+  });
+
+  it('does not revive a historical read conversation on the first scan after startup', async () => {
+    const { service, chat, claims } = harness();
+    chat.getCachedConversations.mockResolvedValue([{ ...incoming(1), unread: false }]);
+
+    await service.processPendingConversations();
+
+    expect(claims.tryClaim).not.toHaveBeenCalled();
+    expect(chat.sendChatMessage).not.toHaveBeenCalled();
+  });
+
   it('retries the same durable reply after an interrupted send', async () => {
     const { service, chat, ai, claims } = harness();
     chat.sendChatMessage.mockRejectedValueOnce(new Error('interrupted'));
@@ -269,7 +298,7 @@ describe('conversation automation', () => {
     claims.listPending.mockReturnValue([pending] as never);
     chat.getCachedConversations.mockResolvedValue([conversation]);
 
-    await service.processUnreadConversations();
+    await service.processPendingConversations();
 
     expect(claims.markSent).toHaveBeenCalledWith('room-alice', 'in-1');
     expect(chat.sendChatMessage).not.toHaveBeenCalled();