From dc3fc6373b81644350735c32b61fc8851559fb18 Mon Sep 17 00:00:00 2001 From: jjbrunton Date: Sun, 9 Aug 2026 16:56:25 +0100 Subject: [PATCH] Fix missed conversation replies --- docs/data-flow.md | 2 +- src/services/conversation-automation.ts | 47 ++++++++++++++----- .../reddit-account-runtime-builder.ts | 2 +- tests/conversation-automation.test.ts | 35 ++++++++++++-- 4 files changed, 68 insertions(+), 18 deletions(-) diff --git a/docs/data-flow.md b/docs/data-flow.md index d30da10..0a3db96 100644 --- a/docs/data-flow.md +++ b/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 diff --git a/src/services/conversation-automation.ts b/src/services/conversation-automation.ts index b1f6489..e431f0b 100644 --- a/src/services/conversation-automation.ts +++ b/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 { + async processPendingConversations(signal?: AbortSignal): Promise { 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 { - 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,19 +350,25 @@ 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; - return state.automation.status === 'active' && state.automation.lastIncomingKey !== incomingKey; + if (state) { + return state.automation.status === 'active' && state.automation.lastIncomingKey !== incomingKey; + } + return true; } private advanceStateForReply( @@ -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); } diff --git a/src/services/reddit-account-runtime-builder.ts b/src/services/reddit-account-runtime-builder.ts index 503a9a2..9a56d77 100644 --- a/src/services/reddit-account-runtime-builder.ts +++ b/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({ diff --git a/tests/conversation-automation.test.ts b/tests/conversation-automation.test.ts index faaf3a4..08cbb5a 100644 --- a/tests/conversation-automation.test.ts +++ b/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();