Harden conversation replies and prioritize newest candidates

tests/candidate-queue.test.ts

@@ -94,6 +94,42 @@ describe('CandidateQueueRepository', () => {
     expect(queue.countActive()).toBe(2);
   });
 
+  it('admits, screens, and sends the newest posts before older higher-priority posts', () => {
+    queue.enqueue([
+      post('old-high-priority', 'alice', 100),
+      post('new-low-priority', 'bob', 300),
+    ], 'image', 1, 1_000, 1);
+
+    expect(queue.listActive().map((candidate) => candidate.postId)).toEqual(['new-low-priority']);
+
+    queue.enqueue([post('old-high-priority', 'alice', 100)], 'image', 10, 1_001);
+    const newestScreening = queue.claimForScreening('screen-new', 100, 1_010)!;
+    expect(newestScreening.postId).toBe('new-low-priority');
+    queue.markReady(newestScreening.postId, 'screen-new', prepared(newestScreening.post), 1_011);
+
+    const olderScreening = queue.claimForScreening('screen-old', 100, 1_012)!;
+    queue.markReady(olderScreening.postId, 'screen-old', prepared(olderScreening.post), 1_013);
+
+    expect(queue.claimReady('account-a', 'send', 100, 1_014)?.postId).toBe('new-low-priority');
+  });
+
+  it('promotes the newest due delivery retry when ready capacity is limited', () => {
+    queue.enqueue([post('new-low-priority', 'bob', 300)], 'image', 1, 1_000);
+    queue.enqueue([post('old-high-priority', 'alice', 100)], 'image', 10, 1_001);
+
+    for (const worker of ['screen-new', 'screen-old']) {
+      const candidate = queue.claimForScreening(worker, 100, 1_010)!;
+      queue.markReady(candidate.postId, worker, prepared(candidate.post), 1_011);
+    }
+    for (const [accountId, worker] of [['account-a', 'send-new'], ['account-b', 'send-old']] as const) {
+      const candidate = queue.claimReady(accountId, worker, 100, 1_012)!;
+      queue.deferDelivery(candidate.postId, worker, 'Matrix rate limited', 1_100, 1_013);
+    }
+
+    expect(queue.promoteDueRetries(1, 1_100)).toBe(1);
+    expect(queue.claimReady('account-a', 'retry', 100, 1_101)?.postId).toBe('new-low-priority');
+  });
+
   it('excludes posts and recipients already handled anywhere', () => {
     db.prepare(`
       INSERT INTO processed_posts (account_id, id, subreddit, author, title, processed_at)
@@ -140,13 +176,13 @@ describe('CandidateQueueRepository', () => {
     });
   });
 
-  it('lists only active candidates in actionable stage order', () => {
+  it('lists only active candidates in newest-post order across stages', () => {
     queue.enqueue([
-      post('waiting', 'alice'),
-      post('screening', 'bob'),
-      post('ready', 'carol'),
-      post('sending', 'dana'),
-      post('rejected', 'erin'),
+      post('waiting', 'alice', 100),
+      post('screening', 'bob', 500),
+      post('ready', 'carol', 400),
+      post('sending', 'dana', 300),
+      post('rejected', 'erin', 200),
     ], 'image', 1, 1_000);
     queue.claimForScreening('worker-1', 1_000, 1_010);
     const ready = queue.claimForScreening('worker-2', 1_000, 1_020)!;
@@ -155,8 +191,9 @@ describe('CandidateQueueRepository', () => {
     const rejected = queue.claimForScreening('worker-4', 1_000, 1_040)!;
     queue.reject(rejected.postId, 'worker-4', 'Seller signals detected', 1_041);
 
-    expect(queue.listActive().map((candidate) => candidate.status)).toEqual([
-      'sending', 'screening', 'discovered', 'discovered',
+    const active = queue.listActive();
+    expect(active.map((candidate) => candidate.postId)).toEqual([
+      'screening', 'ready', 'rejected', 'waiting',
     ]);
     expect(queue.listActive().map((candidate) => candidate.postId)).not.toContain(rejected.postId);
     expect(queue.getOldestActiveAt()).toBe(1_000);

tests/chat-service-outcome-ingestion.test.ts

@@ -2,7 +2,8 @@ import fs from 'fs/promises';
 import os from 'os';
 import path from 'path';
 import { afterEach, describe, expect, it, vi } from 'vitest';
-import { ConversationOutcomesRepository } from '../src/database/repositories.js';
+import { ConversationOutcomesRepository, ConversationStateRepository } from '../src/database/repositories.js';
+import { buildConversationState } from '../src/services/conversation-state.js';
 import { ChatCache } from '../src/services/chat/cache.js';
 import { MatrixClient } from '../src/services/chat/matrix/client.js';
 import type { MatrixSyncResponse } from '../src/services/chat/matrix/types.js';
@@ -46,6 +47,7 @@ describe('ChatService outcome ingestion', () => {
     expect(fixture.outcomes.get(fixture.sentMessageId)?.outcome).toBe('unknown');
     expect(fixture.repos.successfulMessagesRepo.count()).toBe(0);
 
+    fixture.markConversationComplete();
     await fixture.service.refreshTopChats();
     expect(fixture.outcomes.get(fixture.sentMessageId)?.outcome).toBe('engaged');
     expect(fixture.repos.successfulMessagesRepo.count()).toBe(1);
@@ -67,6 +69,7 @@ describe('ChatService outcome ingestion', () => {
       ],
     });
     fixture.repos.messagesRepo.markGotResponse(fixture.sentMessageId);
+    fixture.markConversationComplete();
     fixture.matrix.sync = vi.fn(async () => syncResponse('s2', [])) as typeof fixture.matrix.sync;
 
     await fixture.service.runStartupSync();
@@ -94,6 +97,7 @@ async function createFixture(options: {
   openDatabases.push(db);
   const repos = createTestRepositories(db);
   const outcomes = new ConversationOutcomesRepository(db);
+  const conversationStates = new ConversationStateRepository(db);
   const now = Date.now();
   const roomId = '!alice:example.test';
 
@@ -146,10 +150,19 @@ async function createFixture(options: {
     undefined,
     authenticatedAccountAuth,
     outcomes,
+    conversationStates,
   );
   await service.loadCache();
 
-  return { db, repos, outcomes, now, roomId, sentMessageId, matrix, service };
+  return {
+    db, repos, outcomes, now, roomId, sentMessageId, matrix, service,
+    markConversationComplete() {
+      const state = buildConversationState([]);
+      state.automation.status = 'human-required';
+      state.automation.handoffReason = 'Conversation complete.';
+      conversationStates.save(roomId, state);
+    },
+  };
 }
 
 function syncResponse(nextBatch: string, events: ReturnType<typeof messageEvent>[]): MatrixSyncResponse {

tests/conversation-automation.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, it, vi } from 'vitest';
 import { ConversationAutomationService } from '../src/services/conversation-automation.js';
 import type { ConversationState } from '../src/services/conversation-state.js';
+import type { ActiveChat } from '../src/services/chat.js';
 
 function harness(
   sellerKeywords: string[] = [],
@@ -11,6 +12,7 @@ function harness(
     maxAutomatedTurns: number;
     rapportTurnsBeforeMediaAsk?: number;
   },
+  beforeSendLoader?: (conversationId: string, observed: ActiveChat) => Promise<ActiveChat | undefined>,
 ) {
   const saved = new Map<string, ConversationState>();
   let lastSaved: ConversationState | undefined;
@@ -43,6 +45,7 @@ function harness(
     markAttempt: vi.fn(() => true),
     releasePending: vi.fn(() => true),
     markSent: vi.fn(() => true),
+    markSuperseded: vi.fn(() => true),
   };
   const rateLimiter = {
     throttle: vi.fn(async (operation: () => Promise<boolean>) => operation()),
@@ -59,6 +62,7 @@ function harness(
     }] : [],
     () => 1_000,
     0,
+    beforeSendLoader ?? (async (_conversationId, observed) => observed),
   );
   return { service, chat, ai, states, openers, claims, rateLimiter, getSaved: () => lastSaved };
 }
@@ -75,18 +79,17 @@ function incoming(turn: number, text = `reply ${turn}`, media?: Array<{ original
 }
 
 describe('conversation automation', () => {
-  it('sends once per inbound turn and flags human handoff after the third send', async () => {
+  it('allows only one autonomous media request and then hands off', async () => {
     const { service, chat, getSaved } = harness();
 
     expect((await service.processConversation(incoming(1)))?.remainingTurns).toBe(2);
-    expect((await service.processConversation(incoming(2)))?.remainingTurns).toBe(1);
-    const third = await service.processConversation(incoming(3));
+    const second = await service.processConversation(incoming(2));
 
-    expect(third).toMatchObject({ kind: 'human-required', remainingTurns: 0 });
-    expect(chat.sendChatMessage).toHaveBeenCalledTimes(3);
-    expect(getSaved()?.automation).toMatchObject({ turnsUsed: 3, status: 'human-required' });
-    expect(getSaved()?.automation.aiMessageFingerprints).toHaveLength(3);
-    expect(await service.processConversation(incoming(3))).toBeNull();
+    expect(second).toMatchObject({ kind: 'human-required', remainingTurns: 0 });
+    expect(chat.sendChatMessage).toHaveBeenCalledOnce();
+    expect(getSaved()?.automation).toMatchObject({ turnsUsed: 1, status: 'human-required' });
+    expect(getSaved()?.automation.aiMessageFingerprints).toHaveLength(1);
+    expect(await service.processConversation(incoming(2))).toBeNull();
   });
 
   it('stops without sending when the user declines images', async () => {
@@ -98,6 +101,63 @@ describe('conversation automation', () => {
     expect(getSaved()?.automation.status).toBe('boundary');
   });
 
+  it('hands off immediately when the opener already made the conversation media request', async () => {
+    const fixture = harness();
+    const conversation = incoming(1, 'Sure');
+    conversation.messages[0].text = 'If you feel like it, send a pic where you are smiling.';
+
+    const result = await fixture.service.processConversation(conversation);
+
+    expect(result).toMatchObject({ kind: 'human-required', remainingTurns: 0 });
+    expect(fixture.ai.generateMediaAsk).not.toHaveBeenCalled();
+    expect(fixture.chat.sendChatMessage).not.toHaveBeenCalled();
+    expect(fixture.getSaved()?.automation.handoffReason).toContain('media request has already been made');
+  });
+
+  it('supersedes a queued reply when a refusal arrives during the rate-limit wait', async () => {
+    const tattooEvent = {
+      eventId: 'tattoo-event', sender: 'alice',
+      text: 'I edit out my tattoos for the sake of anonymity but that is the only filtering done.',
+      timestamp: 30, isSelf: false,
+    };
+    const observed: ActiveChat = {
+      username: 'JustProwling', roomName: 'JustProwling', roomId: 'room-alice', url: '/room/room-alice',
+      lastMessageTime: 30, lastMessageSnippet: tattooEvent.text, isFromUs: false, unread: true,
+      messages: [
+        { eventId: 'opener', sender: 'me', text: 'Send a pic where you are smiling.', timestamp: 10, isSelf: true },
+        tattooEvent,
+      ],
+    };
+    const latest: ActiveChat = {
+      ...observed,
+      lastMessageTime: 40,
+      lastMessageSnippet: 'Zero interest in sending you anything lol',
+      messages: [
+        ...observed.messages!,
+        {
+          eventId: 'refusal-event', sender: 'alice',
+          text: 'Zero interest in sending you anything lol', timestamp: 40, isSelf: false,
+        },
+      ],
+    };
+    const fixture = harness([], [], 0, undefined, async () => latest);
+    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,
+      claimedAt: 0, lastAttemptAt: null, attemptCount: 0, sentAt: null, supersededAt: null,
+    } as never);
+
+    const result = await fixture.service.processConversation(observed);
+
+    expect(result?.kind).toBe('boundary');
+    expect(fixture.claims.markSuperseded).toHaveBeenCalledWith('room-alice', 'tattoo-event');
+    expect(fixture.chat.sendChatMessage).not.toHaveBeenCalled();
+    expect(fixture.getSaved()).toMatchObject({
+      boundaries: ['does not want to send photos or videos'],
+      automation: { status: 'boundary', lastIncomingKey: 'refusal-event' },
+    });
+  });
+
   it('flags a potential upsell for human handoff without generating a reply', async () => {
     const { service, chat, ai, getSaved } = harness(['onlyfans'], ['fansly.com']);
     const result = await service.processConversation(incoming(1, 'My Fansly is https://www.fansly.com/example'));
@@ -323,7 +383,7 @@ describe('conversation automation', () => {
     expect(ai.generateMediaAsk).not.toHaveBeenCalled();
     expect(chat.sendChatMessage).toHaveBeenCalledTimes(2);
     expect(getSaved()?.automation).toMatchObject({
-      policyMode: 'r4r', policyVersion: 2, maxAutomatedTurns: 2, turnsUsed: 2,
+      policyMode: 'r4r', policyVersion: 3, maxAutomatedTurns: 2, turnsUsed: 2,
     });
     expect(getSaved()?.askedForMedia).toBe(false);
   });
@@ -396,7 +456,7 @@ describe('conversation automation', () => {
     expect(fixture.getSaved()?.askedForMedia).toBe(true);
     expect(fixture.getSaved()?.automation).toMatchObject({
       policyMode: 'r4r-media',
-      policyVersion: 2,
+      policyVersion: 3,
       rapportTurnsBeforeMediaAsk: 2,
       hybridPhase: 'post-media-ask',
       turnsUsed: 4,
@@ -445,4 +505,14 @@ describe('conversation automation', () => {
     expect(result?.kind).toBe('human-required');
     expect(fixture.chat.sendChatMessage).toHaveBeenCalledTimes(1);
   });
+
+  it('rejects a coercive legacy media request', async () => {
+    const fixture = harness();
+    fixture.ai.generateMediaAsk.mockResolvedValue('Prove me wrong and send a pic.');
+
+    const result = await fixture.service.processConversation(incoming(1));
+
+    expect(result?.kind).toBe('human-required');
+    expect(fixture.chat.sendChatMessage).not.toHaveBeenCalled();
+  });
 });

tests/conversation-outcome-recorder.test.ts

@@ -28,7 +28,11 @@ function message(): SentMessageRecord {
 function dependencies() {
   return {
     messages: { markGotResponse: vi.fn() } as unknown as MessagesRepository,
-    successful: { insert: vi.fn(), existsByMessage: vi.fn(() => false) } as unknown as SuccessfulMessagesRepository,
+    successful: {
+      insert: vi.fn(),
+      existsByMessage: vi.fn(() => false),
+      deleteBySentMessageId: vi.fn(() => false),
+    } as unknown as SuccessfulMessagesRepository,
     outcomes: { record: vi.fn(), suppressRecipient: vi.fn() } as unknown as ConversationOutcomesRepository,
     logger: createMockLogger(),
   };
@@ -44,7 +48,9 @@ describe('recordConversationReply', () => {
 
   it('stores only reciprocal engaged conversations as learning examples', () => {
     const deps = dependencies();
-    expect(recordConversationReply(message(), 'Tell me more', { messageCount: 4, reciprocal: true }, deps)).toBe('engaged');
+    expect(recordConversationReply(message(), 'Tell me more', {
+      messageCount: 4, reciprocal: true, learningEligible: true,
+    }, deps)).toBe('engaged');
     expect(deps.successful.insert).toHaveBeenCalledWith(expect.objectContaining({ sentMessageId: 1 }));
   });
 
@@ -53,5 +59,19 @@ describe('recordConversationReply', () => {
     expect(recordConversationReply(message(), 'Stop messaging me', { messageCount: 2 }, deps)).toBe('boundary');
     expect(deps.outcomes.suppressRecipient).toHaveBeenCalledWith('alice', 'boundary');
     expect(deps.successful.insert).not.toHaveBeenCalled();
+    expect(deps.successful.deleteBySentMessageId).toHaveBeenCalledWith(1);
+  });
+
+  it('removes an earlier learning example when the full conversation later declines', () => {
+    const deps = dependencies();
+    expect(recordConversationReply(message(), "I'm good", {
+      messageCount: 8,
+      reciprocal: true,
+      learningEligible: true,
+      mediaWasRequested: true,
+      responses: ['Sure', 'Zero interest in sending you anything lol', "I'm good"],
+    }, deps)).toBe('declined');
+    expect(deps.successful.deleteBySentMessageId).toHaveBeenCalledWith(1);
+    expect(deps.successful.insert).not.toHaveBeenCalled();
   });
 });

tests/conversation-outcome.test.ts

@@ -15,6 +15,25 @@ describe('conversation outcome classification', () => {
     expect(classifyConversationOutcome('That sounds fun, tell me more', {
       messageCount: 4,
       reciprocal: true,
+      learningEligible: true,
     })).toBe('engaged');
   });
+
+  it('lets a later refusal override earlier reciprocal engagement', () => {
+    expect(classifyConversationOutcome("I'm good", {
+      messageCount: 8,
+      reciprocal: true,
+      learningEligible: true,
+      mediaWasRequested: true,
+      responses: ['Sure, tell me more', 'Zero interest in sending you anything lol', "I'm good"],
+    })).toBe('declined');
+  });
+
+  it('does not promote a positive reciprocal thread while automation is active', () => {
+    expect(classifyConversationOutcome('Sure, tell me more', {
+      messageCount: 4,
+      reciprocal: true,
+      learningEligible: false,
+    })).toBe('unknown');
+  });
 });

tests/conversation-state.test.ts

@@ -50,6 +50,28 @@ describe('conversation state', () => {
     });
   });
 
+  it.each([
+    'Zero interest in sending you anything lol',
+    "I'm good",
+    'Rather not, thanks',
+  ])('treats a contextual media refusal as a boundary: %s', (refusal) => {
+    const state = buildConversationState([
+      message('Would you be comfortable sharing a selfie?', true),
+      message(refusal),
+    ]);
+
+    expect(state.boundaries).toContain('does not want to send photos or videos');
+  });
+
+  it('does not treat a conversational wellbeing answer as a media refusal', () => {
+    const state = buildConversationState([
+      message('How are you?', true),
+      message("I'm good"),
+    ]);
+
+    expect(state.boundaries).toEqual([]);
+  });
+
   it('allows three progressive image-sharing turns before human handoff', () => {
     const state = buildConversationState([message('Hello', true), message('Hey')]);
     expect(remainingAutomationTurns(state)).toBe(3);
@@ -75,7 +97,7 @@ describe('conversation state', () => {
     expect(state?.automation).toMatchObject({ turnsUsed: 0, status: 'active' });
     expect(state?.sellerDetected).toBe(false);
     expect(state?.automation).toMatchObject({
-      policyMode: 'legacy', policyVersion: 2, maxAutomatedTurns: 3,
+      policyMode: 'legacy', policyVersion: 3, maxAutomatedTurns: 3,
       rapportTurnsBeforeMediaAsk: 3, hybridPhase: null,
     });
   });

tests/database.test.ts

@@ -160,7 +160,7 @@ describe('Database Client', () => {
         DROP TABLE conversation_automation_claims;
         DROP TABLE ai_usage;
       `);
-      db.pragma(`user_version = ${latestVersion - 5}`);
+      db.pragma(`user_version = ${latestVersion - 6}`);
       runMigrations(db);
 
       expect(db.prepare('SELECT recipient, reason FROM recipient_suppressions ORDER BY recipient COLLATE NOCASE').all())
@@ -170,6 +170,46 @@ describe('Database Client', () => {
         ]);
     });
 
+    it('preserves pending reply claims when adding auditable superseded status', () => {
+      runMigrations(db);
+      const latestVersion = db.pragma('user_version', { simple: true }) as number;
+      db.exec(`
+        DROP TABLE conversation_automation_claims;
+        CREATE TABLE conversation_automation_claims (
+          account_id TEXT NOT NULL,
+          conversation_id TEXT NOT NULL,
+          incoming_key TEXT NOT NULL,
+          status TEXT NOT NULL CHECK(status IN ('pending', 'sent')),
+          claimed_at INTEGER NOT NULL,
+          sent_at INTEGER,
+          reply_text TEXT,
+          incoming_at INTEGER,
+          last_attempt_at INTEGER,
+          attempt_count INTEGER NOT NULL DEFAULT 0,
+          PRIMARY KEY (account_id, conversation_id, incoming_key),
+          FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id) ON DELETE CASCADE
+        );
+        INSERT INTO conversation_automation_claims (
+          account_id, conversation_id, incoming_key, status, claimed_at,
+          reply_text, incoming_at, last_attempt_at, attempt_count
+        ) VALUES (
+          '__legacy_unassigned__', 'room', 'event', 'pending', 10,
+          'durable reply', 5, 11, 1
+        );
+      `);
+      db.pragma(`user_version = ${latestVersion - 1}`);
+
+      runMigrations(db);
+
+      expect(db.prepare(`
+        SELECT status, reply_text, incoming_at, last_attempt_at, attempt_count, superseded_at
+        FROM conversation_automation_claims
+      `).get()).toEqual({
+        status: 'pending', reply_text: 'durable reply', incoming_at: 5,
+        last_attempt_at: 11, attempt_count: 1, superseded_at: null,
+      });
+    });
+
     it('processed_posts has correct schema', () => {
       runMigrations(db);
       const columns = db.prepare('PRAGMA table_info(processed_posts)').all() as { name: string; notnull: number }[];

tests/repository-account-isolation.test.ts

@@ -121,13 +121,26 @@ describe('account-bound domain repositories', () => {
     expect(claims.getClaim('room', 'event')).toEqual({
       conversationId: 'room', incomingKey: 'event', status: 'pending',
       replyText: 'durable reply', incomingAt: 50, claimedAt: 60,
-      lastAttemptAt: 70, attemptCount: 1, sentAt: null,
+      lastAttemptAt: 70, attemptCount: 1, sentAt: null, supersededAt: null,
     });
     expect(claims.listPending()).toEqual([expect.objectContaining({ incomingKey: 'event' })]);
     expect(claims.releasePending('room', 'event')).toBe(true);
     expect(claims.listPending()).toEqual([]);
   });
 
+  it('keeps superseded reply claims as an auditable terminal record', () => {
+    const claims = new ConversationAutomationClaimRepository(db, ACCOUNT_A);
+    expect(claims.tryClaim('room', 'stale-event', 'stale reply', 50, 60)).toBe(true);
+
+    expect(claims.markSuperseded('room', 'stale-event', 80)).toBe(true);
+    expect(claims.getClaim('room', 'stale-event')).toMatchObject({
+      status: 'superseded',
+      supersededAt: 80,
+      sentAt: null,
+    });
+    expect(claims.listPending()).toEqual([]);
+  });
+
   it('claims recipients globally and blocks usernames with existing chat history', () => {
     db.prepare(`
       INSERT INTO reddit_conversations (