diff --git a/src/services/chat/service.ts b/src/services/chat/service.ts index 65a568a..34d70bb 100644 --- a/src/services/chat/service.ts +++ b/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) { - 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); - } - } - } catch (err) { - this.logger.warn('[chat] failed to associate reply with performance context:', err); - } + /** + * 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; + } 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)`); } } diff --git a/tests/chat-service-outcome-ingestion.test.ts b/tests/chat-service-outcome-ingestion.test.ts new file mode 100644 index 0000000..96c4214 --- /dev/null +++ b/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> = []; + +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[]): 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 }, + }; +} diff --git a/tests/chat-service-progressive-backfill.test.ts b/tests/chat-service-progressive-backfill.test.ts index 0072fbb..a45442b 100644 --- a/tests/chat-service-progressive-backfill.test.ts +++ b/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' }, }, ],