""" migrate_capability_matrix_v213.py — ANU v2.13 capability matrix 마이그레이션 (순수 함수, stdlib only) 기존 matrix 의 capabilities 각 항목에 v2.13 권장 lowercase 필드를 누락 시에만 추가함. 기존 값 절대 파괴 금지. ★ main() 에서 write=True 시 파일을 덮어쓸 수 있으나, 이 task 에서는 함수 정의만이 목적이며 실행은 하지 않음 (main 자동 실행 금지). """ import json import copy import argparse import os import subprocess # --------------------------------------------------------------------------- # 동적 workspace root resolve (HIGH 6~9 — 하드코딩 절대경로 제거) # 1) ANU_WORKSPACE 환경변수 2) git toplevel 3) os.getcwd() # --------------------------------------------------------------------------- # git rev-parse 서브프로세스 결과는 프로세스 생애 내 불변 → module-level 1회 캐시. # ANU_WORKSPACE / cwd 는 호출마다 변할 수 있으므로 캐시하지 않고 매번 평가한다 # (테스트의 monkeypatch.setenv 격리 보존). 비싼 git subprocess 만 캐싱한다. _GIT_TOPLEVEL_CACHE = None _GIT_TOPLEVEL_RESOLVED = False def _git_toplevel(): global _GIT_TOPLEVEL_CACHE, _GIT_TOPLEVEL_RESOLVED if _GIT_TOPLEVEL_RESOLVED: return _GIT_TOPLEVEL_CACHE try: here = os.path.dirname(os.path.abspath(__file__)) top = subprocess.run( ["git", "rev-parse", "--show-toplevel"], cwd=here, capture_output=True, text=True, ).stdout.strip() _GIT_TOPLEVEL_CACHE = top or None except Exception: _GIT_TOPLEVEL_CACHE = None _GIT_TOPLEVEL_RESOLVED = True return _GIT_TOPLEVEL_CACHE def _resolve_root(): env = os.environ.get("ANU_WORKSPACE") if env: return env top = _git_toplevel() if top: return top return os.getcwd() def default_matrix_path(): """automation_capability_matrix.json 의 기본 경로 (동적 resolve).""" return os.path.join( _resolve_root(), "memory", "state", "automation_capability_matrix.json" ) # --------------------------------------------------------------------------- # v2.13 추가 필드 목록 # --------------------------------------------------------------------------- V213_FIELDS = [ "implemented", "verified", "wired", "active", "policy", "rollback_cost", "trust_roots_touched", "semantic_residue", "budget_scope", ] # --------------------------------------------------------------------------- # migrate_entry # --------------------------------------------------------------------------- def migrate_entry(entry): """ 단일 capability entry 를 deepcopy 후 누락된 v2.13 필드만 추가. 기존 키는 절대 변경하지 않음. Parameters ---------- entry : dict Returns ------- dict """ e = copy.deepcopy(entry) # implemented if "implemented" not in e: e["implemented"] = bool(e.get("IMPLEMENTED", False)) # verified if "verified" not in e: e["verified"] = bool(e.get("VERIFIED", False)) # wired if "wired" not in e: if e.get("ACTIVE") in (True, "full"): e["wired"] = "full" elif e.get("WIRED") not in (None, False, "", 0): e["wired"] = "canary" else: e["wired"] = "none" # active if "active" not in e: if e.get("ACTIVE") in (True, "full"): e["active"] = "full" else: e["active"] = "none" # policy if "policy" not in e: e["policy"] = "POLICY_REQUIRES_CHAIR" # rollback_cost if "rollback_cost" not in e: e["rollback_cost"] = "MEDIUM" # trust_roots_touched if "trust_roots_touched" not in e: e["trust_roots_touched"] = [] # semantic_residue if "semantic_residue" not in e: e["semantic_residue"] = False # budget_scope if "budget_scope" not in e: e["budget_scope"] = "chair_approved_goal_id" return e # --------------------------------------------------------------------------- # migrate_matrix # --------------------------------------------------------------------------- def migrate_matrix(matrix): """ matrix dict 전체를 deepcopy 후 capabilities 각 항목에 migrate_entry 적용. 다른 최상위 키는 보존. capabilities 는 두 형태를 모두 지원한다: - dict 형태(실제 automation_capability_matrix.json): {name: entry, ...} → 각 value(entry) 에만 migrate_entry 적용, 키(name) 보존. - list 형태: [entry, ...] → 각 item 에 migrate_entry 적용. 그 외 타입은 그대로 보존. Parameters ---------- matrix : dict Returns ------- dict """ m = copy.deepcopy(matrix) # matrix 가 dict 가 아니면(None/list 등) m.get() 에서 AttributeError → isinstance 방어. # 비-dict 입력은 변형하지 않고 그대로 보존한다 (MEDIUM). if not isinstance(m, dict): return m capabilities = m.get("capabilities") if isinstance(capabilities, dict): m["capabilities"] = { name: (migrate_entry(entry) if isinstance(entry, dict) else entry) for name, entry in capabilities.items() } elif isinstance(capabilities, list): m["capabilities"] = [ (migrate_entry(entry) if isinstance(entry, dict) else entry) for entry in capabilities ] return m # --------------------------------------------------------------------------- # main # --------------------------------------------------------------------------- def main( path=None, write=True, ): """ Parameters ---------- path : str | None — matrix JSON 파일 경로. None 이면 default_matrix_path() 동적 resolve. write : bool — True 면 마이그레이션 결과를 같은 경로에 atomic 저장 (HIGH 10) Returns ------- dict — 마이그레이션된 matrix """ if path is None: path = default_matrix_path() with open(path, "r", encoding="utf-8") as f: matrix = json.load(f) migrated = migrate_matrix(matrix) if write: dirpath = os.path.dirname(path) if dirpath: os.makedirs(dirpath, exist_ok=True) # atomic write: tmp + flush + fsync + os.replace (HIGH 10) tmp_path = path + ".tmp" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(migrated, f, ensure_ascii=False, indent=2) f.flush() os.fsync(f.fileno()) os.replace(tmp_path, path) return migrated if __name__ == "__main__": parser = argparse.ArgumentParser( description="ANU v2.13 capability matrix migration" ) parser.add_argument( "--path", default=None, help="matrix JSON 파일 경로 (미지정 시 ANU_WORKSPACE/git toplevel 기준 동적 resolve)", ) parser.add_argument( "--dry-run", action="store_true", help="파일을 쓰지 않고 결과만 출력", ) args = parser.parse_args() result = main(path=args.path, write=not args.dry_run) print(json.dumps(result, ensure_ascii=False, indent=2))