diff --git a/packages/backend/src/ai/ai.constants.ts b/packages/backend/src/ai/ai.constants.ts index 65ede081..d4410d46 100644 --- a/packages/backend/src/ai/ai.constants.ts +++ b/packages/backend/src/ai/ai.constants.ts @@ -151,13 +151,12 @@ Do NOT include: Requirements: - Traits must be concise, concrete, and attributable to the user. -- Write each trait in third person using their Slack ID placeholder if provided context supports it. - No duplicates or near-duplicates. - Prefer quality over quantity. If only 4 strong traits exist, return 4. Output format: - Return ONLY a JSON array of strings. -- Example: ["JR-15 prefers TypeScript as his primary programming language", "JR-15 strongly dislikes Donald Trump"] +- Example: ["Prefers TypeScript as his primary programming language", "Strongly dislikes Donald Trump"] - If no strong traits are present, return []`; export const DAILY_MEMORY_JOB_CONCURRENCY = 50; diff --git a/packages/backend/src/ai/ai.controller.spec.ts b/packages/backend/src/ai/ai.controller.spec.ts index f0440310..4a55f071 100644 --- a/packages/backend/src/ai/ai.controller.spec.ts +++ b/packages/backend/src/ai/ai.controller.spec.ts @@ -12,10 +12,6 @@ const { generateText, generateImage, promptWithHistory, sendEphemeral, setCustom clearCustomPrompt: vi.fn().mockResolvedValue(true), })); -const { getAllTraitsForUser } = vi.hoisted(() => ({ - getAllTraitsForUser: vi.fn().mockResolvedValue([]), -})); - vi.mock('./ai.service', async () => ({ AIService: classMock(() => ({ generateText, @@ -32,12 +28,6 @@ vi.mock('../shared/services/web/web.service', async () => ({ })), })); -vi.mock('./trait/trait.persistence.service', async () => ({ - TraitPersistenceService: classMock(() => ({ - getAllTraitsForUser, - })), -})); - vi.mock('../shared/middleware/suppression', async () => ({ suppressedMiddleware: (_req: unknown, _res: unknown, next: () => void) => next(), })); @@ -61,7 +51,6 @@ describe('aiController', () => { vi.clearAllMocks(); setCustomPrompt.mockResolvedValue(true); clearCustomPrompt.mockResolvedValue(true); - getAllTraitsForUser.mockResolvedValue([]); }); it('handles /text', async () => { @@ -98,48 +87,6 @@ describe('aiController', () => { expect(sendEphemeral).toHaveBeenCalled(); }); - describe('/traits', () => { - it('returns immediate 200 and sends formatted traits ephemerally', async () => { - getAllTraitsForUser.mockResolvedValue([ - { - content: 'JR-15 prefers TypeScript as his programming language', - updatedAt: new Date('2026-04-01T00:00:00.000Z'), - }, - ]); - - await request(app).post('/traits').send({ user_id: 'U1', team_id: 'T1', channel_id: 'C1' }).expect(200); - - await Promise.resolve(); - - expect(getAllTraitsForUser).toHaveBeenCalledWith('U1', 'T1'); - expect(sendEphemeral).toHaveBeenCalledWith( - 'C1', - expect.stringContaining("Moonbeam's core traits about you:"), - 'U1', - ); - }); - - it('sends no-traits message when user has no traits', async () => { - getAllTraitsForUser.mockResolvedValue([]); - - await request(app).post('/traits').send({ user_id: 'U1', team_id: 'T1', channel_id: 'C1' }).expect(200); - - await Promise.resolve(); - - expect(sendEphemeral).toHaveBeenCalledWith('C1', "Moonbeam doesn't have any core traits about you yet.", 'U1'); - }); - - it('sends fallback error message when trait retrieval fails', async () => { - getAllTraitsForUser.mockRejectedValueOnce(new Error('db fail')); - - await request(app).post('/traits').send({ user_id: 'U1', team_id: 'T1', channel_id: 'C1' }).expect(200); - - await Promise.resolve(); - - expect(sendEphemeral).toHaveBeenCalledWith('C1', 'Sorry, something went wrong fetching your traits.', 'U1'); - }); - }); - describe('/set-prompt', () => { it('clears prompt when text is "clear"', async () => { const res = await request(app) diff --git a/packages/backend/src/ai/ai.controller.ts b/packages/backend/src/ai/ai.controller.ts index e40e6aa0..f13469bf 100644 --- a/packages/backend/src/ai/ai.controller.ts +++ b/packages/backend/src/ai/ai.controller.ts @@ -8,55 +8,15 @@ import { aiMiddleware } from './middleware/aiMiddleware'; import type { SlashCommandRequest } from '../shared/models/slack/slack-models'; import { logError } from '../shared/logger/error-logging'; import { logger } from '../shared/logger/logger'; -import { TraitPersistenceService } from './trait/trait.persistence.service'; export const aiController: Router = express.Router(); const webService = new WebService(); const aiService = new AIService(); -const traitPersistenceService = new TraitPersistenceService(); const aiLogger = logger.child({ module: 'AIController' }); aiController.use(suppressedMiddleware); -aiController.post('/traits', (req, res) => { - const { user_id, team_id, channel_id } = req.body; - - // Respond immediately — Slack requires a response within 3 seconds - res.status(200).send(''); - - void (async () => { - try { - const traits = await traitPersistenceService.getAllTraitsForUser(user_id, team_id); - - if (traits.length === 0) { - void webService.sendEphemeral(channel_id, "Moonbeam doesn't have any core traits about you yet.", user_id); - return; - } - - const formattedTraits = traits - .map((trait, index) => { - const date = new Date(trait.updatedAt).toLocaleDateString('en-US', { - month: 'short', - year: 'numeric', - }); - return `${index + 1}. "${trait.content}" (${date.toLowerCase()})`; - }) - .join('\n'); - - const message = `Moonbeam's core traits about you:\n${formattedTraits}`; - void webService.sendEphemeral(channel_id, message, user_id); - } catch (e) { - logError(aiLogger, 'Failed to fetch traits for /ai/traits command', e, { - userId: user_id, - teamId: team_id, - channelId: channel_id, - }); - void webService.sendEphemeral(channel_id, 'Sorry, something went wrong fetching your traits.', user_id); - } - })(); -}); - aiController.use(textMiddleware); aiController.use(aiMiddleware); diff --git a/packages/backend/src/ai/ai.service.spec.ts b/packages/backend/src/ai/ai.service.spec.ts index e6ea99a6..b9f6ad5b 100644 --- a/packages/backend/src/ai/ai.service.spec.ts +++ b/packages/backend/src/ai/ai.service.spec.ts @@ -5,6 +5,18 @@ import path from 'path'; import { AIService } from './ai.service'; import type { MessageWithName } from '../shared/models/message/message-with-name'; import { MOONBEAM_SLACK_ID } from './ai.constants'; +import { TraitService } from '../trait/trait.service'; + +const { getAllTraitsForUsers } = vi.hoisted(() => ({ + getAllTraitsForUsers: vi.fn().mockResolvedValue(new Map()), +})); + +vi.mock('../trait/trait.persistence.service', async () => ({ + TraitPersistenceService: classMock(() => ({ + getAllTraitsForUsers, + getAllTraitsForUser: vi.fn().mockResolvedValue([]), + })), +})); const buildAiService = (): AIService => { const ai = new AIService(); @@ -35,19 +47,6 @@ const buildAiService = (): AIService => { }, } as unknown as AIService['gemini']; - ai.memoryPersistenceService = { - getAllMemoriesForUsers: vi.fn().mockResolvedValue(new Map()), - getAllMemoriesForUser: vi.fn().mockResolvedValue([]), - saveMemories: vi.fn().mockResolvedValue([]), - reinforceMemory: vi.fn().mockResolvedValue(true), - deleteMemory: vi.fn().mockResolvedValue(true), - } as unknown as AIService['memoryPersistenceService']; - - ai.traitPersistenceService = { - getAllTraitsForUsers: vi.fn().mockResolvedValue(new Map()), - replaceTraitsForUser: vi.fn().mockResolvedValue([]), - } as unknown as AIService['traitPersistenceService']; - ai.historyService = { getHistory: vi.fn().mockResolvedValue([]), getHistoryWithOptions: vi.fn().mockResolvedValue([]), @@ -79,6 +78,8 @@ const buildAiService = (): AIService => { debug: vi.fn(), } as unknown as AIService['aiServiceLogger']; + ai.traitService = new TraitService(); + return ai; }; @@ -350,7 +351,10 @@ describe('AIService', () => { (aiService.historyService.getHistory as Mock).mockResolvedValue([ { name: 'Jane', slackId: 'U2', message: 'Hi there' }, ]); - (aiService.traitPersistenceService.getAllTraitsForUsers as Mock).mockResolvedValue( + const traitPersistenceService = ( + aiService.traitService as unknown as { traitPersistenceService: { getAllTraitsForUsers: unknown } } + ).traitPersistenceService; + (traitPersistenceService.getAllTraitsForUsers as Mock).mockResolvedValue( new Map([['U2', [{ slackId: 'U2', content: 'prefers typescript' }]]]), ); const createSpy = aiService.openAi.responses.create as Mock; @@ -552,7 +556,10 @@ describe('AIService', () => { (aiService.historyService.getHistoryWithOptions as Mock).mockResolvedValue([ { slackId: 'U2', name: 'Jane', message: 'hello' }, ]); - (aiService.traitPersistenceService.getAllTraitsForUsers as Mock).mockResolvedValue( + const traitPersistenceService = ( + aiService.traitService as unknown as { traitPersistenceService: { getAllTraitsForUsers: unknown } } + ).traitPersistenceService; + (traitPersistenceService.getAllTraitsForUsers as Mock).mockResolvedValue( new Map([['U2', [{ slackId: 'U2', content: 'dislikes donald trump' }]]]), ); const createSpy = aiService.openAi.responses.create as Mock; @@ -731,232 +738,6 @@ describe('AIService', () => { }); }); - describe('memory and trait helpers', () => { - type AiServicePrivate = typeof aiService & { - extractParticipantSlackIds: ( - messages: Array<{ slackId: string; name: string; message: string }>, - options: { includeSlackId?: string; excludeSlackIds?: string[] }, - ) => string[]; - formatTraitContext: ( - traits: Array<{ slackId: string; content: string }>, - messages: Array<{ slackId: string; name: string; message: string }>, - ) => string; - appendTraitContext: (base: string, context: string) => string; - fetchTraitContext: ( - participantIds: string[], - teamId: string, - messages: Array<{ slackId: string; name: string; message: string }>, - ) => Promise; - parseTraitExtractionResult: (raw: string | undefined) => string[]; - regenerateTraitsForUsers: (teamId: string, slackIds: string[]) => Promise; - extractMemories: (teamId: string, channelId: string, history: string, participantIds: string[]) => Promise; - }; - - it('extracts participant slack ids with include/exclude rules', () => { - const ids = (aiService as unknown as AiServicePrivate).extractParticipantSlackIds( - [ - { slackId: 'U1', name: 'A', message: 'm1' }, - { slackId: 'U2', name: 'B', message: 'm2' }, - { slackId: 'U2', name: 'B', message: 'm3' }, - ], - { includeSlackId: 'U3', excludeSlackIds: ['U1'] }, - ); - - expect(ids).toEqual(['U2', 'U3']); - }); - - it('formats trait context grouped by participant', () => { - const text = (aiService as unknown as AiServicePrivate).formatTraitContext( - [ - { slackId: 'U1', content: 'prefers typescript' }, - { slackId: 'U2', content: 'dislikes donald trump' }, - ], - [ - { slackId: 'U1', name: 'Alice', message: 'hi' }, - { slackId: 'U2', name: 'Bob', message: 'hello' }, - ], - ); - - expect(text).toContain('traits_context'); - expect(text).toContain('Alice'); - expect(text).toContain('prefers typescript'); - expect(text).toContain('Bob'); - }); - - it('returns base instructions when no context', () => { - const result = (aiService as unknown as AiServicePrivate).appendTraitContext('base', ''); - expect(result).toBe('base'); - }); - - it('inserts context before tag', () => { - const base = 'some instructions\n\nchecklist\n'; - const traitContext = '\ntest trait\n'; - const result = (aiService as unknown as AiServicePrivate).appendTraitContext(base, traitContext); - expect(result).toContain('test trait'); - expect(result.indexOf('traits_context')).toBeLessThan(result.indexOf('')); - }); - - it('fetches trait context end-to-end', async () => { - (aiService.traitPersistenceService.getAllTraitsForUsers as Mock).mockResolvedValue( - new Map([['U1', [{ id: 1, slackId: 'U1', content: 'prefers typescript' }]]]), - ); - - const context = await (aiService as unknown as AiServicePrivate).fetchTraitContext(['U1'], 'T1', [ - { slackId: 'U1', name: 'Alice', message: 'msg' }, - ]); - - expect(context).toContain('prefers typescript'); - }); - - it('parses, de-duplicates, and caps extracted traits', () => { - const traits = (aiService as unknown as AiServicePrivate).parseTraitExtractionResult( - JSON.stringify([...Array.from({ length: 12 }, (_, i) => `trait-${i}`), 'trait-1']), - ); - - expect(traits.length).toBe(10); - expect(new Set(traits).size).toBe(10); - }); - - it('returns empty trait list for malformed extraction payload', () => { - const traits = (aiService as unknown as AiServicePrivate).parseTraitExtractionResult('{nope'); - expect(traits).toEqual([]); - }); - - it('regenerates traits for users based on current memories', async () => { - (aiService.memoryPersistenceService.getAllMemoriesForUser as Mock) - .mockResolvedValueOnce([{ content: 'JR-15 loves TypeScript' }]) - .mockResolvedValueOnce([]); - (aiService.openAi.responses.create as Mock).mockResolvedValue({ - output: [ - { - type: 'message', - content: [{ type: 'output_text', text: JSON.stringify(['JR-15 prefers TypeScript']) }], - }, - ], - }); - - await (aiService as unknown as AiServicePrivate).regenerateTraitsForUsers('T1', ['U1', 'U2']); - - expect(aiService.traitPersistenceService.replaceTraitsForUser).toHaveBeenCalledWith('U1', 'T1', [ - 'JR-15 prefers TypeScript', - ]); - expect(aiService.traitPersistenceService.replaceTraitsForUser).toHaveBeenCalledWith('U2', 'T1', []); - }); - - it('extractMemories returns early when lock exists', async () => { - (aiService.redis.getExtractionLock as Mock).mockResolvedValue('1'); - const infoSpy = vi.spyOn(aiService.aiServiceLogger, 'info'); - - await (aiService as unknown as AiServicePrivate).extractMemories('T1', 'C1', 'history', ['U1']); - - expect(infoSpy).toHaveBeenCalled(); - }); - - it('extractMemories handles NONE response', async () => { - (aiService.redis.getExtractionLock as Mock).mockResolvedValue(null); - (aiService.openAi.responses.create as Mock).mockResolvedValue({ - output: [{ type: 'message', content: [{ type: 'output_text', text: 'NONE' }] }], - }); - - await (aiService as unknown as AiServicePrivate).extractMemories('T1', 'C1', 'history', ['U1']); - - expect(aiService.memoryPersistenceService.saveMemories).not.toHaveBeenCalled(); - expect(aiService.traitPersistenceService.replaceTraitsForUser).not.toHaveBeenCalled(); - }); - - it('extractMemories processes NEW, REINFORCE and EVOLVE modes', async () => { - (aiService.redis.getExtractionLock as Mock).mockResolvedValue(null); - (aiService.openAi.responses.create as Mock).mockResolvedValue({ - output: [ - { - type: 'message', - content: [ - { - type: 'output_text', - text: JSON.stringify([ - { slackId: 'U123ABC', content: 'new memory', mode: 'NEW' }, - { slackId: 'U123ABC', content: 'reinforce memory', mode: 'REINFORCE', existingMemoryId: 10 }, - { slackId: 'U123ABC', content: 'evolved memory', mode: 'EVOLVE', existingMemoryId: 11 }, - ]), - }, - ], - }, - ], - }); - - await (aiService as unknown as AiServicePrivate).extractMemories('T1', 'C1', 'history', ['U123ABC']); - - expect(aiService.memoryPersistenceService.saveMemories).toHaveBeenCalled(); - expect(aiService.memoryPersistenceService.reinforceMemory).toHaveBeenCalledWith(10); - expect(aiService.memoryPersistenceService.deleteMemory).toHaveBeenCalledWith(11); - expect(aiService.traitPersistenceService.replaceTraitsForUser).toHaveBeenCalled(); - }); - - it('extractMemories skips malformed extraction items', async () => { - (aiService.redis.getExtractionLock as Mock).mockResolvedValue(null); - const warnSpy = vi.spyOn(aiService.aiServiceLogger, 'warn'); - (aiService.openAi.responses.create as Mock).mockResolvedValue({ - output: [ - { - type: 'message', - content: [ - { - type: 'output_text', - text: JSON.stringify([ - { mode: 'NEW' }, - { slackId: 'invalid', content: 'x', mode: 'NEW' }, - { slackId: 'U123ABC', content: 'x', mode: 'UNKNOWN' }, - ]), - }, - ], - }, - ], - }); - - await (aiService as unknown as AiServicePrivate).extractMemories('T1', 'C1', 'history', ['U123ABC']); - - expect(warnSpy).toHaveBeenCalled(); - }); - }); - - describe('extractMemoriesForChannel', () => { - it('returns early when there are no messages in the last 24 hours', async () => { - (aiService.historyService.getLast24HoursForChannel as Mock).mockResolvedValue([]); - - await aiService.extractMemoriesForChannel('T1', 'C1'); - - expect(aiService.openAi.responses.create).not.toHaveBeenCalled(); - }); - - it('returns early when there are no non-Moonbeam participants', async () => { - (aiService.historyService.getLast24HoursForChannel as Mock).mockResolvedValue([ - { slackId: MOONBEAM_SLACK_ID, name: 'Moonbeam', message: 'Hello there' }, - ]); - - await aiService.extractMemoriesForChannel('T1', 'C1'); - - expect(aiService.openAi.responses.create).not.toHaveBeenCalled(); - }); - - it('calls extractMemories with formatted history when valid messages exist', async () => { - (aiService.historyService.getLast24HoursForChannel as Mock).mockResolvedValue([ - { slackId: 'U1', name: 'Alice', message: 'Hello' }, - { slackId: MOONBEAM_SLACK_ID, name: 'Moonbeam', message: 'Hi Alice' }, - ]); - (aiService.redis.getExtractionLock as Mock).mockResolvedValue('1'); - - await aiService.extractMemoriesForChannel('T1', 'C1'); - - expect(aiService.historyService.getLast24HoursForChannel).toHaveBeenCalledWith('T1', 'C1'); - }); - - it('propagates errors from getLast24HoursForChannel', async () => { - (aiService.historyService.getLast24HoursForChannel as Mock).mockRejectedValue(new Error('DB error')); - - await expect(aiService.extractMemoriesForChannel('T1', 'C1')).rejects.toThrow('DB error'); - }); - }); - describe('send helpers', () => { it('sendImage posts image block and fallback on slack error', async () => { const sendMock = aiService.webService.sendMessage as Mock; diff --git a/packages/backend/src/ai/ai.service.ts b/packages/backend/src/ai/ai.service.ts index 9f4181e4..dbaf1a6b 100644 --- a/packages/backend/src/ai/ai.service.ts +++ b/packages/backend/src/ai/ai.service.ts @@ -17,15 +17,9 @@ import { MAX_AI_REQUESTS_PER_DAY, REDPLOY_MOONBEAM_IMAGE_PROMPT, REDPLOY_MOONBEAM_TEXT_PROMPT, - GATE_MODEL, MOONBEAM_SLACK_ID, - MEMORY_EXTRACTION_PROMPT, - TRAIT_EXTRACTION_PROMPT, GPT_MODEL, } from './ai.constants'; -import { MemoryPersistenceService } from './memory/memory.persistence.service'; -import { TraitPersistenceService } from './trait/trait.persistence.service'; -import type { TraitWithSlackId } from '../shared/db/models/Trait'; import { logError } from '../shared/logger/error-logging'; import { logger } from '../shared/logger/logger'; import { SlackService } from '../shared/services/slack/slack.service'; @@ -41,13 +35,8 @@ import type { import type { Part } from '@google/genai'; import { GoogleGenAI } from '@google/genai'; import sharp from 'sharp'; - -interface ExtractionResult { - slackId: string; - content: string; - mode: 'NEW' | 'REINFORCE' | 'EVOLVE'; - existingMemoryId: number | null; -} +import { extractParticipantSlackIds } from './helpers/extractParticipantSlackIds'; +import { TraitService } from '../trait/trait.service'; interface ReleaseCommit { sha: string; @@ -106,8 +95,7 @@ export class AIService { webService = new WebService(); slackService = new SlackService(); slackPersistenceService = new SlackPersistenceService(); - memoryPersistenceService = new MemoryPersistenceService(); - traitPersistenceService = new TraitPersistenceService(); + traitService = new TraitService(); aiServiceLogger = logger.child({ module: 'AIService' }); public decrementDaiyRequests(userId: string, teamId: string): Promise { @@ -401,15 +389,15 @@ export class AIService { const customPrompt = await this.slackPersistenceService.getCustomPrompt(user_id, team_id); const normalizedCustomPrompt = customPrompt?.trim() || null; - const traitContext = await this.fetchTraitContext( - this.extractParticipantSlackIds(history, { includeSlackId: user_id }), + const traitContext = await this.traitService.fetchTraitContext( + extractParticipantSlackIds(history, { includeSlackId: user_id }), team_id, history, ); const baseInstructions = normalizedCustomPrompt ? `${normalizedCustomPrompt}\n\n${getHistoryInstructions(formattedHistory)}` : getHistoryInstructions(formattedHistory); - const systemInstructions = this.appendTraitContext(baseInstructions, traitContext); + const systemInstructions = this.traitService.appendTraitContext(baseInstructions, traitContext); return this.openAi.responses .create({ @@ -487,13 +475,12 @@ export class AIService { const customPrompt = userId ? await this.slackPersistenceService.getCustomPrompt(userId, teamId) : null; const normalizedCustomPrompt = customPrompt?.trim() || null; - // Fetch stable user traits instead of raw memories to reduce context size. - const participantSlackIds = this.extractParticipantSlackIds(historyMessages, { + const participantSlackIds = extractParticipantSlackIds(historyMessages, { excludeSlackIds: [MOONBEAM_SLACK_ID], }); - const traitContext = await this.fetchTraitContext(participantSlackIds, teamId, historyMessages); + const traitContext = await this.traitService.fetchTraitContext(participantSlackIds, teamId, historyMessages); const baseInstructions = normalizedCustomPrompt ?? MOONBEAM_SYSTEM_INSTRUCTIONS; - const systemInstructions = this.appendTraitContext(baseInstructions, traitContext); + const systemInstructions = this.traitService.appendTraitContext(baseInstructions, traitContext); const input = `${history}\n\n---\n[Tagged message to respond to]:\n${taggedMessage}`; @@ -533,82 +520,6 @@ export class AIService { }); } - private formatTraitContext(traits: TraitWithSlackId[], history: MessageWithName[]): string { - if (traits.length === 0) return ''; - - const nameMap = new Map(); - history.forEach((msg) => { - if (msg.slackId && msg.name) nameMap.set(msg.slackId, msg.name); - }); - - const grouped = new Map(); - for (const trait of traits) { - const slackId = trait.slackId || 'unknown'; - if (!grouped.has(slackId)) grouped.set(slackId, []); - grouped.get(slackId)!.push(trait); - } - - const lines = Array.from(grouped.entries()) - .map(([slackId, userTraits]) => { - const name = nameMap.get(slackId) || slackId; - const traitLines = userTraits.map((trait) => `"${trait.content}"`).join(', '); - return `- ${name}: ${traitLines}`; - }) - .join('\n'); - - return `\ncore beliefs and stable traits for people in this conversation:\n${lines}\n`; - } - - private extractParticipantSlackIds( - history: MessageWithName[], - options?: { includeSlackId?: string; excludeSlackIds?: string[] }, - ): string[] { - const excludeSet = new Set(options?.excludeSlackIds || []); - const ids = [ - ...new Set(history.filter((msg) => msg.slackId && !excludeSet.has(msg.slackId!)).map((msg) => msg.slackId!)), - ]; - if (options?.includeSlackId && !ids.includes(options.includeSlackId)) { - ids.push(options.includeSlackId); - } - return ids; - } - - private async fetchTraitContext( - participantSlackIds: string[], - teamId: string, - history: MessageWithName[], - ): Promise { - if (participantSlackIds.length === 0) return ''; - const traitsMap = await this.traitPersistenceService.getAllTraitsForUsers(participantSlackIds, teamId); - const traits = Array.from(traitsMap.values()).flat(); - return this.formatTraitContext(traits, history); - } - - private appendTraitContext(baseInstructions: string, memoryContext: string): string { - if (!memoryContext) return baseInstructions; - // Insert memory data before so the verification checklist remains the last thing the model sees - const verificationTag = ''; - const insertionPoint = baseInstructions.lastIndexOf(verificationTag); - if (insertionPoint !== -1) { - return `${baseInstructions.slice(0, insertionPoint)}${memoryContext}\n\n${baseInstructions.slice(insertionPoint)}`; - } - // Fallback for custom prompts that don't use the standard tag - return `${baseInstructions}\n\n${memoryContext}`; - } - - public async extractMemoriesForChannel(teamId: string, channelId: string): Promise { - const historyMessages = await this.historyService.getLast24HoursForChannel(teamId, channelId); - if (historyMessages.length === 0) return; - - const history = this.formatHistory(historyMessages); - const participantSlackIds = this.extractParticipantSlackIds(historyMessages, { - excludeSlackIds: [MOONBEAM_SLACK_ID], - }); - if (participantSlackIds.length === 0) return; - - await this.extractMemories(teamId, channelId, history, participantSlackIds); - } - private async updateMoonbeamProfilePhoto(imageBytes: Buffer): Promise { const profileImage = await sharp(imageBytes) .resize(512, 512, { fit: 'cover', position: 'centre' }) @@ -719,194 +630,6 @@ export class AIService { }; } - private parseTraitExtractionResult(raw: string | undefined): string[] { - if (!raw) { - return []; - } - - try { - const parsed: unknown = JSON.parse(raw.trim()); - if (!Array.isArray(parsed)) { - return []; - } - - return Array.from( - new Set( - parsed - .filter((value): value is string => typeof value === 'string') - .map((value) => value.trim()) - .filter((value) => value.length > 0), - ), - ).slice(0, 10); - } catch { - this.aiServiceLogger.warn(`Trait extraction returned malformed JSON: ${raw}`); - return []; - } - } - - private async processWithConcurrencyLimit( - items: T[], - concurrency: number, - worker: (item: T) => Promise, - ): Promise { - const effectiveConcurrency = Math.max(1, Math.min(concurrency, items.length)); - let nextIndex = 0; - - const runners = Array.from({ length: effectiveConcurrency }, async () => { - const currentIndex = nextIndex; - nextIndex += 1; - - if (currentIndex >= items.length) { - return; - } - - await worker(items[currentIndex]); - }); - - await Promise.all(runners); - } - - private async regenerateTraitsForUsers(teamId: string, slackIds: string[]): Promise { - const uniqueSlackIds = Array.from(new Set(slackIds.filter((id) => /^U[A-Z0-9]+$/.test(id)))); - if (uniqueSlackIds.length === 0) { - return; - } - - const traitRegenerationConcurrency = 3; - - await this.processWithConcurrencyLimit(uniqueSlackIds, traitRegenerationConcurrency, async (slackId) => { - const memories = await this.memoryPersistenceService.getAllMemoriesForUser(slackId, teamId); - if (memories.length === 0) { - await this.traitPersistenceService.replaceTraitsForUser(slackId, teamId, []); - return; - } - - const memoryText = memories.map((memory, index) => `${index + 1}. ${memory.content}`).join('\n'); - const input = `User Slack ID: ${slackId}\n\nMemories:\n${memoryText}`; - - const rawTraits = await this.openAi.responses - .create({ - model: GATE_MODEL, - instructions: TRAIT_EXTRACTION_PROMPT, - input, - }) - .then((response) => extractAndParseOpenAiResponse(response)) - .catch((error) => { - this.aiServiceLogger.warn(`Trait synthesis failed for ${slackId} in ${teamId}:`, error); - return undefined; - }); - - const traits = this.parseTraitExtractionResult(rawTraits); - await this.traitPersistenceService.replaceTraitsForUser(slackId, teamId, traits); - }); - } - - private async extractMemories( - teamId: string, - channelId: string, - conversationHistory: string, - participantSlackIds: string[], - ): Promise { - const locked = await this.redis.getExtractionLock(channelId, teamId); - if (locked) { - this.aiServiceLogger.info(`Extraction lock active for ${channelId}-${teamId}, skipping`); - return; - } - await this.redis.setExtractionLock(channelId, teamId); - - try { - const existingMemoriesMap = await this.memoryPersistenceService.getAllMemoriesForUsers( - participantSlackIds, - teamId, - ); - - const existingMemoriesText = - existingMemoriesMap.size > 0 - ? Array.from(existingMemoriesMap.entries()) - .map(([slackId, memories]) => { - const lines = memories.map((m) => ` [ID:${m.id}] "${m.content}"`).join('\n'); - return `${slackId}:\n${lines}`; - }) - .join('\n\n') - : '(no existing memories)'; - - const extractionInput = conversationHistory; - const prompt = MEMORY_EXTRACTION_PROMPT.replace('{existing_memories}', existingMemoriesText); - - const result = await this.openAi.responses - .create({ - model: GATE_MODEL, - instructions: prompt, - input: extractionInput, - }) - .then((x) => extractAndParseOpenAiResponse(x)); - - if (!result) { - this.aiServiceLogger.warn('Extraction returned no result'); - return; - } - - const trimmed = result.trim(); - if (trimmed === 'NONE' || trimmed === '"NONE"') return; - - let extractions: Array>; - try { - const parsed: unknown = JSON.parse(trimmed); - extractions = Array.isArray(parsed) ? parsed : [parsed]; - } catch { - this.aiServiceLogger.warn(`Extraction returned malformed JSON: ${trimmed}`); - return; - } - - const touchedUsers = new Set(); - - for (const extraction of extractions) { - if (!extraction.slackId || !extraction.content || !extraction.mode) { - this.aiServiceLogger.warn('Extraction missing required fields, skipping:', extraction); - continue; - } - - if (!/^U[A-Z0-9]+$/.test(extraction.slackId)) { - this.aiServiceLogger.warn(`Invalid slackId format: ${extraction.slackId}`); - continue; - } - - switch (extraction.mode) { - case 'NEW': - await this.memoryPersistenceService.saveMemories(extraction.slackId, teamId, [extraction.content]); - touchedUsers.add(extraction.slackId); - break; - - case 'REINFORCE': - if (extraction.existingMemoryId) { - await this.memoryPersistenceService.reinforceMemory(extraction.existingMemoryId); - touchedUsers.add(extraction.slackId); - } else { - this.aiServiceLogger.warn('REINFORCE extraction missing existingMemoryId, skipping'); - } - break; - - case 'EVOLVE': - if (extraction.existingMemoryId) { - await this.memoryPersistenceService.deleteMemory(extraction.existingMemoryId); - } - await this.memoryPersistenceService.saveMemories(extraction.slackId, teamId, [extraction.content]); - touchedUsers.add(extraction.slackId); - break; - - default: - this.aiServiceLogger.warn(`Unknown extraction mode: ${String(extraction.mode)}`); - } - } - - await this.regenerateTraitsForUsers(teamId, [...touchedUsers]); - - this.aiServiceLogger.info(`Extraction complete for ${channelId}: ${extractions.length} observations processed`); - } catch (e) { - this.aiServiceLogger.warn('Memory extraction failed:', e); - } - } - sendImage(image: string | undefined, userId: string, teamId: string, channel: string, text: string): void { if (image) { const blocks: KnownBlock[] = [ diff --git a/packages/backend/src/ai/daily-memory.job.spec.ts b/packages/backend/src/ai/daily-memory.job.spec.ts deleted file mode 100644 index 7897b759..00000000 --- a/packages/backend/src/ai/daily-memory.job.spec.ts +++ /dev/null @@ -1,138 +0,0 @@ -import { vi } from 'vitest'; -import { DailyMemoryJob } from './daily-memory.job'; -import type { SlackChannel } from '../shared/db/models/SlackChannel'; -import { DAILY_MEMORY_JOB_CONCURRENCY } from './ai.constants'; - -vi.mock('typeorm', async () => ({ - getRepository: vi.fn().mockReturnValue({ - find: vi.fn(), - }), - Entity: () => vi.fn(), - Column: () => vi.fn(), - PrimaryGeneratedColumn: () => vi.fn(), - ManyToOne: () => vi.fn(), - OneToMany: () => vi.fn(), - OneToOne: () => vi.fn(), - Unique: () => vi.fn(), - JoinColumn: () => vi.fn(), -})); - -vi.mock('../shared/logger/logger', async () => ({ - logger: { - child: vi.fn().mockReturnValue({ - info: vi.fn(), - warn: vi.fn(), - error: vi.fn(), - debug: vi.fn(), - }), - }, -})); - -import { getRepository } from 'typeorm'; - -const buildJob = (): DailyMemoryJob => { - const job = new DailyMemoryJob(); - - job.aiService = { - extractMemoriesForChannel: vi.fn().mockResolvedValue(undefined), - } as unknown as DailyMemoryJob['aiService']; - - (job as unknown as { jobLogger: Record }).jobLogger = { - info: vi.fn(), - warn: vi.fn(), - error: vi.fn(), - debug: vi.fn(), - }; - - return job; -}; - -describe('DailyMemoryJob', () => { - let job: DailyMemoryJob; - let findMock: Mock; - - beforeEach(() => { - vi.clearAllMocks(); - (getRepository as Mock).mockReturnValue({ - find: vi.fn(), - }); - job = buildJob(); - findMock = (getRepository as Mock)().find as Mock; - }); - - it('calls extractMemoriesForChannel for each channel', async () => { - const channels: Partial[] = [ - { channelId: 'C1', teamId: 'T1' }, - { channelId: 'C2', teamId: 'T1' }, - ]; - findMock.mockResolvedValue(channels); - - await job.run(); - - expect(job.aiService.extractMemoriesForChannel).toHaveBeenCalledTimes(2); - expect(job.aiService.extractMemoriesForChannel).toHaveBeenCalledWith('T1', 'C1'); - expect(job.aiService.extractMemoriesForChannel).toHaveBeenCalledWith('T1', 'C2'); - }); - - it('continues processing remaining channels when one fails', async () => { - const channels: Partial[] = [ - { channelId: 'C1', teamId: 'T1' }, - { channelId: 'C2', teamId: 'T1' }, - ]; - findMock.mockResolvedValue(channels); - (job.aiService.extractMemoriesForChannel as Mock) - .mockRejectedValueOnce(new Error('fail')) - .mockResolvedValueOnce(undefined); - - await job.run(); - - expect(job.aiService.extractMemoriesForChannel).toHaveBeenCalledTimes(2); - }); - - it('handles an empty channel list gracefully', async () => { - findMock.mockResolvedValue([]); - - await job.run(); - - expect(job.aiService.extractMemoriesForChannel).not.toHaveBeenCalled(); - }); - - it('processes all channels even when count exceeds the concurrency limit', async () => { - const channels: Partial[] = Array.from({ length: DAILY_MEMORY_JOB_CONCURRENCY + 2 }, (_, i) => ({ - channelId: `C${i}`, - teamId: 'T1', - })); - findMock.mockResolvedValue(channels); - - await job.run(); - - expect(job.aiService.extractMemoriesForChannel).toHaveBeenCalledTimes(channels.length); - }); - - it('processes channels with a sliding window so at most DAILY_MEMORY_JOB_CONCURRENCY run at once', async () => { - const channels: Partial[] = Array.from({ length: DAILY_MEMORY_JOB_CONCURRENCY + 1 }, (_, i) => ({ - channelId: `C${i}`, - teamId: 'T1', - })); - findMock.mockResolvedValue(channels); - - let maxInflight = 0; - let currentInflight = 0; - (job.aiService.extractMemoriesForChannel as Mock).mockImplementation( - () => - new Promise((resolve) => { - currentInflight++; - if (currentInflight > maxInflight) maxInflight = currentInflight; - setImmediate(() => { - currentInflight--; - resolve(); - }); - }), - ); - - await job.run(); - - expect(maxInflight).toBeLessThanOrEqual(DAILY_MEMORY_JOB_CONCURRENCY); - expect(job.aiService.extractMemoriesForChannel).toHaveBeenCalledTimes(channels.length); - }); -}); diff --git a/packages/backend/src/ai/daily-memory.job.ts b/packages/backend/src/ai/daily-memory.job.ts deleted file mode 100644 index c9737ae4..00000000 --- a/packages/backend/src/ai/daily-memory.job.ts +++ /dev/null @@ -1,61 +0,0 @@ -import { getRepository } from 'typeorm'; -import { SlackChannel } from '../shared/db/models/SlackChannel'; -import { AIService } from './ai.service'; -import { logger } from '../shared/logger/logger'; -import { DAILY_MEMORY_JOB_CONCURRENCY } from './ai.constants'; - -export class DailyMemoryJob { - aiService: AIService; - private jobLogger = logger.child({ module: 'DailyMemoryJob' }); - - constructor(aiService?: AIService) { - this.aiService = aiService ?? new AIService(); - } - - async run(): Promise { - this.jobLogger.info('Starting daily memory extraction job'); - - const channels = await getRepository(SlackChannel).find(); - - const results = await this.runWithConcurrencyLimit( - channels.map((channel) => () => this.aiService.extractMemoriesForChannel(channel.teamId, channel.channelId)), - DAILY_MEMORY_JOB_CONCURRENCY, - ); - - const failed = results - .map((result, index) => ({ result, index })) - .filter((item): item is { result: PromiseRejectedResult; index: number } => item.result.status === 'rejected'); - failed.forEach(({ result, index }) => { - const channel = channels[index]; - this.jobLogger.warn( - `Failed to extract memories for channel ${channel.channelId} (team ${channel.teamId}):`, - result.reason, - ); - }); - - const processed = results.length - failed.length; - this.jobLogger.info(`Daily memory extraction job complete: processed ${processed}/${channels.length} channels`); - } - - private async runWithConcurrencyLimit( - tasks: Array<() => Promise>, - concurrency: number, - ): Promise[]> { - const results: PromiseSettledResult[] = new Array(tasks.length); - let nextIndex = 0; - - const runNext = async (): Promise => { - while (nextIndex < tasks.length) { - const index = nextIndex++; - try { - results[index] = { status: 'fulfilled', value: await tasks[index]() }; - } catch (reason) { - results[index] = { status: 'rejected', reason }; - } - } - }; - - await Promise.all(Array.from({ length: Math.min(concurrency, tasks.length) }, () => runNext())); - return results; - } -} diff --git a/packages/backend/src/ai/helpers/extractParticipantSlackIds.ts b/packages/backend/src/ai/helpers/extractParticipantSlackIds.ts new file mode 100644 index 00000000..def12d38 --- /dev/null +++ b/packages/backend/src/ai/helpers/extractParticipantSlackIds.ts @@ -0,0 +1,15 @@ +import type { MessageWithName } from '../../shared/models/message/message-with-name'; + +export const extractParticipantSlackIds = ( + history: MessageWithName[], + options?: { includeSlackId?: string; excludeSlackIds?: string[] }, +): string[] => { + const excludeSet = new Set(options?.excludeSlackIds || []); + const ids = [ + ...new Set(history.filter((msg) => msg.slackId && !excludeSet.has(msg.slackId!)).map((msg) => msg.slackId!)), + ]; + if (options?.includeSlackId && !ids.includes(options.includeSlackId)) { + ids.push(options.includeSlackId); + } + return ids; +}; diff --git a/packages/backend/src/ai/memory/memory.job.spec.ts b/packages/backend/src/ai/memory/memory.job.spec.ts new file mode 100644 index 00000000..072eaefe --- /dev/null +++ b/packages/backend/src/ai/memory/memory.job.spec.ts @@ -0,0 +1,168 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { MemoryJob } from './memory.job'; + +describe('MemoryJob', () => { + let job: MemoryJob; + let memoryPersistenceService: { + getAllMemoriesForUsers: ReturnType; + saveMemories: ReturnType; + reinforceMemory: ReturnType; + deleteMemory: ReturnType; + }; + let redis: { + getValue: ReturnType; + setValueWithExpire: ReturnType; + }; + let traitJob: { + runForUsers: ReturnType; + }; + let jobLogger: { + info: ReturnType; + warn: ReturnType; + }; + let aiService: { + openAi: { + responses: { + create: ReturnType; + }; + }; + }; + + beforeEach(() => { + job = new MemoryJob({ formatHistory: vi.fn() } as never); + memoryPersistenceService = { + getAllMemoriesForUsers: vi.fn().mockResolvedValue(new Map()), + saveMemories: vi.fn().mockResolvedValue([]), + reinforceMemory: vi.fn().mockResolvedValue(true), + deleteMemory: vi.fn().mockResolvedValue(true), + }; + redis = { + getValue: vi.fn().mockResolvedValue(null), + setValueWithExpire: vi.fn().mockResolvedValue('OK'), + }; + traitJob = { + runForUsers: vi.fn().mockResolvedValue(undefined), + }; + jobLogger = { + info: vi.fn(), + warn: vi.fn(), + }; + aiService = { + openAi: { + responses: { + create: vi.fn(), + }, + }, + }; + + (job as never as { memoryPersistenceService: unknown }).memoryPersistenceService = memoryPersistenceService; + (job as never as { redis: unknown }).redis = redis; + (job as never as { traitJob: unknown }).traitJob = traitJob; + (job as never as { jobLogger: unknown }).jobLogger = jobLogger; + (job as never as { aiService: unknown }).aiService = aiService; + }); + + it('returns early when extraction lock exists', async () => { + redis.getValue.mockResolvedValue('1'); + + await ( + job as never as { + extractMemories: ( + teamId: string, + channelId: string, + conversationHistory: string, + participantSlackIds: string[], + ) => Promise; + } + ).extractMemories('T1', 'C1', 'history', ['U1']); + + expect(jobLogger.info).toHaveBeenCalled(); + }); + + it('does nothing when extractor returns NONE', async () => { + aiService.openAi.responses.create.mockResolvedValue({ + output: [{ type: 'message', content: [{ type: 'output_text', text: 'NONE' }] }], + }); + + await ( + job as never as { + extractMemories: ( + teamId: string, + channelId: string, + conversationHistory: string, + participantSlackIds: string[], + ) => Promise; + } + ).extractMemories('T1', 'C1', 'history', ['U1']); + + expect(memoryPersistenceService.saveMemories).not.toHaveBeenCalled(); + }); + + it('processes NEW, REINFORCE, and EVOLVE extraction modes', async () => { + aiService.openAi.responses.create.mockResolvedValue({ + output: [ + { + type: 'message', + content: [ + { + type: 'output_text', + text: JSON.stringify([ + { slackId: 'U123ABC', content: 'new memory', mode: 'NEW' }, + { slackId: 'U123ABC', content: 'reinforce memory', mode: 'REINFORCE', existingMemoryId: 10 }, + { slackId: 'U123ABC', content: 'evolved memory', mode: 'EVOLVE', existingMemoryId: 11 }, + ]), + }, + ], + }, + ], + }); + + await ( + job as never as { + extractMemories: ( + teamId: string, + channelId: string, + conversationHistory: string, + participantSlackIds: string[], + ) => Promise; + } + ).extractMemories('T1', 'C1', 'history', ['U123ABC']); + + expect(memoryPersistenceService.saveMemories).toHaveBeenCalled(); + expect(memoryPersistenceService.reinforceMemory).toHaveBeenCalledWith(10); + expect(memoryPersistenceService.deleteMemory).toHaveBeenCalledWith(11); + }); + + it('skips malformed extraction items and logs warnings', async () => { + aiService.openAi.responses.create.mockResolvedValue({ + output: [ + { + type: 'message', + content: [ + { + type: 'output_text', + text: JSON.stringify([ + { mode: 'NEW' }, + { slackId: 'invalid', content: 'x', mode: 'NEW' }, + { slackId: 'U123ABC', content: 'x', mode: 'UNKNOWN' }, + ]), + }, + ], + }, + ], + }); + + await ( + job as never as { + extractMemories: ( + teamId: string, + channelId: string, + conversationHistory: string, + participantSlackIds: string[], + ) => Promise; + } + ).extractMemories('T1', 'C1', 'history', ['U123ABC']); + + expect(jobLogger.warn).toHaveBeenCalled(); + }); +}); diff --git a/packages/backend/src/ai/memory/memory.job.ts b/packages/backend/src/ai/memory/memory.job.ts new file mode 100644 index 00000000..1fbb739f --- /dev/null +++ b/packages/backend/src/ai/memory/memory.job.ts @@ -0,0 +1,214 @@ +import { getRepository } from 'typeorm'; +import { SlackChannel } from '../../shared/db/models/SlackChannel'; +import { HistoryPersistenceService } from '../../shared/services/history.persistence.service'; +import { MemoryPersistenceService } from './memory.persistence.service'; +import { RedisPersistenceService } from '../../shared/services/redis.persistence.service'; +import { AIService } from '../ai.service'; +import { logger } from '../../shared/logger/logger'; +import { DAILY_MEMORY_JOB_CONCURRENCY, GATE_MODEL, MEMORY_EXTRACTION_PROMPT } from '../ai.constants'; +import { MOONBEAM_SLACK_ID } from '../ai.constants'; +import type OpenAI from 'openai'; +import { extractParticipantSlackIds } from '../helpers/extractParticipantSlackIds'; + +interface ExtractionResult { + slackId: string; + content: string; + mode: 'NEW' | 'REINFORCE' | 'EVOLVE'; + existingMemoryId: number | null; +} + +const extractAndParseOpenAiResponse = (response: OpenAI.Responses.Response): string | undefined => { + const textBlock = response.output.find((item) => item.type === 'message'); + if (textBlock && 'content' in textBlock) { + const outputText = textBlock.content.find((item) => item.type === 'output_text'); + return outputText?.text.trim(); + } + return undefined; +}; + +export class MemoryJob { + private historyService = new HistoryPersistenceService(); + private memoryPersistenceService = new MemoryPersistenceService(); + private redis = RedisPersistenceService.getInstance(); + private aiService: AIService; + private jobLogger = logger.child({ module: 'MemoryJob' }); + + constructor(aiService?: AIService) { + this.aiService = aiService ?? new AIService(); + } + + async run(): Promise { + this.jobLogger.info('Starting memory extraction job'); + + const channels = await getRepository(SlackChannel).find(); + + const results = await this.runWithConcurrencyLimit( + channels.map((channel) => () => this.extractMemoriesForChannel(channel.teamId, channel.channelId)), + DAILY_MEMORY_JOB_CONCURRENCY, + ); + + const failed = results + .map((result, index) => ({ result, index })) + .filter((item): item is { result: PromiseRejectedResult; index: number } => item.result.status === 'rejected'); + failed.forEach(({ result, index }) => { + const channel = channels[index]; + this.jobLogger.warn( + `Failed to extract memories for channel ${channel.channelId} (team ${channel.teamId}):`, + result.reason, + ); + }); + + const processed = results.length - failed.length; + this.jobLogger.info(`Memory extraction job complete: processed ${processed}/${channels.length} channels`); + } + + private async extractMemoriesForChannel(teamId: string, channelId: string): Promise { + const historyMessages = await this.historyService.getLast24HoursForChannel(teamId, channelId); + if (historyMessages.length === 0) return; + + const history = this.aiService.formatHistory(historyMessages); + const participantSlackIds = extractParticipantSlackIds(historyMessages, { + excludeSlackIds: [MOONBEAM_SLACK_ID], + }); + if (participantSlackIds.length === 0) return; + + await this.extractMemories(teamId, channelId, history, participantSlackIds); + } + + private async extractMemories( + teamId: string, + channelId: string, + conversationHistory: string, + participantSlackIds: string[], + ): Promise { + const lockKey = `memory_extraction_lock:${teamId}:${channelId}`; + const locked = await this.redis.getValue(lockKey); + if (locked) { + this.jobLogger.info(`Extraction lock active for ${channelId}-${teamId}, skipping`); + return; + } + await this.redis.setValueWithExpire(lockKey, 1, 'EX', 300000); // 5 Minutes + + try { + const existingMemoriesMap = await this.memoryPersistenceService.getAllMemoriesForUsers( + participantSlackIds, + teamId, + ); + const existingMemoriesText = + existingMemoriesMap.size > 0 + ? Array.from(existingMemoriesMap.entries()) + .map(([slackId, memories]) => { + const lines = memories.map((memory) => ` [ID:${memory.id}] "${memory.content}"`).join('\n'); + return `${slackId}:\n${lines}`; + }) + .join('\n\n') + : '(no existing memories)'; + + const prompt = MEMORY_EXTRACTION_PROMPT.replace('{existing_memories}', existingMemoriesText); + const result = await this.aiService.openAi.responses + .create({ + model: GATE_MODEL, + instructions: prompt, + input: conversationHistory, + }) + .then((response) => extractAndParseOpenAiResponse(response)); + + if (!result) { + this.jobLogger.warn('Extraction returned no result'); + return; + } + + const trimmed = result.trim(); + if (trimmed === 'NONE' || trimmed === '"NONE"') { + return; + } + + const extractions = this.parseExtractionResults(trimmed); + if (!extractions) { + return; + } + + const touchedUsers = new Set(); + for (const extraction of extractions) { + const wasTouched = await this.applyExtraction(teamId, extraction); + if (wasTouched && extraction.slackId) { + touchedUsers.add(extraction.slackId); + } + } + + this.jobLogger.info(`Extraction complete for ${channelId}: ${extractions.length} observations processed`); + } catch (error) { + this.jobLogger.warn('Memory extraction failed:', error); + } + } + + private parseExtractionResults(trimmedResult: string): Array> | null { + try { + const parsed: Array> = JSON.parse(trimmedResult); + if (Array.isArray(parsed)) { + return parsed; + } + this.jobLogger.warn(`Extraction returned JSON but it was not an array: ${trimmedResult}`); + return null; + } catch { + this.jobLogger.warn(`Extraction returned malformed JSON: ${trimmedResult}`); + return null; + } + } + + private async applyExtraction(teamId: string, extraction: Partial): Promise { + if (!extraction.slackId || !extraction.content || !extraction.mode) { + this.jobLogger.warn('Extraction missing required fields, skipping:', extraction); + return false; + } + + if (!/^U[A-Z0-9]+$/.test(extraction.slackId)) { + this.jobLogger.warn(`Invalid slackId format: ${extraction.slackId}`); + return false; + } + + switch (extraction.mode) { + case 'NEW': + await this.memoryPersistenceService.saveMemories(extraction.slackId, teamId, [extraction.content]); + return true; + case 'REINFORCE': + if (!extraction.existingMemoryId) { + this.jobLogger.warn('REINFORCE extraction missing existingMemoryId, skipping'); + return false; + } + await this.memoryPersistenceService.reinforceMemory(extraction.existingMemoryId); + return true; + case 'EVOLVE': + if (extraction.existingMemoryId) { + await this.memoryPersistenceService.deleteMemory(extraction.existingMemoryId); + } + await this.memoryPersistenceService.saveMemories(extraction.slackId, teamId, [extraction.content]); + return true; + default: + this.jobLogger.warn(`Unknown extraction mode: ${String(extraction.mode)}`); + return false; + } + } + + private async runWithConcurrencyLimit( + tasks: Array<() => Promise>, + concurrency: number, + ): Promise[]> { + const results: PromiseSettledResult[] = new Array(tasks.length); + let nextIndex = 0; + + const runNext = async (): Promise => { + while (nextIndex < tasks.length) { + const index = nextIndex++; + try { + results[index] = { status: 'fulfilled', value: await tasks[index]() }; + } catch (reason) { + results[index] = { status: 'rejected', reason }; + } + } + }; + + await Promise.all(Array.from({ length: Math.min(concurrency, tasks.length) }, () => runNext())); + return results; + } +} diff --git a/packages/backend/src/index.ts b/packages/backend/src/index.ts index accd7a02..a9b61749 100644 --- a/packages/backend/src/index.ts +++ b/packages/backend/src/index.ts @@ -2,7 +2,6 @@ import 'reflect-metadata'; // Necessary for TypeORM entities. import 'dotenv/config'; import bodyParser from 'body-parser'; import cors from 'cors'; -import cron from 'node-cron'; import type { Application } from 'express'; import express from 'express'; @@ -30,9 +29,7 @@ import { signatureVerificationMiddleware } from './shared/middleware/signatureVe import { WebService } from './shared/services/web/web.service'; import { logger } from './shared/logger/logger'; import { AIService } from './ai/ai.service'; -import { DailyMemoryJob } from './ai/daily-memory.job'; -import { FunFactJob } from './jobs/fun-fact.job'; -import { PricingJob } from './jobs/pricing.job'; +import { JobService } from './job.service'; import { resolveTypeOrmEntities } from './shared/db/typeorm-options'; import { portfolioController } from './portfolio/portfolio.controller'; import { hookController } from './hook/hook.controller'; @@ -40,6 +37,7 @@ import { searchController } from './search/search.controller'; import { authController } from './auth/auth.controller'; import { authMiddleware } from './shared/middleware/authMiddleware'; import { dashboardController } from './dashboard/dashboard.controller'; +import { traitController } from './trait/trait.controller'; const app: Application = express(); const PORT = process.env.PORT || 3000; @@ -109,14 +107,13 @@ app.use('/quote', quoteController); app.use('/rep', reactionController); app.use('/store', storeController); app.use('/summary', summaryController); +app.use('/traits', traitController); app.use('/walkie', walkieController); const slackService = new SlackService(); const webService = new WebService(); const aiService = new AIService(); -const dailyMemoryJob = new DailyMemoryJob(aiService); -const funFactJob = new FunFactJob(); -const pricingJob = new PricingJob(); +const jobService = new JobService(); const indexLogger = logger.child({ module: 'Index' }); const connectToDb = async (): Promise => { @@ -198,32 +195,7 @@ app.listen(PORT, (e?: Error) => { } else { indexLogger.info('Database connection established successfully.'); void aiService.redeployMoonbeam(); - cron.schedule( - '0 3 * * *', - () => { - void dailyMemoryJob.run(); - }, - { timezone: 'America/New_York' }, - ); - indexLogger.info('Daily memory extraction job scheduled daily at 3AM America/New_York time.'); - cron.schedule( - '0 9 * * *', - () => { - void funFactJob.run(); - }, - { timezone: 'America/New_York' }, - ); - indexLogger.info('Fun-fact job scheduled daily at 9AM America/New_York time.'); - cron.schedule( - '10 * * * *', - () => { - void pricingJob.run().catch((error) => { - indexLogger.error('Pricing job failed:', error); - }); - }, - { timezone: 'America/New_York' }, - ); - indexLogger.info('Pricing job scheduled every hour at minute 10 America/New_York time.'); + jobService.scheduleCronJobs(); } }) .catch((error) => { diff --git a/packages/backend/src/job.service.ts b/packages/backend/src/job.service.ts new file mode 100644 index 00000000..6869f382 --- /dev/null +++ b/packages/backend/src/job.service.ts @@ -0,0 +1,148 @@ +import cron from 'node-cron'; +import { MemoryJob } from './ai/memory/memory.job'; +import { FunFactJob } from './jobs/fun-fact.job'; +import { PricingJob } from './jobs/pricing.job'; +import { logger } from './shared/logger/logger'; +import { TraitJob } from './trait/trait.job'; + +export class JobService { + private memoryJob: MemoryJob; + private traitJob: TraitJob; + private funFactJob: FunFactJob; + private pricingJob: PricingJob; + private jobServiceLogger = logger.child({ module: 'JobService' }); + + constructor() { + this.memoryJob = new MemoryJob(); + this.traitJob = new TraitJob(); + this.funFactJob = new FunFactJob(); + this.pricingJob = new PricingJob(); + } + + /** + * Run the memory and trait jobs in sequence. + * Memory job runs first, then trait job runs only if memory job succeeds. + */ + async runMemoryAndTraitJobs(): Promise { + this.jobServiceLogger.info('Starting memory and trait job sequence'); + + try { + // Run memory job first + this.jobServiceLogger.info('Running memory job...'); + await this.memoryJob.run(); + this.jobServiceLogger.info('Memory job succeeded, proceeding with trait job'); + + // Run trait job only if memory job succeeds + this.jobServiceLogger.info('Running trait job...'); + await this.traitJob.run(); + this.jobServiceLogger.info('Trait job succeeded'); + + this.jobServiceLogger.info('Memory and trait job sequence completed successfully'); + } catch (error) { + this.jobServiceLogger.error('Memory and trait job sequence failed:', error); + throw error; + } + } + + /** + * Run the fun fact job + */ + async runFunFactJob(): Promise { + this.jobServiceLogger.info('Running fun fact job'); + try { + await this.funFactJob.run(); + this.jobServiceLogger.info('Fun fact job completed successfully'); + } catch (error) { + this.jobServiceLogger.error('Fun fact job failed:', error); + throw error; + } + } + + /** + * Run the pricing job + */ + async runPricingJob(): Promise { + this.jobServiceLogger.info('Running pricing job'); + try { + await this.pricingJob.run(); + this.jobServiceLogger.info('Pricing job completed successfully'); + } catch (error) { + this.jobServiceLogger.error('Pricing job failed:', error); + throw error; + } + } + + /** + * Run the memory job in isolation + */ + async runMemoryJob(): Promise { + this.jobServiceLogger.info('Running memory job in isolation'); + try { + await this.memoryJob.run(); + this.jobServiceLogger.info('Memory job completed successfully'); + } catch (error) { + this.jobServiceLogger.error('Memory job failed:', error); + throw error; + } + } + + /** + * Run the trait job in isolation + */ + async runTraitJob(): Promise { + this.jobServiceLogger.info('Running trait job in isolation'); + try { + await this.traitJob.run(); + this.jobServiceLogger.info('Trait job completed successfully'); + } catch (error) { + this.jobServiceLogger.error('Trait job failed:', error); + throw error; + } + } + + /** + * Schedule all cron jobs on startup. + * Memory and trait jobs run daily at 3AM. + * Fun fact job runs daily at 9AM. + * Pricing job runs every hour at minute 10. + */ + scheduleCronJobs(): void { + this.jobServiceLogger.info('Scheduling cron jobs'); + + // Memory and trait jobs: daily at 3AM America/New_York + cron.schedule( + '0 3 * * *', + () => { + this.runMemoryAndTraitJobs().catch((error) => { + this.jobServiceLogger.error('Memory and trait job sequence failed:', error); + }); + }, + { timezone: 'America/New_York' }, + ); + this.jobServiceLogger.info('Memory and trait job sequence scheduled daily at 3AM America/New_York time.'); + + // Fun fact job: daily at 9AM America/New_York + cron.schedule( + '0 9 * * *', + () => { + this.runFunFactJob().catch((error) => { + this.jobServiceLogger.error('Fun-fact job failed:', error); + }); + }, + { timezone: 'America/New_York' }, + ); + this.jobServiceLogger.info('Fun-fact job scheduled daily at 9AM America/New_York time.'); + + // Pricing job: every hour at minute 10 America/New_York + cron.schedule( + '10 * * * *', + () => { + this.runPricingJob().catch((error) => { + this.jobServiceLogger.error('Pricing job failed:', error); + }); + }, + { timezone: 'America/New_York' }, + ); + this.jobServiceLogger.info('Pricing job scheduled every hour at minute 10 America/New_York time.'); + } +} diff --git a/packages/backend/src/trait/trait.controller.spec.ts b/packages/backend/src/trait/trait.controller.spec.ts new file mode 100644 index 00000000..a0c2617c --- /dev/null +++ b/packages/backend/src/trait/trait.controller.spec.ts @@ -0,0 +1,35 @@ +import { vi } from 'vitest'; +import express from 'express'; +import request from 'supertest'; + +const { sendTraitsForUser } = vi.hoisted(() => ({ + sendTraitsForUser: vi.fn().mockResolvedValue(undefined), +})); + +vi.mock('./trait.service', async () => ({ + TraitService: classMock(() => ({ + sendTraitsForUser, + })), +})); + +vi.mock('../shared/middleware/suppression', async () => ({ + suppressedMiddleware: (_req: unknown, _res: unknown, next: () => void) => next(), +})); + +import { traitController } from './trait.controller'; + +describe('traitController', () => { + const app = express(); + app.use(express.json()); + app.use('/', traitController); + + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('handles /traits and responds immediately', async () => { + await request(app).post('/').send({ user_id: 'U1', team_id: 'T1', channel_id: 'C1' }).expect(200, ''); + + expect(sendTraitsForUser).toHaveBeenCalledWith('U1', 'T1', 'C1'); + }); +}); diff --git a/packages/backend/src/trait/trait.controller.ts b/packages/backend/src/trait/trait.controller.ts new file mode 100644 index 00000000..b455e0bb --- /dev/null +++ b/packages/backend/src/trait/trait.controller.ts @@ -0,0 +1,19 @@ +import type { Router } from 'express'; +import express from 'express'; +import { suppressedMiddleware } from '../shared/middleware/suppression'; +import { TraitService } from './trait.service'; + +export const traitController: Router = express.Router(); + +const traitService = new TraitService(); + +traitController.use(suppressedMiddleware); + +traitController.post('/', (req, res) => { + const { user_id, team_id, channel_id } = req.body; + + // Respond immediately — Slack requires a response within 3 seconds + res.status(200).send(''); + + void traitService.sendTraitsForUser(user_id, team_id, channel_id); +}); diff --git a/packages/backend/src/trait/trait.job.spec.ts b/packages/backend/src/trait/trait.job.spec.ts new file mode 100644 index 00000000..8650a8aa --- /dev/null +++ b/packages/backend/src/trait/trait.job.spec.ts @@ -0,0 +1,92 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { TraitJob } from './trait.job'; + +describe('TraitJob', () => { + let job: TraitJob; + let traitPersistenceService: { + replaceTraitsForUser: ReturnType; + }; + let memoryPersistenceService: { + getAllMemoriesForUser: ReturnType; + }; + let aiService: { + openAi: { + responses: { + create: ReturnType; + }; + }; + }; + let jobLogger: { + info: ReturnType; + warn: ReturnType; + error: ReturnType; + }; + + beforeEach(() => { + job = new TraitJob({} as never); + traitPersistenceService = { + replaceTraitsForUser: vi.fn().mockResolvedValue([]), + }; + memoryPersistenceService = { + getAllMemoriesForUser: vi.fn().mockResolvedValue([]), + }; + aiService = { + openAi: { + responses: { + create: vi.fn(), + }, + }, + }; + jobLogger = { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }; + + (job as never as { traitPersistenceService: unknown }).traitPersistenceService = traitPersistenceService; + (job as never as { memoryPersistenceService: unknown }).memoryPersistenceService = memoryPersistenceService; + (job as never as { aiService: unknown }).aiService = aiService; + (job as never as { jobLogger: unknown }).jobLogger = jobLogger; + }); + + it('parses, de-duplicates, and caps extracted traits', () => { + const traits = ( + job as never as { parseTraitExtractionResult: (raw: string) => string[] } + ).parseTraitExtractionResult( + JSON.stringify([...Array.from({ length: 12 }, (_, index) => `trait-${index}`), 'trait-1']), + ); + + expect(traits).toHaveLength(10); + expect(new Set(traits).size).toBe(10); + }); + + it('returns empty traits for malformed extraction payload', () => { + const traits = ( + job as never as { parseTraitExtractionResult: (raw: string) => string[] } + ).parseTraitExtractionResult('{bad'); + + expect(traits).toEqual([]); + expect(jobLogger.warn).toHaveBeenCalled(); + }); + + it('regenerates traits for users from memories', async () => { + memoryPersistenceService.getAllMemoriesForUser + .mockResolvedValueOnce([{ content: 'JR-15 loves TypeScript' }]) + .mockResolvedValueOnce([]); + + const synthesizeTraits = vi.fn().mockResolvedValue(JSON.stringify(['JR-15 prefers TypeScript'])); + + await ( + job as never as { + regenerateTraitsForUsers: ( + teamId: string, + slackIds: string[], + synthesizeTraits: (input: string) => Promise, + ) => Promise; + } + ).regenerateTraitsForUsers('T1', ['U1', 'U2'], synthesizeTraits); + + expect(traitPersistenceService.replaceTraitsForUser).toHaveBeenCalledWith('U1', 'T1', ['JR-15 prefers TypeScript']); + expect(traitPersistenceService.replaceTraitsForUser).toHaveBeenCalledWith('U2', 'T1', []); + }); +}); diff --git a/packages/backend/src/trait/trait.job.ts b/packages/backend/src/trait/trait.job.ts new file mode 100644 index 00000000..3ab82a88 --- /dev/null +++ b/packages/backend/src/trait/trait.job.ts @@ -0,0 +1,151 @@ +import { getRepository } from 'typeorm'; +import { TraitPersistenceService } from './trait.persistence.service'; +import { GATE_MODEL, TRAIT_EXTRACTION_PROMPT } from '../ai/ai.constants'; +import { AIService } from '../ai/ai.service'; +import { MemoryPersistenceService } from '../ai/memory/memory.persistence.service'; +import { SlackUser } from '../shared/db/models/SlackUser'; +import { logger } from '../shared/logger/logger'; + +export class TraitJob { + private traitPersistenceService = new TraitPersistenceService(); + private memoryPersistenceService = new MemoryPersistenceService(); + private aiService: AIService; + private jobLogger = logger.child({ module: 'TraitJob' }); + + constructor(aiService?: AIService) { + this.aiService = aiService ?? new AIService(); + } + + async run(): Promise { + this.jobLogger.info('Starting trait regeneration job'); + + try { + // Get all users + const users = await getRepository(SlackUser).find(); + + if (users.length === 0) { + this.jobLogger.info('No users found for trait regeneration'); + return; + } + + // Extract all team IDs to regenerate traits for + const teamIds = Array.from(new Set(users.map((u) => u.teamId))); + + let totalUsers = 0; + let processedUsers = 0; + + for (const teamId of teamIds) { + const teamUsers = users.filter((u) => u.teamId === teamId); + totalUsers += teamUsers.length; + + const slackIds = teamUsers.map((u) => u.slackId); + + try { + await this.runForUsers(teamId, slackIds); + processedUsers += slackIds.length; + } catch (error) { + this.jobLogger.warn(`Failed to regenerate traits for team ${teamId}:`, error); + } + } + + this.jobLogger.info(`Trait regeneration job complete: processed ${processedUsers}/${totalUsers} users`); + } catch (error) { + this.jobLogger.error('Trait regeneration job failed:', error); + throw error; + } + } + + async runForUsers(teamId: string, slackIds: string[]): Promise { + await this.regenerateTraitsForUsers(teamId, slackIds, async (input) => { + return this.aiService.openAi.responses + .create({ + model: GATE_MODEL, + instructions: TRAIT_EXTRACTION_PROMPT, + input, + user: `trait-job-${teamId}`, + }) + .then((response) => { + const textBlock = response.output.find((item) => item.type === 'message'); + if (textBlock && 'content' in textBlock) { + const outputText = textBlock.content.find((item) => item.type === 'output_text'); + return outputText?.text.trim(); + } + return undefined; + }); + }); + } + + private parseTraitExtractionResult(raw: string | undefined): string[] { + if (!raw) { + return []; + } + + try { + const parsed: unknown = JSON.parse(raw.trim()); + if (!Array.isArray(parsed)) { + return []; + } + + return Array.from( + new Set( + parsed + .filter((value): value is string => typeof value === 'string') + .map((value) => value.trim()) + .filter((value) => value.length > 0), + ), + ).slice(0, 10); + } catch { + this.jobLogger.warn(`Trait extraction returned malformed JSON: ${raw}`); + return []; + } + } + + private async regenerateTraitsForUsers( + teamId: string, + slackIds: string[], + synthesizeTraits: (input: string) => Promise, + ): Promise { + const uniqueSlackIds = Array.from(new Set(slackIds.filter((id) => /^U[A-Z0-9]+$/.test(id)))); + if (uniqueSlackIds.length === 0) { + return; + } + + await this.processWithConcurrencyLimit(uniqueSlackIds, 3, async (slackId) => { + const memories = await this.memoryPersistenceService.getAllMemoriesForUser(slackId, teamId); + if (memories.length === 0) { + await this.traitPersistenceService.replaceTraitsForUser(slackId, teamId, []); + return; + } + + const memoryText = memories.map((memory, index) => `${index + 1}. ${memory.content}`).join('\n'); + const input = `User Slack ID: ${slackId}\n\nMemories:\n${memoryText}`; + + const rawTraits = await synthesizeTraits(input).catch((error) => { + this.jobLogger.warn(`Trait synthesis failed for ${slackId} in ${teamId}:`, error); + return undefined; + }); + + const traits = this.parseTraitExtractionResult(rawTraits); + await this.traitPersistenceService.replaceTraitsForUser(slackId, teamId, traits); + }); + } + + private async processWithConcurrencyLimit( + items: T[], + concurrency: number, + worker: (item: T) => Promise, + ): Promise { + const effectiveConcurrency = Math.max(1, Math.min(concurrency, items.length)); + let nextIndex = 0; + + const runners = Array.from({ length: effectiveConcurrency }, async () => { + while (nextIndex < items.length) { + const currentIndex = nextIndex; + nextIndex += 1; + await worker(items[currentIndex]); + } + }); + + await Promise.all(runners); + } +} diff --git a/packages/backend/src/ai/trait/trait.persistence.service.spec.ts b/packages/backend/src/trait/trait.persistence.service.spec.ts similarity index 97% rename from packages/backend/src/ai/trait/trait.persistence.service.spec.ts rename to packages/backend/src/trait/trait.persistence.service.spec.ts index 8caf1ea6..840c85cc 100644 --- a/packages/backend/src/ai/trait/trait.persistence.service.spec.ts +++ b/packages/backend/src/trait/trait.persistence.service.spec.ts @@ -1,8 +1,8 @@ import { vi } from 'vitest'; import { getRepository } from 'typeorm'; import { TraitPersistenceService } from './trait.persistence.service'; -import { SlackUser } from '../../shared/db/models/SlackUser'; -import { Trait } from '../../shared/db/models/Trait'; +import { SlackUser } from '../shared/db/models/SlackUser'; +import { Trait } from '../shared/db/models/Trait'; vi.mock('typeorm', async () => ({ getRepository: vi.fn(), diff --git a/packages/backend/src/ai/trait/trait.persistence.service.ts b/packages/backend/src/trait/trait.persistence.service.ts similarity index 91% rename from packages/backend/src/ai/trait/trait.persistence.service.ts rename to packages/backend/src/trait/trait.persistence.service.ts index 3da79b45..39cac408 100644 --- a/packages/backend/src/ai/trait/trait.persistence.service.ts +++ b/packages/backend/src/trait/trait.persistence.service.ts @@ -1,9 +1,8 @@ import { getRepository } from 'typeorm'; -import type { TraitWithSlackId } from '../../shared/db/models/Trait'; -import { Trait } from '../../shared/db/models/Trait'; -import { SlackUser } from '../../shared/db/models/SlackUser'; -import { logError } from '../../shared/logger/error-logging'; -import { logger } from '../../shared/logger/logger'; +import { logger } from '../shared/logger/logger'; +import { SlackUser } from '../shared/db/models/SlackUser'; +import { Trait, type TraitWithSlackId } from '../shared/db/models/Trait'; +import { logError } from '../shared/logger/error-logging'; const MAX_TRAITS_PER_USER = 10; diff --git a/packages/backend/src/trait/trait.service.spec.ts b/packages/backend/src/trait/trait.service.spec.ts new file mode 100644 index 00000000..3e319dd7 --- /dev/null +++ b/packages/backend/src/trait/trait.service.spec.ts @@ -0,0 +1,104 @@ +import { vi } from 'vitest'; +import { TraitService } from './trait.service'; + +const { getAllTraitsForUser, getAllTraitsForUsers, sendEphemeral } = vi.hoisted(() => ({ + getAllTraitsForUser: vi.fn().mockResolvedValue([]), + getAllTraitsForUsers: vi.fn().mockResolvedValue(new Map()), + sendEphemeral: vi.fn().mockResolvedValue({ ok: true }), +})); + +vi.mock('./trait.persistence.service', async () => ({ + TraitPersistenceService: classMock(() => ({ + getAllTraitsForUser, + getAllTraitsForUsers, + })), +})); + +vi.mock('../shared/services/web/web.service', async () => ({ + WebService: classMock(() => ({ + sendEphemeral, + })), +})); + +describe('TraitService', () => { + let service: TraitService; + + beforeEach(() => { + vi.clearAllMocks(); + service = new TraitService(); + }); + + it('sends formatted traits when they exist', async () => { + getAllTraitsForUser.mockResolvedValue([ + { + content: 'JR-15 prefers TypeScript as his programming language', + updatedAt: new Date('2026-04-01T00:00:00.000Z'), + }, + ]); + + await service.sendTraitsForUser('U1', 'T1', 'C1'); + + expect(getAllTraitsForUser).toHaveBeenCalledWith('U1', 'T1'); + expect(sendEphemeral).toHaveBeenCalledWith( + 'C1', + expect.stringContaining("Moonbeam's core traits about you:"), + 'U1', + ); + }); + + it('sends no-traits message when user has no traits', async () => { + getAllTraitsForUser.mockResolvedValue([]); + + await service.sendTraitsForUser('U1', 'T1', 'C1'); + + expect(sendEphemeral).toHaveBeenCalledWith('C1', "Moonbeam doesn't have any core traits about you yet.", 'U1'); + }); + + it('sends fallback error message when trait retrieval fails', async () => { + getAllTraitsForUser.mockRejectedValueOnce(new Error('db fail')); + + await service.sendTraitsForUser('U1', 'T1', 'C1'); + + expect(sendEphemeral).toHaveBeenCalledWith('C1', 'Sorry, something went wrong fetching your traits.', 'U1'); + }); + + it('formats trait context grouped by participant name', () => { + const text = service.formatTraitContext( + [ + { slackId: 'U1', content: 'prefers typescript' } as never, + { slackId: 'U2', content: 'dislikes donald trump' } as never, + ], + [ + { slackId: 'U1', name: 'Alice', message: 'hi' } as never, + { slackId: 'U2', name: 'Bob', message: 'hello' } as never, + ], + ); + + expect(text).toContain('traits_context'); + expect(text).toContain('Alice'); + expect(text).toContain('prefers typescript'); + expect(text).toContain('Bob'); + }); + + it('returns base instructions when there is no trait context', () => { + expect(service.appendTraitContext('base', '')).toBe('base'); + }); + + it('inserts trait context before verification section', () => { + const base = 'instructions\n\nchecklist\n'; + const context = '\ntest trait\n'; + + const result = service.appendTraitContext(base, context); + + expect(result).toContain('test trait'); + expect(result.indexOf('traits_context')).toBeLessThan(result.indexOf('')); + }); + + it('fetches trait context from persistence layer', async () => { + getAllTraitsForUsers.mockResolvedValue(new Map([['U1', [{ slackId: 'U1', content: 'prefers typescript' }]]])); + + const context = await service.fetchTraitContext(['U1'], 'T1', [{ slackId: 'U1', name: 'Alice' } as never]); + + expect(context).toContain('prefers typescript'); + }); +}); diff --git a/packages/backend/src/trait/trait.service.ts b/packages/backend/src/trait/trait.service.ts new file mode 100644 index 00000000..3f4eb029 --- /dev/null +++ b/packages/backend/src/trait/trait.service.ts @@ -0,0 +1,93 @@ +import { WebService } from '../shared/services/web/web.service'; +import { logError } from '../shared/logger/error-logging'; +import { logger } from '../shared/logger/logger'; +import type { MessageWithName } from '../shared/models/message/message-with-name'; +import type { TraitWithSlackId } from '../shared/db/models/Trait'; +import { TraitPersistenceService } from './trait.persistence.service'; + +export class TraitService { + private readonly traitPersistenceService = new TraitPersistenceService(); + private readonly webService = new WebService(); + private readonly traitLogger = logger.child({ module: 'TraitService' }); + + public async sendTraitsForUser(userId: string, teamId: string, channelId: string): Promise { + try { + const traits = await this.traitPersistenceService.getAllTraitsForUser(userId, teamId); + + if (traits.length === 0) { + await this.webService.sendEphemeral(channelId, "Moonbeam doesn't have any core traits about you yet.", userId); + return; + } + + const formattedTraits = traits + .map((trait, index) => { + const date = new Date(trait.updatedAt).toLocaleDateString('en-US', { + month: 'short', + year: 'numeric', + }); + return `${index + 1}. "${trait.content}" (${date.toLowerCase()})`; + }) + .join('\n'); + + const message = `Moonbeam's core traits about you:\n${formattedTraits}`; + await this.webService.sendEphemeral(channelId, message, userId); + } catch (e) { + logError(this.traitLogger, 'Failed to fetch traits for /traits command', e, { + userId, + teamId, + channelId, + }); + await this.webService.sendEphemeral(channelId, 'Sorry, something went wrong fetching your traits.', userId); + } + } + + public formatTraitContext(traits: TraitWithSlackId[], history: MessageWithName[]): string { + if (traits.length === 0) return ''; + + const nameMap = new Map(); + history.forEach((msg) => { + if (msg.slackId && msg.name) nameMap.set(msg.slackId, msg.name); + }); + + const grouped = new Map(); + for (const trait of traits) { + const slackId = trait.slackId || 'unknown'; + if (!grouped.has(slackId)) grouped.set(slackId, []); + grouped.get(slackId)!.push(trait); + } + + const lines = Array.from(grouped.entries()) + .map(([slackId, userTraits]) => { + const name = nameMap.get(slackId) || slackId; + const traitLines = userTraits.map((trait) => `"${trait.content}"`).join(', '); + return `- ${name}: ${traitLines}`; + }) + .join('\n'); + + return `\ncore beliefs and stable traits for people in this conversation:\n${lines}\n`; + } + + public async fetchTraitContext( + participantSlackIds: string[], + teamId: string, + history: MessageWithName[], + ): Promise { + if (participantSlackIds.length === 0) return ''; + const traitsMap = await this.traitPersistenceService.getAllTraitsForUsers(participantSlackIds, teamId); + const traits = Array.from(traitsMap.values()).flat(); + return this.formatTraitContext(traits, history); + } + + public appendTraitContext(baseInstructions: string, traitContext: string): string { + if (!traitContext) return baseInstructions; + + // Insert trait data before so the checklist remains the last thing the model sees. + const verificationTag = ''; + const insertionPoint = baseInstructions.lastIndexOf(verificationTag); + if (insertionPoint !== -1) { + return `${baseInstructions.slice(0, insertionPoint)}${traitContext}\n\n${baseInstructions.slice(insertionPoint)}`; + } + + return `${baseInstructions}\n\n${traitContext}`; + } +}