"""사람 author PR을 bot equivalent로 재작성. 원본 commit 보존(cherry-pick + amend).""" from __future__ import annotations import json import os import subprocess from datetime import datetime, timezone from pathlib import Path from typing import Any WORKSPACE = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) HANDOFF_LOG = WORKSPACE / "memory" / "orchestration-audit" / "handoff-to-bot.jsonl" class HandoffError(Exception): """handoff-to-bot 작업 실패 시 raise.""" def _now() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _default_runner(cmd: list[str], **kwargs: Any) -> subprocess.CompletedProcess: """기본 subprocess.run 래퍼.""" return subprocess.run(cmd, capture_output=True, text=True, **kwargs) def _default_pr_view(pr_number: int, runner: Any) -> dict[str, Any]: """gh pr view로 PR 메타 조회.""" proc = runner( ["gh", "pr", "view", str(pr_number), "--json", "number,author,headRefName,commits,url,state"], timeout=30, ) if proc.returncode != 0: raise HandoffError(f"gh pr view 실패 (PR #{pr_number}): {proc.stderr[-300:].strip()}") try: return json.loads(proc.stdout or "{}") except json.JSONDecodeError as exc: raise HandoffError(f"gh pr view JSON 파싱 실패: {exc}") from exc def handoff_pr_to_bot( pr_number: int, *, bot_login: str = "jeon-jonghyuk-taskctl-bot", bot_email: str = "jeon-jonghyuk-taskctl-bot[bot]@users.noreply.github.com", bot_token: str | None = None, dry_run: bool = False, pr_view_func=None, # 테스트 주입용 runner=None, # subprocess.run 래퍼 주입용 ) -> dict[str, Any]: """사람 author PR을 bot equivalent로 재작성. 1) 원본 PR 메타 조회 (number, author, head_branch, commits[]) 2) 새 branch 'task/handoff--bot' 생성 3) cherry-pick 각 commit + git commit --amend --author --reset-author로 author=bot 강제 4) git push (bot_token), gh pr create (author=bot) 5) 원본 PR close 6) audit jsonl append 7) 반환: {original_pr, original_author, new_pr, new_branch, mapping_log_path} 원본 commit 영구 삭제 절대 금지(cherry-pick은 원본 브랜치를 건드리지 않음). dry_run=True면 실제 git/gh 호출 없이 mapping dict만 반환 + audit jsonl 기록. """ # runner / pr_view_func 기본값 설정 if runner is None: runner = _default_runner if pr_view_func is None: _pv_runner = runner def _resolved_pr_view(pr_num: int) -> dict[str, Any]: return _default_pr_view(pr_num, _pv_runner) _pr_view_fn = _resolved_pr_view else: _pr_view_fn = pr_view_func # 1) 원본 PR 메타 조회 try: pr_data = _pr_view_fn(pr_number) except HandoffError: raise except Exception as exc: raise HandoffError(f"PR 메타 조회 실패: {exc}") from exc original_author = (pr_data.get("author") or {}).get("login", "unknown") head_branch = pr_data.get("headRefName", f"pr-{pr_number}") commits_raw = pr_data.get("commits") or [] # commits는 [{oid: ..., messageHeadline: ...}, ...] 형태일 수 있음 original_commits: list[str] = [] for c in commits_raw: if isinstance(c, dict): oid = c.get("oid") or c.get("sha") or "" else: oid = str(c) if oid: original_commits.append(oid) new_branch = f"task/handoff-{pr_number}-bot" new_commits: list[str] = [] new_pr_number: int | None = None # dry_run이면 실제 git/gh 호출 없이 audit 기록 + mapping만 반환 if dry_run: _append_audit_log( pr_number=pr_number, original_author=original_author, new_pr=None, new_branch=new_branch, original_commits=original_commits, new_commits=[], dry_run=True, ) return { "original_pr": pr_number, "original_author": original_author, "new_pr": None, "new_branch": new_branch, "mapping_log_path": str(HANDOFF_LOG), "dry_run": True, } # 2) 새 branch 생성 (origin/main 기준) proc = runner(["git", "checkout", "-b", new_branch, "origin/main"], timeout=60) if proc.returncode != 0: # 이미 존재하면 checkout proc2 = runner(["git", "checkout", new_branch], timeout=30) if proc2.returncode != 0: raise HandoffError( f"새 branch '{new_branch}' 생성/checkout 실패: {proc.stderr[-200:].strip()}" ) # 3) cherry-pick 각 commit + author amend for sha in original_commits: cp_proc = runner(["git", "cherry-pick", sha], timeout=60) if cp_proc.returncode != 0: # cherry-pick 실패 시 abort runner(["git", "cherry-pick", "--abort"], timeout=30) raise HandoffError( f"cherry-pick 실패 (sha={sha}): {cp_proc.stderr[-200:].strip()}" ) # --amend로 author를 bot으로 변경 amend_proc = runner( [ "git", "commit", "--amend", "--no-edit", f"--author={bot_login} <{bot_email}>", "--reset-author", ], timeout=60, ) if amend_proc.returncode != 0: raise HandoffError( f"git commit --amend 실패 (sha={sha}): {amend_proc.stderr[-200:].strip()}" ) # 새 commit SHA 수집 rev_proc = runner(["git", "rev-parse", "HEAD"], timeout=10) if rev_proc.returncode == 0: new_commits.append(rev_proc.stdout.strip()) # 4) git push + gh pr create push_env: dict[str, str] = {} if bot_token: push_env["GH_TOKEN"] = bot_token push_proc = runner( ["git", "push", "origin", new_branch, "--force-with-lease"], timeout=120, env={**os.environ, **push_env}, ) if push_proc.returncode != 0: raise HandoffError(f"git push 실패: {push_proc.stderr[-200:].strip()}") pr_title = f"[handoff-bot] PR #{pr_number} (original author: {original_author})" pr_body = ( f"Bot handoff: original PR #{pr_number} by @{original_author}.\n\n" f"Original branch: {head_branch}\n" f"Original commits: {', '.join(original_commits)}\n" f"Handoff branch: {new_branch}\n" ) create_cmd = [ "gh", "pr", "create", "--title", pr_title, "--body", pr_body, "--head", new_branch, ] create_proc = runner( create_cmd, timeout=60, env={**os.environ, **push_env}, ) if create_proc.returncode != 0: raise HandoffError(f"gh pr create 실패: {create_proc.stderr[-200:].strip()}") # 새 PR 번호 파싱 (URL 또는 gh pr view) pr_url = create_proc.stdout.strip() view_proc = runner( ["gh", "pr", "view", new_branch, "--json", "number"], timeout=30, env={**os.environ, **push_env}, ) if view_proc.returncode == 0: try: new_pr_number = json.loads(view_proc.stdout).get("number") except Exception: new_pr_number = None # 5) 원본 PR close (★ 원자성: 실패 시 hard fail + audit 박제) close_proc = runner( ["gh", "pr", "close", str(pr_number), "--comment", f"Closed: bot handoff to PR #{new_pr_number} on branch {new_branch}"], timeout=30, env={**os.environ, **push_env}, ) if close_proc.returncode != 0: # 원본 PR close 실패 → 새 PR과 원본 PR 동시 open 상태가 됨 → hard fail # audit에 'CLOSE_FAILED' outcome 박제 후 raise _append_audit_log( pr_number=pr_number, original_author=original_author, new_pr=new_pr_number, new_branch=new_branch, original_commits=original_commits, new_commits=new_commits, dry_run=False, outcome="CLOSE_FAILED", ) raise HandoffError( f"원본 PR #{pr_number} close 실패 — 새 PR #{new_pr_number}와 원본이 모두 열린 상태. " f"수동 정리 필요. stderr={close_proc.stderr[-200:].strip()}" ) # 6) audit jsonl append (성공) _append_audit_log( pr_number=pr_number, original_author=original_author, new_pr=new_pr_number, new_branch=new_branch, original_commits=original_commits, new_commits=new_commits, dry_run=False, outcome="HANDED_OFF", ) # 7) 반환 return { "original_pr": pr_number, "original_author": original_author, "new_pr": new_pr_number, "new_branch": new_branch, "mapping_log_path": str(HANDOFF_LOG), "new_commits": new_commits, "pr_url": pr_url, } def _append_audit_log( *, pr_number: int, original_author: str, new_pr: int | None, new_branch: str, original_commits: list[str], new_commits: list[str], dry_run: bool, outcome: str = "OK", ) -> None: """audit jsonl에 handoff 기록 추가. outcome: 'OK' | 'DRY_RUN' | 'HANDED_OFF' | 'CLOSE_FAILED' 등. """ HANDOFF_LOG.parent.mkdir(parents=True, exist_ok=True) record = { "ts": _now(), "original_pr": pr_number, "original_author": original_author, "new_pr": new_pr, "new_branch": new_branch, "original_commits": original_commits, "new_commits": new_commits, "dry_run": dry_run, "outcome": outcome, } with HANDOFF_LOG.open("a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n")