#!/usr/bin/env python3 """ 세션 시작 라우팅 레이어 활성 프로젝트만 로드하여 토큰 90% 절약 """ import json from datetime import datetime, timedelta from pathlib import Path from typing import Dict, List, Optional class SessionRouter: """세션 시작 시 최소 컨텍스트만 로드""" def __init__(self, workspace_path: str = None): """초기화""" if workspace_path is None: workspace_path = Path.home() / ".cokacdir" / "workspace" self.workspace_path = Path(workspace_path) self.memory_path = self.workspace_path / "memory" # 활성 프로젝트 추적 파일 self.active_projects_file = self.memory_path / "active-projects.json" # 오늘 날짜 self.today = datetime.now().strftime("%Y-%m-%d") def load_minimal_context(self) -> Dict: """ 최소 컨텍스트 로드 (토큰 90% 절약) Returns: 최소 컨텍스트 (약 550 tokens) """ context = { "session_start": self.today, "active_projects": self.get_active_projects(), "today_priority": self.get_today_priority(), "soul_summary": self.get_soul_summary(), "total_tokens": 0, } # 토큰 추정 context["total_tokens"] = self._estimate_tokens(context) return context def get_active_projects(self) -> List[Dict]: """ 활성 프로젝트만 로드 Returns: 활성 프로젝트 목록 (최대 3개) """ # active-projects.json 확인 if self.active_projects_file.exists(): with open(self.active_projects_file, "r", encoding="utf-8") as f: data = json.load(f) return data.get("active", []) else: # 없으면 빈 리스트 반환 return [] def get_today_priority(self) -> List[str]: """ 오늘의 우선순위만 로드 Returns: 우선순위 목록 (최대 3개) """ # 오늘 업무일지 확인 daily_file = self.memory_path / "daily" / f"{self.today}.md" if daily_file.exists(): # 첫 10줄만 읽기 (요약) with open(daily_file, "r", encoding="utf-8") as f: lines = f.readlines()[:10] # 우선순위 추출 priorities = [] for line in lines: if "우선순위" in line or "할 일" in line or "- [" in line: priorities.append(line.strip()) return priorities[:3] # 최대 3개 return ["활성 프로젝트 없음"] def get_soul_summary(self) -> str: """ SOUL.md 요약만 로드 (전체 대신) Returns: 요약 문자열 (약 100 tokens) """ soul_file = self.workspace_path / "SOUL.md" if soul_file.exists(): with open(soul_file, "r", encoding="utf-8") as f: content = f.read() # 첫 200자만 (요약) summary = content[:200] + "..." if len(content) > 200 else content return summary return "SOUL.md 없음" def update_active_projects(self, projects: List[Dict]): """ 활성 프로젝트 업데이트 Args: projects: 활성 프로젝트 목록 """ data = {"last_updated": datetime.now().isoformat(), "active": projects[:3]} # 최대 3개만 유지 with open(self.active_projects_file, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) def add_active_project(self, project: Dict): """ 활성 프로젝트 추가 Args: project: 프로젝트 정보 """ current = self.get_active_projects() # 이미 있으면 업데이트 updated = False for i, p in enumerate(current): if p.get("name") == project.get("name"): current[i] = project updated = True break # 없으면 추가 if not updated: current.append(project) # 최대 3개 유지 self.update_active_projects(current[-3:]) def remove_active_project(self, project_name: str): """ 활성 프로젝트 제거 Args: project_name: 프로젝트명 """ current = self.get_active_projects() filtered = [p for p in current if p.get("name") != project_name] self.update_active_projects(filtered) def _estimate_tokens(self, context: Dict) -> int: """ 컨텍스트 토큰 수 추정 Args: context: 컨텍스트 딕셔너리 Returns: 추정 토큰 수 """ # JSON 문자열로 변환 후 길이 계산 json_str = json.dumps(context, ensure_ascii=False) # 한글은 약 1.5 characters/token estimated = len(json_str) / 1.5 return int(estimated) def get_full_context_for_project(self, project_name: str) -> Dict: """ 특정 프로젝트의 전체 컨텍스트 로드 (필요 시에만) Args: project_name: 프로젝트명 Returns: 프로젝트 전체 컨텍스트 """ # 프로젝트 폴더 찾기 project_path = self.memory_path / "projects" / project_name.lower().replace(" ", "-") if not project_path.exists(): return {"error": f"프로젝트 없음: {project_name}"} # 프로젝트 문서 로드 context = { "project_name": project_name, "phases": self._load_project_phases(project_path), "workflows": self._load_project_workflows(project_path), } return context def _load_project_phases(self, project_path: Path) -> List[Dict]: """프로젝트 Phase 로드""" phases_path = project_path / "phases" phases = [] if phases_path.exists(): for phase_file in sorted(phases_path.glob("*.md")): with open(phase_file, "r", encoding="utf-8") as f: content = f.read() phases.append({"file": phase_file.name, "summary": content[:200]}) # 요약만 return phases def _load_project_workflows(self, project_path: Path) -> List[Dict]: """프로젝트 Workflow 로드""" workflows_path = project_path / "workflows" workflows = [] if workflows_path.exists(): for workflow_dir in sorted(workflows_path.iterdir()): if workflow_dir.is_dir(): # workflow 상태 확인 status = self._check_workflow_status(workflow_dir) workflows.append({"id": workflow_dir.name, "status": status}) return workflows def _check_workflow_status(self, workflow_dir: Path) -> str: """Workflow 상태 확인""" # feedback.md 있으면 완료 if list(workflow_dir.glob("*feedback*")): return "completed" # implementation.md 있으면 진행 중 if list(workflow_dir.glob("*implementation*")): return "in-progress" # plan.md만 있으면 계획 중 if list(workflow_dir.glob("*plan*")): return "planning" return "unknown" # CLI 인터페이스 if __name__ == "__main__": import sys router = SessionRouter() if len(sys.argv) < 2: # 기본: 최소 컨텍스트 로드 context = router.load_minimal_context() print(json.dumps(context, indent=2, ensure_ascii=False)) print(f"\n📊 추정 토큰: {context['total_tokens']:,}") else: command = sys.argv[1] if command == "status": # 현재 상태 context = router.load_minimal_context() print("📊 세션 시작 컨텍스트:") print(f" - 활성 프로젝트: {len(context['active_projects'])}개") print(f" - 오늘 우선순위: {len(context['today_priority'])}개") print(f" - 추정 토큰: {context['total_tokens']:,}") elif command == "add": # 활성 프로젝트 추가 if len(sys.argv) < 3: print("사용법: python3 routing_layer.py add <프로젝트명>") sys.exit(1) project_name = sys.argv[2] router.add_active_project({"name": project_name, "added": datetime.now().isoformat()}) print(f"✅ 활성 프로젝트 추가: {project_name}") elif command == "remove": # 활성 프로젝트 제거 if len(sys.argv) < 3: print("사용법: python3 routing_layer.py remove <프로젝트명>") sys.exit(1) project_name = sys.argv[2] router.remove_active_project(project_name) print(f"✅ 활성 프로젝트 제거: {project_name}") elif command == "full": # 특정 프로젝트 전체 컨텍스트 if len(sys.argv) < 3: print("사용법: python3 routing_layer.py full <프로젝트명>") sys.exit(1) project_name = sys.argv[2] context = router.get_full_context_for_project(project_name) print(json.dumps(context, indent=2, ensure_ascii=False)) else: print("알 수 없는 명령:", command) print("사용법:") print(" python3 routing_layer.py status # 현재 상태") print(" python3 routing_layer.py add <프로젝트> # 활성 추가") print(" python3 routing_layer.py remove <프로젝트> # 활성 제거") print(" python3 routing_layer.py full <프로젝트> # 전체 컨텍스트")