108 lines
4.0 KiB
Python
108 lines
4.0 KiB
Python
"""arm-manifest 로드 + pre-flight 검증. arm 정체성은 full commit hash 로 pin, arm C 는 실제
|
|
resolve 되는 profile 이 전부 active 여야(draft fallback 0) 완전한 P3-B arm 으로 인정한다."""
|
|
import os
|
|
import subprocess
|
|
|
|
import yaml
|
|
|
|
from . import paths
|
|
|
|
_ACT_REL = "org-os/00-role-registry/method-contract-activations.yaml"
|
|
|
|
|
|
def load():
|
|
with open(os.path.join(paths.controller_dir(), "arm-manifest.yaml"), encoding="utf-8") as f:
|
|
return yaml.safe_load(f)
|
|
|
|
|
|
def git_state(commit):
|
|
r = subprocess.run(["git", "cat-file", "-e", commit + "^{commit}"],
|
|
cwd=paths.ROOT, capture_output=True, text=True)
|
|
return {"exists": r.returncode == 0, "clean": r.returncode == 0}
|
|
|
|
|
|
def _show(commit, relpath):
|
|
r = subprocess.run(["git", "show", f"{commit}:{relpath}"],
|
|
cwd=paths.ROOT, capture_output=True, text=True)
|
|
return r.stdout if r.returncode == 0 else None
|
|
|
|
|
|
def _unwrap_roles(data):
|
|
"""실제 registry 는 `method-contract-activations: {version, roles: {role: {methods:...}}}`
|
|
로 감싸져 있다. 과거/대안 형식(top-level `activations:` 키, 또는 role 이 바로 top-level에
|
|
오는 bare mapping)도 함께 허용해 스키마 변화에 견고하게 대응한다."""
|
|
if not isinstance(data, dict):
|
|
return {}
|
|
for key in ("method-contract-activations", "activations"):
|
|
nested = data.get(key)
|
|
if isinstance(nested, dict):
|
|
data = nested
|
|
break
|
|
roles = data.get("roles")
|
|
if isinstance(roles, dict):
|
|
return roles
|
|
# bare role mapping(래퍼 없이 role 이 바로 top-level) — dict 값만 role record 로 취급
|
|
return {k: v for k, v in data.items() if isinstance(v, dict)}
|
|
|
|
|
|
def active_methods_at(commit):
|
|
"""그 commit 의 activation registry 를 읽어 {role: [active method-id]}."""
|
|
body = _show(commit, _ACT_REL)
|
|
if not body:
|
|
return {}
|
|
data = yaml.safe_load(body) or {}
|
|
out = {}
|
|
for role, rec in _unwrap_roles(data).items():
|
|
if not isinstance(rec, dict):
|
|
continue
|
|
methods = rec.get("methods")
|
|
if not isinstance(methods, dict):
|
|
continue
|
|
act = [m for m, d in methods.items()
|
|
if isinstance(d, dict) and d.get("status") == "active"]
|
|
if act:
|
|
out[role] = act
|
|
return out
|
|
|
|
|
|
def command_exists_at(commit, name):
|
|
return _show(commit, f".claude/commands/{name}.md") is not None
|
|
|
|
|
|
def preflight(man=None):
|
|
man = man or load()
|
|
v = []
|
|
arms = man["arms"]
|
|
for a in ("A", "B", "C"):
|
|
c = arms[a]["commit"]
|
|
st = git_state(c)
|
|
if not st["exists"]:
|
|
v.append(f"arm {a}: commit {c[:8]} 부재")
|
|
continue
|
|
for cmd in man.get("required-commands", ["ground", "decide", "design-direction"]):
|
|
if not command_exists_at(c, cmd):
|
|
v.append(f"arm {a}: command /{cmd} 부재({c[:8]})")
|
|
# arm B: P3-B active 미혼입
|
|
if arms["B"]["commit"] and sum(len(x) for x in active_methods_at(arms["B"]["commit"]).values()) > 0:
|
|
v.append("arm B: P3-B active 계약 혼입(구조이동 arm 아님)")
|
|
# arm C: 요구 profile 전부 active(draft fallback 0)
|
|
amC = active_methods_at(arms["C"]["commit"])
|
|
for spec in man.get("pilot-invoked-methods", []):
|
|
role = spec["role"]
|
|
for mid in spec["methods"]:
|
|
if mid not in amC.get(role, []):
|
|
v.append(f"arm C: {role}/{mid} 가 active 아님(draft fallback — 완전한 P3-B arm 아님)")
|
|
return v
|
|
|
|
|
|
def drift(man, resolved_method_plan):
|
|
"""수기 pilot-invoked-methods 와 dry-run resolved plan 대조. resolved 에 있으나 manifest 에
|
|
없는 (role, method) 를 위반으로 반환."""
|
|
declared = {(s["role"], m) for s in man.get("pilot-invoked-methods", []) for m in s["methods"]}
|
|
v = []
|
|
for r in resolved_method_plan or []:
|
|
key = (r.get("role-id"), r.get("method-id"))
|
|
if key not in declared:
|
|
v.append(f"drift: resolved {key} 가 manifest pilot-invoked-methods 에 없음")
|
|
return v
|