shadow_brain_core/compaction.py

"""
🔱 Imperial Compaction Engine (지능형 컨텍스트 압축 엔진)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Claw Code의 'Compaction' 기술을 쉐도우 브레인 코어에 최적화하여 이식한 모듈입니다.
대화가 길어질 때 핵심 컨텍스트(파일, TODO, 타임라인)를 보존하며 토큰을 절약합니다.
"""

import os
import json
import time
import re
from typing import List, Dict, Any, Optional

class CompactionEngine:
    """대화 내역을 분석하고 지능적으로 압축하는 핵심 엔진."""
    
    PREAMBLE = "이 세션은 이전 대화에서 컨텍스트가 부족하여 요약본으로부터 이어지는 내용입니다.\n\n"
    
    def __init__(self, project_root: str):
        self.project_root = project_root
        self.settings_file = os.path.join(project_root, "docker_data", "shared_workspace", "JarvisRun", "JARVIS_SHADOW_SETTINGS.json")
        
    def _get_settings(self) -> Dict[str, Any]:
        """설정 파일에서 핫-로드(Hot-Reload)로 설정을 읽어옵니다."""
        defaults = {
            "compaction_enabled": True,
            "compaction_threshold": 50, # 메시지 개수 기준
            "preserve_recent": 5        # 압축 시 보존할 최신 메시지 수
        }
        if os.path.exists(self.settings_file):
            try:
                with open(self.settings_file, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                # merge with defaults
                for k, v in defaults.items():
                    if k not in data: data[k] = v
                return data
            except: pass
        return defaults

    def should_compact(self, messages: List[Dict[str, Any]]) -> bool:
        """압축이 필요한 시점인지 판단합니다."""
        settings = self._get_settings()
        if not settings.get("compaction_enabled", True):
            return False
            
        threshold = int(settings.get("compaction_threshold", 50))
        return len(messages) >= threshold

    def extract_metadata(self, messages: List[Dict[str, Any]]) -> Dict[str, Any]:
        """대화 내역에서 핵심 메타데이터(파일, TODO, 타임라인, 감정)를 추출합니다."""
        files = set()
        todos = []
        timeline = []
        emotions = {} # sender별 주로 표현된 감정
        
        # 파일 확장자 정규식
        file_pattern = re.compile(r'[\w/._-]+\.(?:py|rs|dart|js|ts|md|html|css|json|yaml|yml|sh|ps1|bat)')
        
        for msg in messages:
            content = msg.get('text', '') or ''
            role = msg.get('relayGuardian') or msg.get('sender_name') or msg.get('sender_id') or 'unknown'
            emotion = msg.get('emotion')
            
            # 1. 파일 경로 추출
            found_files = file_pattern.findall(content)
            for f in found_files:
                if len(f) > 3: files.add(f)
            
            # 2. TODO/Next Step 추론
            lower_content = content.lower()
            if any(kw in lower_content for kw in ['todo', '다음', 'next', '할 일', 'pending', 'remain']):
                lines = [line.strip() for line in content.split('\n') if line.strip()]
                for line in lines:
                    if any(kw in line.lower() for kw in ['todo', '다음', 'next', '할 일']):
                        todos.append(line[:150] + "..." if len(line) > 150 else line)
            
            # 3. 감정 추적
            if emotion:
                if role not in emotions: emotions[role] = []
                emotions[role].append(emotion)
            
            # 4. 타임라인 (핵심 이벤트 흐름 요약)
            snippet = content.replace('\n', ' ').strip()
            if len(snippet) > 80: snippet = snippet[:77] + "..."
            emo_str = f" ({emotion})" if emotion else ""
            timeline.append(f"{role}{emo_str}: {snippet}")

        # 가장 빈번한 감정 요약
        summary_emotions = {}
        for role, emo_list in emotions.items():
            if emo_list:
                from collections import Counter
                most_common = Counter(emo_list).most_common(1)[0][0]
                summary_emotions[role] = most_common

        return {
            "files": sorted(list(files))[:15],
            "todos": todos[-5:],
            "timeline": timeline[-20:],
            "emotions": summary_emotions,
            "count": len(messages)
        }

    def generate_summary_message(self, metadata: Dict[str, Any]) -> str:
        """추출된 메타데이터를 사용하여 마크다운 요약본을 생성합니다."""
        summary = ["<summary>", "🔱 **System Context Summary (이전 대화 요약)**"]
        
        summary.append(f"\n- **압축 범위**: 이전 {metadata['count']}개의 메시지 통합")
        
        if metadata['emotions']:
            emo_lines = [f"{role}: {emo}" for role, emo in metadata['emotions'].items()]
            summary.append(f"- **주요 감정 상태**: {', '.join(emo_lines)}")

        if metadata['files']:
            summary.append(f"- **참조된 주요 파일**: {', '.join([f'`{f}`' for f in metadata['files']])}")
            
        if metadata['todos']:
            summary.append("- **남은 과제 및 TODO**:")
            for todo in metadata['todos']:
                summary.append(f"  - {todo}")
                
        summary.append("- **주요 타임라인**:")
        for event in metadata['timeline']:
            summary.append(f"  - {event}")
            
        summary.append("\n</summary>")
        
        return self.PREAMBLE + "\n".join(summary)

    def compact(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """메시지 리스트를 소버린 하네스 방식으로 압축합니다."""
        if not self.should_compact(messages):
            return messages
            
        settings = self._get_settings()
        preserve_count = int(settings.get("preserve_recent", 5))
        
        # 압축 대상과 보존 대상 분리
        split_idx = len(messages) - preserve_count
        to_compact = messages[:split_idx]
        preserved = messages[split_idx:]
        
        # 요약 생성
        metadata = self.extract_metadata(to_compact)
        summary_text = self.generate_summary_message(metadata)
        
        # 시스템 요약 메시지 생성
        summary_msg = {
            "text": summary_text,
            "type": 3, # 🔱 Imperial Type 3: System Context Summary
            "relayGuardian": "System",
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
            "sender_id": "97157d36-0b88-473f-8131-fab3e46aa560" # Shadow Brain ID
        }
        
        return [summary_msg] + preserved

# Global Singleton for Engine
import os
core_root = os.path.dirname(os.path.abspath(__file__))
compaction_engine = CompactionEngine(core_root)