import { Injectable, Logger } from '@nestjs/common'; import { createQueue, type QueueHandle } from '@mosaic/queue'; const SESSION_SYSTEM_KEY = (sessionId: string) => `mosaic:session:${sessionId}:system`; const SESSION_SYSTEM_FRAGMENTS_KEY = (sessionId: string) => `mosaic:session:${sessionId}:system:fragments`; const SYSTEM_OVERRIDE_TTL_SECONDS = 604800; // 7 days interface OverrideFragment { text: string; addedAt: number; } @Injectable() export class SystemOverrideService { private readonly logger = new Logger(SystemOverrideService.name); private readonly handle: QueueHandle; constructor() { this.handle = createQueue(); } async set(sessionId: string, override: string): Promise { // Load existing fragments const existing = await this.handle.redis.get(SESSION_SYSTEM_FRAGMENTS_KEY(sessionId)); const fragments: OverrideFragment[] = existing ? (JSON.parse(existing) as OverrideFragment[]) : []; // Append new fragment fragments.push({ text: override, addedAt: Date.now() }); // Condense fragments into one coherent override const texts = fragments.map((f) => f.text); const condensed = await this.condenseOverrides(texts); // Store both: fragments array and condensed result const pipeline = this.handle.redis.pipeline(); pipeline.setex( SESSION_SYSTEM_FRAGMENTS_KEY(sessionId), SYSTEM_OVERRIDE_TTL_SECONDS, JSON.stringify(fragments), ); pipeline.setex(SESSION_SYSTEM_KEY(sessionId), SYSTEM_OVERRIDE_TTL_SECONDS, condensed); await pipeline.exec(); this.logger.debug( `Set system override for session ${sessionId} (${fragments.length} fragment(s), TTL=${SYSTEM_OVERRIDE_TTL_SECONDS}s)`, ); } async get(sessionId: string): Promise { return this.handle.redis.get(SESSION_SYSTEM_KEY(sessionId)); } async renew(sessionId: string): Promise { const pipeline = this.handle.redis.pipeline(); pipeline.expire(SESSION_SYSTEM_KEY(sessionId), SYSTEM_OVERRIDE_TTL_SECONDS); pipeline.expire(SESSION_SYSTEM_FRAGMENTS_KEY(sessionId), SYSTEM_OVERRIDE_TTL_SECONDS); await pipeline.exec(); } async clear(sessionId: string): Promise { await this.handle.redis.del( SESSION_SYSTEM_KEY(sessionId), SESSION_SYSTEM_FRAGMENTS_KEY(sessionId), ); this.logger.debug(`Cleared system override for session ${sessionId}`); } /** * Merge an array of override fragments into one coherent string. * If only one fragment exists, returns it as-is. * For multiple fragments, calls Haiku to produce a merged instruction. * Falls back to newline concatenation if the LLM call fails. */ async condenseOverrides(fragments: string[]): Promise { if (fragments.length === 0) return ''; if (fragments.length === 1) return fragments[0]!; const numbered = fragments.map((f, i) => `${i + 1}. ${f}`).join('\n'); const prompt = `Merge these system prompt instructions into one coherent paragraph. ` + `If instructions conflict, favor the most recently added (last in the list). ` + `Be concise — output only the merged instruction, nothing else.\n\n` + `Instructions (oldest first):\n${numbered}`; const apiKey = process.env['ANTHROPIC_API_KEY']; if (!apiKey) { this.logger.warn('ANTHROPIC_API_KEY not set — falling back to newline concatenation'); return fragments.join('\n'); } try { const response = await fetch('https://api.anthropic.com/v1/messages', { method: 'POST', headers: { 'Content-Type': 'application/json', 'x-api-key': apiKey, 'anthropic-version': '2023-06-01', }, body: JSON.stringify({ model: 'claude-haiku-4-5-20251001', max_tokens: 1024, messages: [{ role: 'user', content: prompt }], }), }); if (!response.ok) { const errorText = await response.text(); throw new Error(`Anthropic API error ${response.status}: ${errorText}`); } const data = (await response.json()) as { content: Array<{ type: string; text: string }>; }; const textBlock = data.content.find((c) => c.type === 'text'); if (!textBlock) { throw new Error('No text block in Anthropic response'); } return textBlock.text.trim(); } catch (err) { this.logger.error( `Condensation LLM call failed — falling back to newline concatenation: ${String(err)}`, ); return fragments.join('\n'); } } }