diff --git a/docs/data-flow.md b/docs/data-flow.md index e4083ae..9d03fdf 100644 --- a/docs/data-flow.md +++ b/docs/data-flow.md @@ -64,6 +64,8 @@ Do not add local retries around this layer without preserving the shared schedul 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. +Outgoing Matrix events are attributed against the recorded opener, durable automated-reply fingerprints, and confirmed exchange-library event records. If synchronization observes an unmatched outgoing message, the conversation becomes operator-owned before another reply can be generated. An unmatched outgoing message also supersedes, rather than confirms, a pending automated claim; only the claimed reply text can reconcile that claim as sent. + ```mermaid flowchart TD IN["Unread inbound Matrix event"] --> VERIFIED{"Confirmed after a recorded opener?"} diff --git a/src/services/conversation-automation.ts b/src/services/conversation-automation.ts index 2d76f5c..71c56a8 100644 --- a/src/services/conversation-automation.ts +++ b/src/services/conversation-automation.ts @@ -189,6 +189,22 @@ export class ConversationAutomationService { this.states.save(conversationId, state); } const incomingKey = messageKey(latest); + const mediaAttribution = this.mediaExchange?.getOutboundAttribution?.(conversationId); + if (state.automation.status === 'active' + && hasUnattributedSelfMessage( + messages, + opener, + state.automation.aiMessageFingerprints, + mediaAttribution?.eventIds, + mediaAttribution?.texts, + )) { + state.automation.status = 'user-overridden'; + state.automation.handoffReason = 'A manual reply was observed in this conversation.'; + this.states.save(conversationId, state); + this.claims.markSuperseded?.(conversationId, incomingKey); + this.logger.info(`[automation] Paused u/${conversation.username}; an operator-authored message was observed`); + return null; + } const existingClaim = this.claims.getClaim?.(conversationId, incomingKey); if (existingClaim?.status === 'pending') { const observedOutbound = messages.some((message) => message.isSelf && message.timestamp > latest.timestamp); @@ -398,9 +414,17 @@ export class ConversationAutomationService { 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)) { + const laterSelfMessages = messages.slice(incomingIndex + 1).filter((message) => message.isSelf); + if (laterSelfMessages.length === 0) continue; + const expectedFingerprint = claim.replyText + ? conversationMessageFingerprint(claim.replyText) + : null; + const observedClaimedReply = expectedFingerprint !== null + && laterSelfMessages.some((message) => conversationMessageFingerprint(message.text) === expectedFingerprint); + if (observedClaimedReply && this.claims.markSent(claim.conversationId, claim.incomingKey)) { this.logger.info(`[automation] Reconciled interrupted reply for u/${conversation.username} from Matrix history`); + } else if (!observedClaimedReply && this.claims.markSuperseded?.(claim.conversationId, claim.incomingKey)) { + this.logger.info(`[automation] Superseded queued reply for u/${conversation.username}; an operator-authored message was observed`); } } } @@ -632,6 +656,33 @@ function messageKey(message: ChatMessage): string { return message.eventId ?? `${message.timestamp}:${message.sender}:${message.text}`; } +function hasUnattributedSelfMessage( + messages: readonly ChatMessage[], + opener: SentMessageRecord, + automatedFingerprints: readonly string[], + automatedEventIds: readonly string[] = [], + automatedTexts: readonly string[] = [], +): boolean { + const knownEventIds = new Set(automatedEventIds); + const knownFingerprints = new Map(); + const addKnown = (fingerprint: string) => { + knownFingerprints.set(fingerprint, (knownFingerprints.get(fingerprint) ?? 0) + 1); + }; + addKnown(conversationMessageFingerprint(opener.message)); + for (const fingerprint of automatedFingerprints) addKnown(fingerprint); + for (const text of automatedTexts) addKnown(conversationMessageFingerprint(text)); + + for (const message of messages) { + if (!message.isSelf || message.timestamp < opener.sentAt) continue; + if (message.eventId && knownEventIds.has(message.eventId)) continue; + const fingerprint = conversationMessageFingerprint(message.text); + const remaining = knownFingerprints.get(fingerprint) ?? 0; + if (remaining === 0) return true; + knownFingerprints.set(fingerprint, remaining - 1); + } + return false; +} + function replyPlan(state: ConversationState): { isLegacy: boolean; isHybridMediaAsk: boolean } { const isLegacy = state.automation.policyMode === 'legacy'; const isHybridMediaAsk = state.automation.policyMode === 'r4r-media' diff --git a/src/services/media-library-exchange.ts b/src/services/media-library-exchange.ts index b52b038..97cd523 100644 --- a/src/services/media-library-exchange.ts +++ b/src/services/media-library-exchange.ts @@ -14,6 +14,13 @@ export interface MediaExchangeChat { sendChatMessage(roomId: string, message: string, txnId?: string): Promise; } +export interface MediaExchangeOutboundAttribution { + eventIds: string[]; + texts: string[]; +} + +const RECIPROCAL_MESSAGE = 'Your turn — send me one back if you’re comfortable.'; + export type MediaExchangeResult = | { kind: 'not-requested' } | { kind: 'dry-run'; requestKind: MediaRequestKind } @@ -40,6 +47,14 @@ export class MediaLibraryExchangeService { ); } + getOutboundAttribution(conversationId: string): MediaExchangeOutboundAttribution { + const sends = this.repository.listConversationSends(this.accountId, conversationId); + return { + eventIds: sends.flatMap((send) => send.matrixMediaEventId ? [send.matrixMediaEventId] : []), + texts: sends.flatMap((send) => send.matrixTextEventId ? [RECIPROCAL_MESSAGE] : []), + }; + } + async handle( conversationId: string, incomingKey: string, @@ -103,7 +118,7 @@ export class MediaLibraryExchangeService { ? true : await this.chat.sendChatMessage( conversationId, - 'Your turn — send me one back if you’re comfortable.', + RECIPROCAL_MESSAGE, `rbot-library-reciprocal-${send.id}`, ); if (!textSent) { diff --git a/tests/conversation-automation.test.ts b/tests/conversation-automation.test.ts index d7d4ed8..7db16a9 100644 --- a/tests/conversation-automation.test.ts +++ b/tests/conversation-automation.test.ts @@ -13,7 +13,11 @@ function harness( rapportTurnsBeforeMediaAsk?: number; }, beforeSendLoader?: (conversationId: string, observed: ActiveChat) => Promise, - mediaExchange?: { detect(conversationId: string, text: string): string | null; handle(conversationId: string, incomingKey: string, text: string): Promise }, + mediaExchange?: { + detect(conversationId: string, text: string): string | null; + handle(conversationId: string, incomingKey: string, text: string): Promise; + getOutboundAttribution?(conversationId: string): { eventIds: string[]; texts: string[] }; + }, ) { const saved = new Map(); let lastSaved: ConversationState | undefined; @@ -37,7 +41,7 @@ function harness( const openers = { getByChatRoomId: vi.fn(() => openerSentAt === null ? undefined - : { sentAt: openerSentAt, subreddit: 'r4r' }), + : { sentAt: openerSentAt, subreddit: 'r4r', message: 'hello' }), }; const claims = { tryClaim: vi.fn(() => true), @@ -123,6 +127,11 @@ describe('conversation automation', () => { const fixture = harness(); const conversation = incoming(1, 'Sure'); conversation.messages[0].text = 'If you feel like it, send a pic where you are smiling.'; + fixture.openers.getByChatRoomId.mockReturnValue({ + sentAt: 0, + subreddit: 'r4r', + message: 'If you feel like it, send a pic where you are smiling.', + }); const result = await fixture.service.processConversation(conversation); @@ -159,6 +168,11 @@ describe('conversation automation', () => { ], }; const fixture = harness([], [], 0, undefined, async () => latest); + fixture.openers.getByChatRoomId.mockReturnValue({ + sentAt: 0, + subreddit: 'r4r', + message: 'Send a pic where you are smiling.', + }); fixture.claims.getClaim.mockReturnValue({ conversationId: 'room-alice', incomingKey: 'tattoo-event', status: 'pending', replyText: 'If you ever feel like sharing a quick snap, I am all eyes.', incomingAt: 30, @@ -216,6 +230,89 @@ describe('conversation automation', () => { expect(ai.generateMediaAsk).toHaveBeenCalledTimes(1); }); + it('persists a human override when a direct Reddit message appears before the next inbound reply', async () => { + const fixture = harness([], [], 0, { mode: 'r4r', maxAutomatedTurns: 5 }); + const conversation = incoming(2, 'recipient follow-up'); + conversation.messages = [ + { eventId: 'opener', sender: 'me', text: 'hello', timestamp: 1, isSelf: true }, + { eventId: 'first-inbound', sender: 'alice', text: 'first reply', timestamp: 2, isSelf: false }, + { eventId: 'manual', sender: 'me', text: 'operator reply', timestamp: 3, isSelf: true }, + { eventId: 'next-inbound', sender: 'alice', text: 'recipient follow-up', timestamp: 4, isSelf: false }, + ]; + + expect(await fixture.service.processConversation(conversation)).toBeNull(); + + expect(fixture.chat.sendChatMessage).not.toHaveBeenCalled(); + expect(fixture.ai.generateReply).not.toHaveBeenCalled(); + expect(fixture.getSaved()?.automation).toMatchObject({ + status: 'user-overridden', + handoffReason: 'A manual reply was observed in this conversation.', + }); + }); + + it('does not mistake a recorded automated reply for an operator message', async () => { + const fixture = harness([], [], 0, { mode: 'r4r', maxAutomatedTurns: 5 }); + const first = await fixture.service.processConversation(incoming(1)); + expect(first?.kind).toBe('sent'); + const conversation = incoming(2, 'second reply'); + conversation.messages = [ + { eventId: 'opener', sender: 'me', text: 'hello', timestamp: 1, isSelf: true }, + { eventId: 'first-inbound', sender: 'alice', text: 'first reply', timestamp: 2, isSelf: false }, + { eventId: 'automated', sender: 'me', text: 'That sounds fun. What got you into it?', timestamp: 3, isSelf: true }, + { eventId: 'second-inbound', sender: 'alice', text: 'second reply', timestamp: 4, isSelf: false }, + ]; + + expect((await fixture.service.processConversation(conversation))?.kind).toBe('sent'); + expect(fixture.chat.sendChatMessage).toHaveBeenCalledTimes(2); + expect(fixture.getSaved()?.automation.status).toBe('active'); + }); + + it('does not mistake confirmed exchange-library sends for operator messages', async () => { + const reciprocal = 'Your turn — send me one back if you’re comfortable.'; + const mediaExchange = { + detect: vi.fn().mockReturnValue(null), + handle: vi.fn(), + getOutboundAttribution: vi.fn().mockReturnValue({ + eventIds: ['library-image'], + texts: [reciprocal], + }), + }; + const fixture = harness([], [], 0, { mode: 'r4r', maxAutomatedTurns: 5 }, undefined, mediaExchange); + const conversation = incoming(2, 'thanks'); + conversation.messages = [ + { eventId: 'opener', sender: 'me', text: 'hello', timestamp: 1, isSelf: true }, + { eventId: 'request', sender: 'alice', text: 'send a pic', timestamp: 2, isSelf: false }, + { eventId: 'library-image', sender: 'me', text: 'image.jpg', timestamp: 3, isSelf: true }, + { eventId: 'library-text', sender: 'me', text: reciprocal, timestamp: 4, isSelf: true }, + { eventId: 'thanks', sender: 'alice', text: 'thanks', timestamp: 5, isSelf: false }, + ]; + + expect((await fixture.service.processConversation(conversation))?.kind).toBe('sent'); + expect(fixture.getSaved()?.automation.status).toBe('active'); + }); + + it('supersedes a pending reply when Matrix contains a different operator-authored message', async () => { + const fixture = harness(); + fixture.claims.listPending.mockReturnValue([{ + conversationId: 'room-alice', incomingKey: 'first-inbound', status: 'pending', + replyText: 'Expected automated reply', incomingAt: 2, claimedAt: 2, + lastAttemptAt: null, attemptCount: 0, sentAt: null, supersededAt: null, + }] as never); + fixture.chat.getCachedConversations.mockResolvedValue([{ + ...incoming(2), + messages: [ + { eventId: 'opener', sender: 'me', text: 'hello', timestamp: 1, isSelf: true }, + { eventId: 'first-inbound', sender: 'alice', text: 'first reply', timestamp: 2, isSelf: false }, + { eventId: 'manual', sender: 'me', text: 'Different operator reply', timestamp: 3, isSelf: true }, + ], + }]); + + await fixture.service.processPendingConversations(); + + expect(fixture.claims.markSent).not.toHaveBeenCalled(); + expect(fixture.claims.markSuperseded).toHaveBeenCalledWith('room-alice', 'first-inbound'); + }); + it('does not reply when the room has no recorded opener', async () => { const { service, chat, ai, claims } = harness([], [], null); @@ -365,7 +462,7 @@ describe('conversation automation', () => { ...incoming(1), unread: false, isFromUs: true, lastMessageTime: 4, messages: [ ...incoming(1).messages, - { eventId: 'out-reply', sender: 'me', text: 'reply', timestamp: 4, isSelf: true }, + { eventId: 'out-reply', sender: 'me', text: 'Would you be comfortable sharing a selfie?', timestamp: 4, isSelf: true }, ], }; const pending = { diff --git a/tests/media-library-exchange.test.ts b/tests/media-library-exchange.test.ts index e272e08..2d12866 100644 --- a/tests/media-library-exchange.test.ts +++ b/tests/media-library-exchange.test.ts @@ -58,6 +58,10 @@ describe('MediaLibraryExchangeService', () => { expect(new Set(sentAssets).size).toBe(sentAssets.length); expect(chat.sendChatImage.mock.invocationCallOrder[0]).toBeLessThan(chat.sendChatMessage.mock.invocationCallOrder[0]); expect(repository.get('penis')).toMatchObject({ sendCount: 1, lastSentAt: expect.any(Number) }); + expect(service.getOutboundAttribution('room')).toEqual({ + eventIds: ['$1', '$2', '$3'], + texts: Array(3).fill('Your turn — send me one back if you’re comfortable.'), + }); db.close(); });