Pause automation after manual Reddit replies

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?"}

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<string, number>();
+  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'

src/services/media-library-exchange.ts

@@ -14,6 +14,13 @@ export interface MediaExchangeChat {
   sendChatMessage(roomId: string, message: string, txnId?: string): Promise<boolean>;
 }
 
+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) {

tests/conversation-automation.test.ts

@@ -13,7 +13,11 @@ function harness(
     rapportTurnsBeforeMediaAsk?: number;
   },
   beforeSendLoader?: (conversationId: string, observed: ActiveChat) => Promise<ActiveChat | undefined>,
-  mediaExchange?: { detect(conversationId: string, text: string): string | null; handle(conversationId: string, incomingKey: string, text: string): Promise<object> },
+  mediaExchange?: {
+    detect(conversationId: string, text: string): string | null;
+    handle(conversationId: string, incomingKey: string, text: string): Promise<object>;
+    getOutboundAttribution?(conversationId: string): { eventIds: string[]; texts: string[] };
+  },
 ) {
   const saved = new Map<string, ConversationState>();
   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 = {

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();
   });