#!/usr/bin/env python3 """activate_method_contract — Contract v2 활성화 trusted CLI (P3-B §14). method-contract-activations.yaml 의 **유일한 정당 writer**. guard_tools 가 그 파일의 직접 Write/Edit·Bash redirection·언어레벨 write 를 차단하므로, 활성화는 이 CLI 를 거쳐야 하고 이 CLI 는 아래 4단 검증을 통과해야만 write 한다. draft→active 는 되돌릴 수 없는 품질 게이트다. 검증 순서(하나라도 실패 → 거부, write 없음): 1. 계약 profile 실존 — resolve_method_profile(role, method) != None 2. contract-sha256 일치 — canonical_contract_hash(profile) == --contract-sha256 (리뷰된 계약과 실제 활성화 대상이 같음을 보장 — drift 차단) 3. golden validation-report — --validation-report 파일 실존 & sha256 == --validation-report-sha256 4. HUMAN signoff(위조불가) — 의 human-signoff.jsonl 에 stage `method-contract:::`(또는 '*') 승인 존재. guard 가 에이전트의 signoff 파일 write·`state_engine signoff` 호출을 모두 막으므로 사람만 세션 밖에서 발급 가능(P0-4 soft-boundary). 성공 시: roles[role].methods[method] = {status, contract-sha256, validation-report, validation-report-sha256, acceptance-workflow, acceptance-stage, activated-by, activated-at, previous-status} 를 임시파일→os.replace 로 원자 교체. 레코드 전체가 감사 추적(provenance)이다. Usage(사람 또는 오케스트레이터 — 4단 게이트가 실제 방어): activate_method_contract.py activate --role DES-DIRECTOR --method converge \ --contract-sha256 --validation-report \ --validation-report-sha256 --acceptance-workflow [--activated-by ID] activate_method_contract.py retire --role ... --method ... --acceptance-workflow API(테스트·프로그램): verify(role, method, contract_sha256, report_path, report_sha256, signoff_wf, *, profile=..., has_signoff=..., now=...) -> (ok: bool, errors: list[str]) apply_activation(role, method, record, registry_path=...) -> None # 원자 write """ import argparse import hashlib import os import sys import yaml HOOKS = os.path.dirname(os.path.abspath(__file__)) if HOOKS not in sys.path: sys.path.insert(0, HOOKS) import method_contracts as mc # noqa: E402 try: import state_engine as se # noqa: E402 except Exception: # noqa: BLE001 — state_engine 미가용 시 signoff 확인은 주입 필요 se = None def _sha256_file(path): h = hashlib.sha256() with open(path, "rb") as fh: for chunk in iter(lambda: fh.read(65536), b""): h.update(chunk) return h.hexdigest() def signoff_stage(role, method, contract_sha256): """계약별 HUMAN signoff stage 토큰 — 사람 승인을 이 계약 hash 에 바인딩(무관 signoff 재사용 차단).""" return f"method-contract:{role}:{method}:{contract_sha256[:12]}" def _default_has_signoff(wf, stage): if se is None: return False return se._has_human_signoff(wf, stage) def verify(role, method, contract_sha256, report_path, report_sha256, signoff_wf, *, profile=None, has_signoff=None): """4단 게이트. (ok, errors) 반환 — write 하지 않는다.""" errors = [] prof = profile if profile is not None else mc.resolve_method_profile(role, method) if not prof: return False, [f"{role}/{method}: 계약 profile 미존재(v1 이거나 미정의) — 활성화 불가"] computed = mc.canonical_contract_hash(prof) if not contract_sha256 or computed != contract_sha256: errors.append(f"contract-sha256 불일치: 인자={contract_sha256!r} != 계약={computed!r} " "(리뷰된 계약과 활성화 대상이 다름)") if not report_path or not os.path.exists(report_path): errors.append(f"validation-report 파일 미존재: {report_path!r}(golden task 산출 필요)") elif not report_sha256 or _sha256_file(report_path) != report_sha256: errors.append("validation-report sha256 불일치(golden report 위조·교체 의심)") hs = has_signoff or _default_has_signoff stage = signoff_stage(role, method, contract_sha256 or computed) if not (hs(signoff_wf, stage) or hs(signoff_wf, "*")): errors.append(f"HUMAN signoff 없음: workflow={signoff_wf!r} stage={stage!r} — " "사람이 세션 밖에서 golden+계약을 수용해야 활성화 가능(OPS-ORCH 단독 불가)") return (not errors), errors def _registry_path(registry_path=None): return registry_path or mc.ACTIVATIONS def _load_doc(registry_path): if os.path.exists(registry_path): return yaml.safe_load(open(registry_path)) or {} return {} def apply_activation(role, method, record, registry_path=None): """활성화 레코드를 원자 교체(임시파일→os.replace)로 write. 검증은 호출측(verify) 책임.""" rp = _registry_path(registry_path) doc = _load_doc(rp) doc.setdefault("method-contract-activations", {}).setdefault("version", 1) roles = doc["method-contract-activations"].setdefault("roles", {}) prev = ((roles.get(role) or {}).get("methods") or {}).get(method) or {} record["previous-status"] = prev.get("status", "draft") roles.setdefault(role, {}).setdefault("methods", {})[method] = record tmp = rp + ".tmp" with open(tmp, "w", encoding="utf-8") as fh: yaml.safe_dump(doc, fh, allow_unicode=True, sort_keys=False) os.replace(tmp, rp) def _now(): if se is not None and hasattr(se, "_now"): return se._now() return "unknown" def cmd_activate(args, *, new_status="active"): ok, errors = verify(args.role, args.method, args.contract_sha256, args.validation_report, args.validation_report_sha256, args.acceptance_workflow) if not ok: sys.stderr.write(f"[activate] 거부({new_status}): {args.role}/{args.method}\n") for e in errors: sys.stderr.write(f" - {e}\n") return 2 record = { "status": new_status, "contract-sha256": args.contract_sha256, "validation-report": args.validation_report, "validation-report-sha256": args.validation_report_sha256, "acceptance-workflow": args.acceptance_workflow, "acceptance-stage": signoff_stage(args.role, args.method, args.contract_sha256), "activated-by": args.activated_by, "activated-at": _now(), } apply_activation(args.role, args.method, record, registry_path=args.registry) print(f"{args.role}/{args.method}: {new_status} " f"(contract-sha256={args.contract_sha256[:12]}…, report={os.path.basename(args.validation_report)})") return 0 def cmd_retire(args): """active→retired. 동일 4단 게이트(재활성 아닌 은퇴도 사람 승인).""" return cmd_activate(args, new_status="retired") def build_parser(): p = argparse.ArgumentParser(description="Contract v2 활성화 trusted CLI") sub = p.add_subparsers(dest="cmd", required=True) for name in ("activate", "retire"): s = sub.add_parser(name) s.add_argument("--role", required=True) s.add_argument("--method", required=True) s.add_argument("--contract-sha256", dest="contract_sha256", required=True) s.add_argument("--validation-report", dest="validation_report", required=True) s.add_argument("--validation-report-sha256", dest="validation_report_sha256", required=True) s.add_argument("--acceptance-workflow", dest="acceptance_workflow", required=True) s.add_argument("--activated-by", dest="activated_by", default=None) s.add_argument("--registry", dest="registry", default=None) return p def main(argv=None): args = build_parser().parse_args(argv) return cmd_retire(args) if args.cmd == "retire" else cmd_activate(args) if __name__ == "__main__": sys.exit(main())