"""passive_feedback.py - 파일 변경 감지 및 피드백 학습 저장 유틸리티.""" import argparse import difflib import json import os import sys from datetime import datetime from pathlib import Path from typing import Optional import anthropic sys.path.insert(0, str(Path(__file__).resolve().parent)) from learnings_archiver import add_learning def _get_workspace_root() -> Path: v = os.environ.get("WORKSPACE_ROOT") return Path(v) if v else Path(__file__).resolve().parent.parent def save_snapshot( skill_name: str, file_path: str, task_id: Optional[str] = None, ) -> dict: """file_path 내용을 스냅샷 디렉토리에 복사하고 메타데이터를 반환한다.""" workspace_root = _get_workspace_root() now = datetime.now() timestamp_str = now.strftime("%Y%m%d_%H%M%S") snap_dir = workspace_root / "memory" / "skill-feedback" / skill_name / "snapshots" snap_dir.mkdir(parents=True, exist_ok=True) snap_path = snap_dir / f"{timestamp_str}.txt" snap_path.write_text(Path(file_path).read_text(encoding="utf-8"), encoding="utf-8") return { "snapshot_path": str(snap_path.resolve()), "skill_name": skill_name, "timestamp": now.isoformat(), "task_id": task_id, } def _compute_meaningful_diff( original: str, modified: str, ) -> Optional[tuple[str, list[dict]]]: """두 텍스트를 비교하여 의미 있는 diff를 반환한다. 공백만 다른 경우 None 반환.""" orig_lines = original.splitlines(keepends=True) mod_lines = modified.splitlines(keepends=True) diff_lines: list[dict] = [] for line in difflib.unified_diff(orig_lines, mod_lines, lineterm=""): if line.startswith("---") or line.startswith("+++") or line.startswith("@@"): continue if line.startswith("-"): actual = line[1:].rstrip("\n") if actual.strip(): diff_lines.append({"type": "remove", "line": actual}) elif line.startswith("+"): actual = line[1:].rstrip("\n") if actual.strip(): diff_lines.append({"type": "add", "line": actual}) if not diff_lines: return None remove_stripped = sorted(d["line"].strip() for d in diff_lines if d["type"] == "remove") add_stripped = sorted(d["line"].strip() for d in diff_lines if d["type"] == "add") if remove_stripped == add_stripped: return None summary = "\n".join(("-" if d["type"] == "remove" else "+") + " " + d["line"] for d in diff_lines) return summary, diff_lines def _extract_learning( skill_name: str, diff_summary: str, diff_lines: list[dict], ) -> str: """anthropic SDK로 diff에서 learning을 추출한다.""" client = anthropic.Anthropic() prompt = ( f"스킬 '{skill_name}'의 파일이 다음과 같이 변경되었습니다:\n\n" f"{diff_summary}\n\n" "이 diff에서 어떤 개선이 이뤄졌는지 분석해줘. 한 문장으로 핵심 인사이트를 요약해줘." ) response = client.messages.create( model="claude-sonnet-4-6", max_tokens=500, messages=[{"role": "user", "content": prompt}], ) block = response.content[0] text = getattr(block, "text", None) return text if isinstance(text, str) else str(block) def _save_diff_record(record: dict, skill_name: str) -> None: """passive-diffs.jsonl에 레코드를 append한다.""" workspace_root = _get_workspace_root() diff_dir = workspace_root / "memory" / "skill-feedback" / skill_name diff_dir.mkdir(parents=True, exist_ok=True) with (diff_dir / "passive-diffs.jsonl").open("a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") def detect_diff( skill_name: str, original_path: str, modified_path: str, task_id: Optional[str] = None, ) -> Optional[dict]: """두 파일을 비교하여 의미 있는 diff 감지 후 learning 저장. 공백만 변경 시 None 반환.""" original = Path(original_path).read_text(encoding="utf-8") modified = Path(modified_path).read_text(encoding="utf-8") result = _compute_meaningful_diff(original, modified) if result is None: return None diff_summary, diff_lines = result learning_text = _extract_learning(skill_name, diff_summary, diff_lines) workspace_root = _get_workspace_root() learnings_path = workspace_root / "memory" / "skill-learning" / "learnings.jsonl" learning_entry = add_learning( skill_name=skill_name, source="jay-feedback", learning=learning_text, learnings_path=learnings_path, ) record: dict = { "timestamp": datetime.now().isoformat(), "skill_name": skill_name, "task_id": task_id, "snapshot_path": str(workspace_root / "memory" / "skill-feedback" / skill_name / "snapshots"), "diff_summary": diff_summary, "diff_lines": diff_lines, "learning_extracted": learning_text, "learning_id": str(learning_entry.get("id", "")), } _save_diff_record(record, skill_name) return record def get_history(skill_name: str) -> list[dict]: """스킬의 passive-diffs.jsonl 히스토리를 반환한다. 파일 없으면 빈 리스트.""" workspace_root = _get_workspace_root() diff_file = workspace_root / "memory" / "skill-feedback" / skill_name / "passive-diffs.jsonl" if not diff_file.exists(): return [] return [json.loads(line) for line in diff_file.read_text(encoding="utf-8").strip().splitlines() if line.strip()] def main() -> None: """CLI 진입점.""" parser = argparse.ArgumentParser(description="passive-feedback 유틸리티") subparsers = parser.add_subparsers(dest="command", help="명령어") snap_p = subparsers.add_parser("snapshot") snap_p.add_argument("--skill", required=True) snap_p.add_argument("--file", required=True) snap_p.add_argument("--task-id", dest="task_id", default=None) det_p = subparsers.add_parser("detect") det_p.add_argument("--skill", required=True) det_p.add_argument("--original", required=True) det_p.add_argument("--modified", required=True) det_p.add_argument("--task-id", dest="task_id", default=None) hist_p = subparsers.add_parser("history") hist_p.add_argument("--skill", required=True) args = parser.parse_args() if args.command == "snapshot": r = save_snapshot(skill_name=args.skill, file_path=args.file, task_id=args.task_id) print(json.dumps(r, ensure_ascii=False, indent=2)) elif args.command == "detect": res = detect_diff(args.skill, args.original, args.modified, task_id=args.task_id) if res is None: print("의미 있는 변경 없음 (공백/포맷만 변경).") else: print(json.dumps(res, ensure_ascii=False, indent=2)) elif args.command == "history": hist = get_history(args.skill) if not hist: print(f"'{args.skill}' 스킬의 diff 히스토리가 없습니다.") else: for rec in hist: print(json.dumps(rec, ensure_ascii=False)) else: parser.print_help() sys.exit(1) if __name__ == "__main__": main()