#!/usr/bin/env python3 """ utils/session_auto_compress.py — 세션 자동 압축 모듈 두 가지 기능을 제공합니다: 1. 자동 압축 훅: CRITICAL 감지 시 ContextCompressor를 통해 메시지 압축 2. 세션 요약 + 재시작 파일 저장: 90% 초과 시 세션 요약을 마크다운으로 저장 Usage: from utils.session_auto_compress import SessionAutoCompress from utils.session_monitor import SessionMonitor monitor = SessionMonitor(context_limit=200_000) sac = SessionAutoCompress(monitor=monitor) sac.setup_auto_hooks() compressed, event = sac.auto_compress(messages) path = sac.save_session_summary(messages, task_id="task-100.1", team_id="dev6-team") """ from __future__ import annotations import json import os from datetime import datetime from pathlib import Path from typing import Any from utils.context_compressor import ContextCompressor, _estimate_tokens from utils.context_summarizer import generate_summary from utils.logger import get_logger from utils.session_monitor import SessionMonitor logger = get_logger(__name__) _WORKSPACE_ROOT = os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace") def _estimate_tokens_for_messages(messages: list[dict[str, Any]]) -> int: """메시지 리스트의 총 토큰 수를 추정한다.""" total_text = "" for msg in messages: content = msg.get("content") or "" total_text += content tool_calls = msg.get("tool_calls") if tool_calls: try: total_text += json.dumps(tool_calls, ensure_ascii=False) except (TypeError, ValueError): total_text += str(tool_calls) return _estimate_tokens(total_text) if total_text else 0 class SessionAutoCompress: """세션 자동 압축 및 요약 저장 클래스. Args: monitor: SessionMonitor 인스턴스 context_limit: 컨텍스트 최대 토큰 수 (기본값: 200,000) events_dir: 이벤트 파일 저장 디렉토리 (기본값: WORKSPACE/memory/events/) sessions_dir: 세션 요약 파일 저장 디렉토리 (기본값: WORKSPACE/memory/sessions/) """ def __init__( self, monitor: SessionMonitor, context_limit: int = 200_000, events_dir: str | Path | None = None, sessions_dir: str | Path | None = None, ) -> None: self.monitor = monitor self.context_limit = context_limit workspace = Path(_WORKSPACE_ROOT) self.events_dir: Path = Path(events_dir) if events_dir is not None else workspace / "memory" / "events" self.sessions_dir: Path = Path(sessions_dir) if sessions_dir is not None else workspace / "memory" / "sessions" logger.debug( "SessionAutoCompress 초기화: context_limit=%d, events_dir=%s, sessions_dir=%s", context_limit, self.events_dir, self.sessions_dir, ) # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def auto_compress(self, messages: list[dict]) -> tuple[list[dict], dict]: """압축 수행. ContextCompressor를 사용해 메시지를 압축하고, 이벤트 파일을 저장하며 monitor의 토큰 카운터를 리셋한다. Args: messages: 압축할 메시지 리스트 Returns: (압축된 메시지 리스트, 이벤트 정보 dict) 튜플 """ before_count = len(messages) tokens_before = _estimate_tokens_for_messages(messages) if not messages: event_info = self._build_event_info( before_count=0, after_count=0, tokens_before=0, tokens_after=0, ) self._save_event(event_info) return [], event_info compressor = ContextCompressor(context_limit=self.context_limit) compressed = compressor.compress(messages) after_count = len(compressed) tokens_after = _estimate_tokens_for_messages(compressed) # 모니터 리셋: 압축 후 추정 토큰 수로 설정 self.monitor.reset(new_total=tokens_after) logger.info( "auto_compress: %d → %d messages, tokens %d → %d", before_count, after_count, tokens_before, tokens_after, ) event_info = self._build_event_info( before_count=before_count, after_count=after_count, tokens_before=tokens_before, tokens_after=tokens_after, ) self._save_event(event_info) return compressed, event_info def save_session_summary( self, messages: list[dict], task_id: str, team_id: str, modified_files: list[str] | None = None, remaining_tasks: list[str] | None = None, last_success_step: str | None = None, errors: list[str] | None = None, ) -> Path: """세션 요약을 마크다운 파일로 저장하고 경로를 반환한다. generate_summary()를 호출하여 기본 요약 텍스트를 생성하고, 추가 정보와 함께 마크다운 형식으로 저장한다. Args: messages: 요약할 메시지 리스트 task_id: 작업 ID team_id: 팀 ID modified_files: 수정한 파일 목록 remaining_tasks: 남은 작업 목록 last_success_step: 마지막 성공 단계 설명 errors: 에러/실패 목록 Returns: 저장된 파일의 Path 객체 """ timestamp = datetime.now().strftime("%Y%m%dT%H%M%S") summary_text = generate_summary(messages, use_llm=True) or "(요약 없음)" content = self._build_markdown( task_id=task_id, team_id=team_id, timestamp=timestamp, summary_text=summary_text, modified_files=modified_files, remaining_tasks=remaining_tasks, last_success_step=last_success_step, errors=errors, ) self.sessions_dir.mkdir(parents=True, exist_ok=True) filename = f"summary-{task_id}-{timestamp}.md" path = self.sessions_dir / filename path.write_text(content, encoding="utf-8") logger.info("세션 요약 저장: %s", path) return path def setup_auto_hooks(self) -> None: """모니터에 CRITICAL 콜백으로 _on_critical을 자동 등록한다.""" self.monitor.register_callback("critical", self._on_critical) logger.debug("setup_auto_hooks: critical 콜백 등록 완료") # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ def _on_critical(self, status: dict) -> None: """CRITICAL 레벨 도달 시 호출되는 내부 콜백. 로그만 남긴다.""" logger.warning( "CRITICAL 레벨 감지: total_tokens=%d, usage_pct=%.1f%% — 외부에서 auto_compress() 호출 필요", status.get("total_tokens", 0), status.get("usage_pct", 0.0), ) def _build_event_info( self, before_count: int, after_count: int, tokens_before: int, tokens_after: int, ) -> dict: """이벤트 정보 dict를 생성한다.""" return { "event": "session_compressed", "timestamp": datetime.now().isoformat(), "before_count": before_count, "after_count": after_count, "tokens_before": tokens_before, "tokens_after": tokens_after, } def _save_event(self, event_info: dict) -> None: """이벤트 정보를 JSON 파일로 저장한다.""" self.events_dir.mkdir(parents=True, exist_ok=True) timestamp_str = datetime.now().strftime("%Y%m%dT%H%M%S%f") filename = f"session-compressed-{timestamp_str}.json" path = self.events_dir / filename with open(path, "w", encoding="utf-8") as f: json.dump(event_info, f, ensure_ascii=False, indent=2) logger.debug("이벤트 파일 저장: %s", path) def _build_markdown( self, task_id: str, team_id: str, timestamp: str, summary_text: str, modified_files: list[str] | None, remaining_tasks: list[str] | None, last_success_step: str | None, errors: list[str] | None, ) -> str: """세션 요약 마크다운 문자열을 생성한다.""" lines: list[str] = [ f"# 세션 요약: {task_id}", "", "## 기본 정보", f"- 작업 ID: {task_id}", f"- 팀: {team_id}", f"- 생성 시각: {timestamp}", "", "## 자동 요약", summary_text, "", "## 수정한 파일", ] if modified_files: for f in modified_files: lines.append(f"- {f}") else: lines.append("(없음)") lines += [ "", "## 남은 작업", ] if remaining_tasks: for task in remaining_tasks: lines.append(f"- [ ] {task}") else: lines.append("(없음)") lines += [ "", "## 마지막 성공 단계", last_success_step if last_success_step else "(없음)", "", "## 에러/실패", ] if errors: for error in errors: lines.append(f"- {error}") else: lines.append("(없음)") return "\n".join(lines) + "\n"