Fix conversation outcome learning ingestion

src/services/chat/service.ts

@@ -480,12 +480,14 @@ export class ChatService {
     if (!lastFullSyncAt || this.cache.get().length === 0) {
       this.logger.info('[chat] startup: no prior full sync recorded → running full backfill');
       const chats = await this.fetchAllMessages();
+      this.backfillConversationOutcomes();
       return { kind: 'full', chats: chats.length };
     }
     this.logger.info(
       `[chat] startup: full sync was ${new Date(lastFullSyncAt).toISOString()} → running delta`,
     );
     const chats = await this.getActiveChats();
+    this.backfillConversationOutcomes();
     return { kind: 'delta', chats: chats.length };
   }
 
@@ -772,11 +774,6 @@ export class ChatService {
       if (this.mediaService && scraped.some((message) => message.mediaUrls.length > 0)) {
         try {
           newMessages = await this.mediaService.attachMedia(chat, scraped);
-          if (newMessages.some(
-            (message) => !message.isSelf && (message.media?.length ?? 0) > 0,
-          )) {
-            this.recordReceivedImage(chat);
-          }
         } catch (err) {
           this.logger.warn(
             `[chat] delta media attach failed for ${roomId}: ${(err as Error).message}`,
@@ -784,6 +781,7 @@ export class ChatService {
         }
       }
       this.mergeNewMessages(chat, newMessages);
+      this.recordConversationPerformance(chat);
     }
 
     const merged = Array.from(existing.values()).sort(
@@ -856,8 +854,8 @@ export class ChatService {
   }
 
   private mergeNewMessages(chat: ActiveChat, newMessages: ChatMessage[]): void {
-    if (!chat.messages || chat.messages.length === 0) return;
-    chat.messages = this.normalizeMessages([...chat.messages, ...newMessages]);
+    if (newMessages.length === 0) return;
+    chat.messages = this.normalizeMessages([...(chat.messages ?? []), ...newMessages]);
   }
 
   /**
@@ -1005,43 +1003,65 @@ export class ChatService {
     const firstSelf = chat.messages.find((m) => m.isSelf);
     if (firstSelf) chat.originMessage = firstSelf.text;
 
-    const hasReply = chat.messages.some((m) => !m.isSelf);
-    const hasReceivedImage = chat.messages.some(
-      (message) => !message.isSelf && (message.media?.length ?? 0) > 0,
-    );
-    let performanceMessage: SentMessageRecord | undefined;
-    let firstAttributedReply = false;
+    this.recordConversationPerformance(chat);
+  }
 
-    if (firstSelf && chat.roomId && this.messagesRepo) {
+  /**
+   * Project a cached or newly-synced conversation into durable performance
+   * state. Safe to call repeatedly: outcome rows are upserted and successful
+   * examples are de-duplicated by their original message.
+   */
+  private recordConversationPerformance(chat: ActiveChat): boolean {
+    const messages = chat.messages ?? [];
+    if (!chat.roomId || !this.messagesRepo) return false;
+
+    let performanceMessage: SentMessageRecord | undefined;
     try {
       const existing = this.messagesRepo.getByChatRoomId(chat.roomId);
       const hasPerformanceContext = Boolean(
         existing?.postId || existing?.strategyId || existing?.subreddit,
       );
-        if (existing && hasPerformanceContext) {
-          performanceMessage = existing;
-          if (hasReply && !existing.gotResponse) {
-            firstAttributedReply = true;
-          }
-          if (hasReceivedImage && !existing.gotImage) {
-            this.messagesRepo.markGotImage(existing.id);
-          }
-        }
+      if (existing && hasPerformanceContext) performanceMessage = existing;
     } catch (err) {
       this.logger.warn('[chat] failed to associate reply with performance context:', err);
+      return false;
     }
+    if (!performanceMessage) return false;
+
+    const firstSelf = messages.find(
+      (message) => message.isSelf && message.timestamp >= performanceMessage.sentAt - 1_000,
+    );
+    const conversationStart = firstSelf?.timestamp ?? performanceMessage.sentAt;
+
+    // Ignore messages from an older exchange in a reused room. The one-second
+    // tolerance accommodates the optimistic opener timestamp being reconciled
+    // with Matrix's authoritative event timestamp. Legacy caches can omit the
+    // opener, so the durable sent timestamp is the fallback boundary.
+    const conversationMessages = messages.filter(
+      (message) => message.timestamp >= conversationStart - 1_000,
+    );
+    const incomingMessages = conversationMessages.filter((message) => !message.isSelf);
+    const hasReply = incomingMessages.length > 0;
+    const hasReceivedImage = incomingMessages.some(
+      (message) => (message.media?.length ?? 0) > 0,
+    );
+    const firstAttributedReply = hasReply && !performanceMessage.gotResponse;
+    if (hasReceivedImage && !performanceMessage.gotImage) {
+      this.messagesRepo.markGotImage(performanceMessage.id);
     }
 
-    if (firstSelf && hasReply && performanceMessage && this.successfulMessagesRepo && this.outcomesRepo && this.messagesRepo) {
-      const firstReply = chat.messages.find((m) => !m.isSelf);
-      const selfCount = chat.messages.filter((m) => m.isSelf).length;
-      const otherCount = chat.messages.length - selfCount;
+    if (hasReply && this.successfulMessagesRepo && this.outcomesRepo) {
+      const firstReply = incomingMessages[0];
+      const cachedSelfCount = conversationMessages.filter((message) => message.isSelf).length;
+      const selfCount = cachedSelfCount + (firstSelf ? 0 : 1);
+      const otherCount = incomingMessages.length;
+      const wasMissingOutcome = this.outcomesRepo.get(performanceMessage.id) === undefined;
       try {
         recordConversationReply(
           performanceMessage,
           firstReply?.text.slice(0, 200) ?? '',
           {
-            messageCount: chat.messages.length,
+            messageCount: selfCount + otherCount,
             reciprocal: selfCount >= 2 && otherCount >= 2,
           },
           {
@@ -1065,24 +1085,24 @@ export class ChatService {
         }
       } catch (err) {
         this.logger.warn('[chat] failed to record conversation outcome:', err);
+        return false;
       }
-    } else if (hasReply && performanceMessage && this.messagesRepo && !performanceMessage.gotResponse) {
+      return wasMissingOutcome;
+    } else if (hasReply && !performanceMessage.gotResponse) {
       this.messagesRepo.markGotResponse(performanceMessage.id);
       this.logger.warn('[chat] response recorded without canonical outcome repository; excluded from learning');
     }
+    return false;
   }
 
-  private recordReceivedImage(chat: ActiveChat): void {
-    if (!chat.roomId || !this.messagesRepo) return;
-    try {
-      const existing = this.messagesRepo.getByChatRoomId(chat.roomId);
-      const hasPerformanceContext = Boolean(
-        existing?.postId || existing?.strategyId || existing?.subreddit,
-      );
-      if (!existing || !hasPerformanceContext) return;
-      if (!existing.gotImage) this.messagesRepo.markGotImage(existing.id);
-    } catch (err) {
-      this.logger.warn('[chat] failed to associate received image with performance context:', err);
+  /** Repair cached replies that predate canonical outcome recording. */
+  private backfillConversationOutcomes(): void {
+    let repaired = 0;
+    for (const chat of this.cache.get()) {
+      if (this.recordConversationPerformance(chat)) repaired += 1;
+    }
+    if (repaired > 0) {
+      this.logger.info(`[chat] backfilled ${repaired} missing conversation outcome(s)`);
     }
   }
 

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

@@ -0,0 +1,176 @@
+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 { 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';
+import { ChatService } from '../src/services/chat/service.js';
+import { createMockLogger } from './mockLogger.js';
+import { authenticatedAccountAuth, createTestDb, createTestRepositories } from './testUtils.js';
+
+const credentials = {
+  homeserver: 'https://matrix.example.test',
+  accessToken: 'token',
+  userId: '@me:example.test',
+  capturedAt: Date.now(),
+};
+
+const temporaryDirectories: string[] = [];
+const openDatabases: Array<ReturnType<typeof createTestDb>> = [];
+
+afterEach(async () => {
+  for (const database of openDatabases.splice(0)) database.close();
+  await Promise.all(temporaryDirectories.splice(0).map((directory) =>
+    fs.rm(directory, { recursive: true, force: true })));
+});
+
+describe('ChatService outcome ingestion', () => {
+  it('records delta replies, promotes reciprocal engagement, and avoids duplicate examples', async () => {
+    const fixture = await createFixture();
+    const firstReplyAt = fixture.now - 2_000;
+    const secondSelfAt = fixture.now - 1_500;
+    const secondReplyAt = fixture.now - 1_000;
+    fixture.matrix.sync = vi.fn()
+      .mockResolvedValueOnce(syncResponse('s2', [
+        messageEvent('$reply-1', '@alice:example.test', firstReplyAt, 'Thanks!'),
+      ]))
+      .mockResolvedValue(syncResponse('s3', [
+        messageEvent('$reply-1', '@alice:example.test', firstReplyAt, 'Thanks!'),
+        messageEvent('$self-2', credentials.userId, secondSelfAt, 'What do you think?'),
+        messageEvent('$reply-2', '@alice:example.test', secondReplyAt, 'Tell me more'),
+      ])) as typeof fixture.matrix.sync;
+
+    await fixture.service.refreshTopChats();
+    expect(fixture.outcomes.get(fixture.sentMessageId)?.outcome).toBe('unknown');
+    expect(fixture.repos.successfulMessagesRepo.count()).toBe(0);
+
+    await fixture.service.refreshTopChats();
+    expect(fixture.outcomes.get(fixture.sentMessageId)?.outcome).toBe('engaged');
+    expect(fixture.repos.successfulMessagesRepo.count()).toBe(1);
+
+    await fixture.service.refreshTopChats();
+    expect(fixture.repos.successfulMessagesRepo.count()).toBe(1);
+    expect(fixture.repos.messagesRepo.getByChatRoomId(fixture.roomId)).toMatchObject({
+      gotResponse: true,
+    });
+  });
+
+  it('backfills cached replies with missing outcomes once at startup', async () => {
+    const fixture = await createFixture({
+      cachedMessages: [
+        { sender: 'Me', text: 'Hello', timestampOffset: -5_000, isSelf: true },
+        { sender: 'alice', text: 'Thanks!', timestampOffset: -4_000, isSelf: false },
+        { sender: 'Me', text: 'What do you think?', timestampOffset: -3_000, isSelf: true },
+        { sender: 'alice', text: 'Tell me more', timestampOffset: -2_000, isSelf: false },
+      ],
+    });
+    fixture.repos.messagesRepo.markGotResponse(fixture.sentMessageId);
+    fixture.matrix.sync = vi.fn(async () => syncResponse('s2', [])) as typeof fixture.matrix.sync;
+
+    await fixture.service.runStartupSync();
+    expect(fixture.outcomes.get(fixture.sentMessageId)?.outcome).toBe('engaged');
+    expect(fixture.repos.successfulMessagesRepo.count()).toBe(1);
+
+    await fixture.service.runStartupSync();
+    expect(fixture.repos.successfulMessagesRepo.count()).toBe(1);
+  });
+});
+
+async function createFixture(options: {
+  cachedMessages?: Array<{
+    sender: string;
+    text: string;
+    timestampOffset: number;
+    isSelf: boolean;
+  }>;
+} = {}) {
+  const logger = createMockLogger();
+  const directory = await fs.mkdtemp(path.join(os.tmpdir(), 'chat-outcomes-'));
+  temporaryDirectories.push(directory);
+  const cacheFile = path.join(directory, 'chat-cache.json');
+  const db = createTestDb();
+  openDatabases.push(db);
+  const repos = createTestRepositories(db);
+  const outcomes = new ConversationOutcomesRepository(db);
+  const now = Date.now();
+  const roomId = '!alice:example.test';
+
+  repos.postsRepo.insert({
+    id: 'source-post', subreddit: 'example', author: 'alice', title: 'Source',
+    imageUrl: null, processedAt: now - 10_000,
+  });
+  repos.messagesRepo.insert({
+    postId: 'source-post', recipient: 'alice', subject: 'alice', message: 'Hello',
+    sentAt: now - 9_000, strategyId: 'direct', chatRoomId: roomId,
+  });
+  const sentMessageId = repos.messagesRepo.getByChatRoomId(roomId)!.id;
+
+  const cachedMessages = options.cachedMessages ?? [
+    { sender: 'Me', text: 'Hello', timestampOffset: -5_000, isSelf: true },
+  ];
+  await fs.writeFile(cacheFile, JSON.stringify({
+    version: 1,
+    savedAt: now - 60_000,
+    chats: [{
+      username: 'alice',
+      roomName: 'alice',
+      lastMessageTime: now - 5_000,
+      lastMessageSnippet: 'Hello',
+      isFromUs: true,
+      unread: false,
+      roomId,
+      messages: cachedMessages.map(({ timestampOffset, ...message }) => ({
+        ...message,
+        timestamp: now + timestampOffset,
+      })),
+    }],
+    nextBatch: 's1',
+    lastFullSyncAt: now - 120_000,
+    lastDeltaSyncAt: now - 60_000,
+    cooldown: null,
+  }), 'utf8');
+
+  const matrix = new MatrixClient(credentials, logger, { minRequestIntervalMs: 0 });
+  const cache = new ChatCache(logger, cacheFile);
+  const service = new ChatService(
+    matrix,
+    logger,
+    repos.successfulMessagesRepo,
+    undefined,
+    undefined,
+    undefined,
+    cache,
+    repos.messagesRepo,
+    undefined,
+    authenticatedAccountAuth,
+    outcomes,
+  );
+  await service.loadCache();
+
+  return { db, repos, outcomes, now, roomId, sentMessageId, matrix, service };
+}
+
+function syncResponse(nextBatch: string, events: ReturnType<typeof messageEvent>[]): MatrixSyncResponse {
+  return {
+    next_batch: nextBatch,
+    rooms: {
+      join: {
+        '!alice:example.test': {
+          timeline: { events },
+        },
+      },
+    },
+  };
+}
+
+function messageEvent(eventId: string, sender: string, timestamp: number, body: string) {
+  return {
+    event_id: eventId,
+    type: 'm.room.message' as const,
+    sender,
+    origin_server_ts: timestamp,
+    content: { body },
+  };
+}

tests/chat-service-progressive-backfill.test.ts

@@ -24,20 +24,21 @@ describe('ChatService progressive full backfill', () => {
     const repos = createTestRepositories(db);
 
     try {
+      const sentAt = Date.now();
       repos.postsRepo.insert({
         id: 'post-alice',
         subreddit: 'example',
         author: 'alice',
         title: 'Source post',
         imageUrl: null,
-        processedAt: 1_000,
+        processedAt: sentAt - 1_000,
       });
       repos.messagesRepo.insert({
         postId: 'post-alice',
         recipient: 'alice',
         subject: 'alice',
         message: 'Hello Alice',
-        sentAt: Date.now(),
+        sentAt,
         strategyId: 'direct',
         chatRoomId: '!alice:reddit.com',
       });
@@ -51,14 +52,14 @@ describe('ChatService progressive full backfill', () => {
             event_id: `$reply-${roomId}`,
             type: 'm.room.message',
             sender: roomId === '!alice:reddit.com' ? '@t2_alice:reddit.com' : '@t2_bob:reddit.com',
-            origin_server_ts: 2_000,
+            origin_server_ts: sentAt + 1_000,
             content: { body: 'Hi there' },
           },
           {
             event_id: `$first-${roomId}`,
             type: 'm.room.message',
             sender: '@me:reddit.com',
-            origin_server_ts: 1_000,
+            origin_server_ts: sentAt,
             content: { body: roomId === '!alice:reddit.com' ? 'Hello Alice' : 'Hello Bob' },
           },
         ],