Files

826 lines
39 KiB
Python

#!/usr/bin/env python3
"""doctor.py — Org OS 하네스 preflight 점검(`orgos doctor` / `/doctor`).
문서상 "강제"라고 적힌 계약들이 실제로 켜져 있는지, 실행 전에 한 번에 확인한다.
점검 항목(WP-1 / spec 2026-07-10-p0-execution-integrity):
1. .claude/settings.json 존재 + hook 배선이 C7 배선표와 일치(5 이벤트, 참조 스크립트).
2. 배선이 참조하는 hook 스크립트 파일이 .claude/hooks/ 아래 실존(부재 시 WARN — 병렬 WP가 만드는 중일 수 있음).
3. python3 동작 + pyyaml import 가능.
4. workspace 해석 가능: _workspace.py import + ORGOS_WORKSPACE 또는 .orgos-workspace 설정 여부.
(WP-4 이후 미설정 workspace는 오류 — 여기서 명확히 표면화한다.)
5. lint_refs.py(WP-3)가 있으면 실행해 커맨드→agent 참조 무결성을 접어 넣는다(없으면 우아하게 skip).
hard problem(FAIL)이 하나라도 있으면 비영점 종료. 사람이 읽는 섹션형 리포트를 출력한다.
이 스크립트는 형제 WP 스크립트가 아직 없어도 크래시하지 않는다(부재는 보고만).
"""
import glob
import importlib
import json
import os
import re
import subprocess
import sys
# ---- repo / 경로 해석 (CLAUDE_PROJECT_DIR 우선, 없으면 이 파일 기준 2단계 위) -------------
REPO = os.environ.get("CLAUDE_PROJECT_DIR") or os.path.dirname(
os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
)
HOOKS_DIR = os.path.join(REPO, ".claude", "hooks")
SETTINGS_PATH = os.path.join(REPO, ".claude", "settings.json")
# ---- C7 배선표 (spec 고정 계약) ------------------------------------------------------------
# (event, expected_matcher, script, required_flag, matcher_required)
C7 = [
("PreToolUse", "Bash|Write|Edit|NotebookEdit", "guard_tools.py", None, True),
("PostToolUse", "Bash|Write|Edit", "evidence_ledger.py", None, True),
("SubagentStart", None, "subagent_register.py", None, False),
("SubagentStop", None, "stop_validate.py", None, False),
("Stop", None, "stop_validate.py", "--main", False),
]
# 형제 WP가 생성/재작성하는 스크립트 — 부재 시 안내 주석용
SIBLING_WP = {
"evidence_ledger.py": "WP-6 (PostToolUse evidence receipts)",
"subagent_register.py": "WP-2 (SubagentStart registry)",
"stop_validate.py": "WP-2 (rewrite)",
"lint_refs.py": "WP-3 (ref linter)",
"guard_tools.py": "existing",
}
SCRIPT_RE = re.compile(r"\.claude/hooks/([A-Za-z0-9_]+\.py)")
class Report:
"""섹션별 OK/WARN/FAIL 누적 + 출력."""
def __init__(self):
self.entries = [] # (section, level, msg)
self.n_ok = 0
self.n_warn = 0
self.n_fail = 0
def ok(self, section, msg):
self.entries.append((section, "OK", msg))
self.n_ok += 1
def warn(self, section, msg):
self.entries.append((section, "WARN", msg))
self.n_warn += 1
def fail(self, section, msg):
self.entries.append((section, "FAIL", msg))
self.n_fail += 1
def render(self, sections):
mark = {"OK": "[ OK ]", "WARN": "[WARN]", "FAIL": "[FAIL]"}
out = []
out.append("=" * 68)
out.append("orgos doctor — 하네스 실행 무결성 preflight")
out.append(f"repo: {REPO}")
out.append("=" * 68)
for sec in sections:
rows = [e for e in self.entries if e[0] == sec]
if not rows:
continue
out.append("")
out.append(f"## {sec}")
for _, level, msg in rows:
# 여러 줄 메시지는 들여쓰기 유지
first, *rest = msg.splitlines() or [""]
out.append(f" {mark[level]} {first}")
for line in rest:
out.append(f" {line}")
out.append("")
out.append("-" * 68)
verdict = "FAIL" if self.n_fail else ("WARN" if self.n_warn else "OK")
out.append(
f"summary: {self.n_ok} OK · {self.n_warn} WARN · {self.n_fail} FAIL"
f" → verdict: {verdict}"
)
if self.n_fail:
out.append("hard problem(FAIL)이 있어 종료코드 1로 나갑니다. 위 FAIL을 먼저 고치세요.")
elif self.n_warn:
out.append("WARN은 대개 형제 WP가 진행 중이라 나타납니다(부재 스크립트 등). 배선 자체는 유효.")
print("\n".join(out))
# ---------------------------------------------------------------------------------------------
def load_settings(report):
"""settings.json 로드 + JSON 유효성. 반환: dict | None."""
section = "1. settings.json + hook 배선(C7)"
if not os.path.exists(SETTINGS_PATH):
report.fail(section, ".claude/settings.json 이 없음 — hook이 실제로 꺼져 있음(문서상 강제와 불일치).")
return None
try:
with open(SETTINGS_PATH, encoding="utf-8") as f:
data = json.load(f)
except Exception as e: # noqa: BLE001
report.fail(section, f".claude/settings.json JSON 파싱 실패: {e}")
return None
report.ok(section, ".claude/settings.json 존재 · 유효 JSON")
return data
def check_wiring(report, settings):
"""C7 배선표와 대조. 반환: 참조된 스크립트 파일명 집합."""
section = "1. settings.json + hook 배선(C7)"
referenced = set()
if not settings:
return referenced
hooks = settings.get("hooks")
if not isinstance(hooks, dict):
report.fail(section, "settings.json 에 'hooks' 객체가 없음.")
return referenced
# 전 이벤트에서 참조 스크립트 수집(존재성 점검용)
for ev, groups in hooks.items():
if not isinstance(groups, list):
continue
for g in groups:
for h in (g or {}).get("hooks", []) or []:
cmd = (h or {}).get("command", "") or ""
for m in SCRIPT_RE.findall(cmd):
referenced.add(m)
# C7 이벤트별 검증
for ev, exp_matcher, script, req_flag, matcher_required in C7:
groups = hooks.get(ev)
if not groups:
report.fail(section, f"{ev}: 배선 없음 (기대 스크립트 {script}).")
continue
# 이벤트 내 모든 command 문자열/matcher 수집
cmds = []
matchers = []
for g in groups:
if isinstance(g, dict) and "matcher" in g:
matchers.append(g.get("matcher"))
for h in (g or {}).get("hooks", []) or []:
cmds.append((h or {}).get("command", "") or "")
# 스크립트 참조 + (필요 시) 플래그 확인
hit = [c for c in cmds if script in c and (req_flag is None or req_flag in c)]
if not hit:
need = f"{script}" + (f" {req_flag}" if req_flag else "")
report.fail(section, f"{ev}: 기대 스크립트 미배선 ({need}). 실제 command: {cmds or '없음'}")
continue
detail = f"{ev}{script}" + (f" {req_flag}" if req_flag else "")
# matcher 검증(PreToolUse/PostToolUse만 필수). exp_matcher는 **최소 커버 집합**이다 —
# 실제 matcher가 이 도구들을 모두 포함하면 OK(추가 도구는 허용). 예: guard가 finding #11로
# Read|Grep|Glob 을 더 커버해도 정상. (예전엔 정확일치라 정당한 확장을 WARN 처리했음.)
if matcher_required:
need_tools = set((exp_matcher or "").split("|"))
covered = any(need_tools <= set((m or "").split("|")) for m in (matchers or []))
if covered:
actual = next((m for m in matchers if need_tools <= set((m or "").split("|"))), exp_matcher)
extra = " (+확장)" if actual != exp_matcher else ""
report.ok(section, f"{detail} (matcher: {actual}{extra})")
else:
report.warn(
section,
f"{detail} 배선됨, 다만 matcher가 최소기대({exp_matcher})를 포함하지 않음: {matchers or '없음'}",
)
else:
report.ok(section, detail)
return referenced
def check_referenced_scripts(report, referenced):
section = "2. 참조 hook 스크립트 실존"
if not referenced:
report.warn(section, "settings.json에서 참조된 스크립트를 찾지 못함(배선 확인 필요).")
return
for name in sorted(referenced):
path = os.path.join(HOOKS_DIR, name)
if os.path.exists(path):
report.ok(section, f"{name} 존재")
else:
wp = SIBLING_WP.get(name, "미상 WP")
report.warn(section, f"{name} 없음 — {wp}가 생성 예정(현재는 배선만 되어 있음).")
def _ver_tuple(s):
"""'6.0.1' / 'v24.14.0' / '0.7.1' -> (6,0,1). 파싱 실패 시 ()."""
import re as _re
m = _re.search(r"(\d+(?:\.\d+)+)", str(s or ""))
if not m:
return ()
return tuple(int(x) for x in m.group(1).split("."))
def _cli_version(cmd_args):
"""CLI 버전 문자열을 얻는다(없으면 None)."""
import shutil as _sh
if not _sh.which(cmd_args[0]):
return None
try:
r = subprocess.run(cmd_args, capture_output=True, text=True, timeout=10)
return (r.stdout + r.stderr).strip()
except Exception: # noqa: BLE001
return None
def check_python_deps(report):
# 섹션명은 render(sections)의 목록과 **정확히** 일치해야 한다(render 는 exact match 로 그룹핑).
# 예전엔 뒤에 "(finding #18…)"가 붙어 3번 섹션 전체가 출력에서 숨겨졌다(재리뷰 지적).
section = "3. python / 의존성"
report.ok(section, f"python3 실행 가능: {sys.version.split()[0]} ({sys.executable}) — tool-versions.yaml 대조(#18)")
# tool-versions.yaml(min/tested)을 읽어 런타임 도구를 대조한다. 없으면 최소검사만.
tv_path = os.path.join(REPO, ".claude", "tool-versions.yaml")
tv = {}
try:
import yaml as _y
tv = (_y.safe_load(open(tv_path, encoding="utf-8")) or {}).get("tool-versions", {})
except Exception: # noqa: BLE001
tv = {}
def _min_of(tier, key, default_min):
spec = ((tv.get(tier) or {}).get(key) or {})
return spec.get("min", default_min)
# pyyaml (required)
try:
import yaml # noqa: F811
ver = getattr(yaml, "__version__", "?")
need = _min_of("required", "pyyaml", "6.0")
if _ver_tuple(ver) and _ver_tuple(ver) < _ver_tuple(need):
report.fail(section, f"pyyaml {ver} < 최소 {need} (requirements.txt로 업그레이드).")
else:
report.ok(section, f"pyyaml {ver} (>= {need})")
except Exception as e: # noqa: BLE001
report.fail(section, f"pyyaml import 실패: {e} — `pip install -r requirements.txt` 필요.")
# python min (required)
pneed = _min_of("required", "python", "3.10")
pv = sys.version.split()[0]
if _ver_tuple(pv) < _ver_tuple(pneed):
report.fail(section, f"python {pv} < 최소 {pneed}.")
# jsonschema (recommended: 없으면 폴백 → WARN)
try:
import jsonschema # noqa: F401
try:
import importlib.metadata as _md
jver = _md.version("jsonschema")
except Exception: # noqa: BLE001
jver = "?"
report.ok(section, f"jsonschema {jver} (유형별 스키마 검증 활성)")
except Exception: # noqa: BLE001
report.warn(section, "jsonschema 없음 — validate_report가 최소검증 폴백으로 degrade "
"(`pip install -r requirements.txt` 권장).")
# node / d2 (recommended: design-system·diagram 렌더), marp (optional)
for tier, key, args, feat in [
("recommended", "node", ["node", "--version"], "design-system(vite)·preview_ui"),
("recommended", "d2", ["d2", "--version"], "diagram-as-code 실물 렌더"),
("optional", "marp", ["marp", "--version"], "consult 덱(.pptx/.pdf)"),
]:
v = _cli_version(args)
need = _min_of(tier, key, "0")
if v is None:
(report.warn if tier == "recommended" else report.ok)(
section, f"{key} 없음 — {feat} 제한"
+ ("" if tier == "recommended" else "(대체 경로 있음)") + ".")
elif _ver_tuple(v) and _ver_tuple(need) and _ver_tuple(v) < _ver_tuple(need):
report.warn(section, f"{key} {v.splitlines()[0]} < 권장 {need} ({feat}).")
else:
report.ok(section, f"{key} {v.splitlines()[0].strip()} (>= {need})")
def check_workspace(report):
section = "4. workspace 해석"
if HOOKS_DIR not in sys.path:
sys.path.insert(0, HOOKS_DIR)
try:
ws = importlib.import_module("_workspace")
except Exception as e: # noqa: BLE001
report.fail(section, f"_workspace.py import 실패: {e}")
return
env_val = (os.environ.get("ORGOS_WORKSPACE") or "").strip()
ptr_path = os.path.join(REPO, ".orgos-workspace")
ptr_val = ""
if os.path.exists(ptr_path):
try:
ptr_val = open(ptr_path, encoding="utf-8").read().strip()
except OSError:
ptr_val = ""
# 해석 시도 — WP-4가 도입할 WorkspaceNotSetError를 포함해 모든 예외를 안전 처리.
try:
name = ws.workspace_name()
root = ws.work_root()
except Exception as e: # noqa: BLE001
cls = type(e).__name__
if cls == "WorkspaceNotSetError" or "workspace" in str(e).lower():
report.fail(
section,
f"workspace 미설정: {e}\n"
"→ ORGOS_WORKSPACE 환경변수 또는 .orgos-workspace 포인터를 지정하세요.",
)
else:
report.fail(section, f"workspace 해석 중 예외 {cls}: {e}")
return
exists = os.path.isdir(root)
# 재리뷰 지적: 존재하지 않는 명시 workspace 경로도 OK 로 집계됐다. 이제 미존재 디렉터리는
# FAIL 로 처리한다(운영 훅이 그 경로에 산출물을 쓰지 못하므로 무결성 위반).
root_note = " [경고: 해당 디렉터리 미존재]"
if not exists:
report.fail(section, f"workspace 디렉터리 미존재: '{name}' → {root}\n"
"→ 그 경로가 실존해야 운영 훅이 산출물(state/evidence/reports)을 쓸 수 있습니다.")
return
if env_val:
report.ok(section, f"ORGOS_WORKSPACE 명시 → '{name}' ({root})")
elif ptr_val:
default_like = name in ("_sandbox",)
msg = f".orgos-workspace 포인터 → '{name}' ({root})"
if default_like:
report.warn(
section,
msg + "\n→ 로컬/테스트 기본값입니다. 실제 운영에서는 ORGOS_WORKSPACE를 명시하세요(WP-4).",
)
else:
report.ok(section, msg)
else:
# env·포인터 둘 다 없는데 해석됐다면 하드코딩 기본값(WP-4 이전 상태)에 의존한 것.
report.fail(
section,
f"ORGOS_WORKSPACE·.orgos-workspace 둘 다 미설정. 현재 하드코딩 기본값 '{name}'로 해석됨.\n"
"→ WP-4 적용 후 이는 오류가 됩니다. 지금 workspace를 명시하세요.",
)
def check_lint_refs(report):
section = "5. 커맨드→agent 참조 무결성(lint_refs.py)"
path = os.path.join(HOOKS_DIR, "lint_refs.py")
if not os.path.exists(path):
report.warn(section, "lint_refs.py 아직 없음(WP-3) — 커맨드/agent 참조 검사 skip.")
return
try:
r = subprocess.run(
[sys.executable, path],
cwd=REPO,
capture_output=True,
text=True,
timeout=90,
)
except Exception as e: # noqa: BLE001
report.fail(section, f"lint_refs.py 실행 실패: {e}")
return
out = (r.stdout + ("\n" + r.stderr if r.stderr else "")).strip()
if r.returncode == 0:
report.ok(section, "lint_refs.py 통과 — 모든 참조 해소.")
else:
head = out if out else "(출력 없음)"
report.fail(section, f"lint_refs.py rc={r.returncode} — 깨진 참조 존재:\n{head}")
def check_ssot_consumption(report):
"""finding #13: 정책 YAML이 **실제로 hook에 소비되는지** 정직하게 보고한다.
'SSOT'라 부르면서 코드가 안 읽는 prose-only YAML을 가시화한다(과장 방지·회귀 감지)."""
section = "6. SSOT 소비 현황(#13: 정책 YAML이 코드에 실제로 읽히나)"
reg = os.path.join(REPO, "org-os", "00-role-registry")
aw = os.path.join(REPO, "org-os", "06-agent-work")
policies = {
"state-transition-rules.yaml": reg, "tool-permission-matrix.yaml": reg,
"lens-registry.yaml": reg, "drai-matrix.yaml": reg,
"role-selection-scorecard.yaml": reg, "collaboration-map.yaml": aw,
"governance-tiers.yaml": aw, "collaboration-modes.yaml": aw,
"execution-policy.yaml": aw, "context-package-spec.yaml": aw,
"report-templates.yaml": aw, "design-brief-spec.yaml": aw,
"agent-operating-kpi.yaml": aw,
}
# doctor.py 자신은 감사 목적으로 모든 YAML 이름을 언급하므로 소비자 스캔에서 제외(오탐 방지).
hook_files = [f for f in glob.glob(os.path.join(HOOKS_DIR, "**", "*.py"), recursive=True)
if os.path.realpath(f) != os.path.realpath(__file__)]
texts = {}
for hf in hook_files:
try:
texts[os.path.relpath(hf, HOOKS_DIR)] = open(
hf, encoding="utf-8", errors="replace"
).read()
except OSError:
pass
def _consumes(text, yml):
"""재리뷰 지적: 예전엔 파일명이 텍스트 어디든(주석 포함) 있으면 '소비'로 판정했다.
이제 **비주석 코드 라인 + 파일접근 관용구(open/load/read/join/Path/glob)**와 함께
나타날 때만 실제 소비로 본다(주석 언급만으로는 소비 아님)."""
for raw in text.splitlines():
line = raw.lstrip()
if line.startswith("#"):
continue
code = line.split("#", 1)[0] # rough inline-comment strip
if yml in code and re.search(r"open|safe_load|\bload\b|read|join|Path|glob", code):
return True
return False
consumed, prose = [], []
for yml, base in policies.items():
if not os.path.exists(os.path.join(base, yml)):
report.warn(section, f"{yml}: SoT 파일 부재.")
continue
readers = sorted(h for h, t in texts.items() if _consumes(t, yml))
if readers:
consumed.append(yml)
report.ok(section, f"{yml} ← 소비: {', '.join(readers)}")
else:
prose.append(yml)
if prose:
# prose-only 는 실패 아님(일부는 정당한 서술 가이드) — 단 '코드 강제 아님'을 정직히 표시.
report.ok(section, f"prose-only(코드 미소비, 'SSOT' 아닌 서술 가이드): {', '.join(prose)}")
report.ok(section, f"요약: 소비 {len(consumed)} · prose-only {len(prose)} / 총 {len(policies)}")
def check_company_context_lint(report):
"""7. company-context.yaml 내부 정합(lint_company_context Hard Fail 0)."""
section = "7. company-context 정합(lint_company_context)"
if HOOKS_DIR not in sys.path:
sys.path.insert(0, HOOKS_DIR)
try:
import lint_company_context as L
path = os.path.join(REPO, "org-os", "01-company", "company-context.yaml")
hard, warn = L.lint_file(path, is_candidate=False)
for w in warn:
report.warn(section, w)
if hard:
report.fail(section, "company-context.yaml Hard Fail: " + "; ".join(hard))
else:
report.ok(section, f"company-context.yaml lint OK (warnings {len(warn)})")
except Exception as e: # noqa: BLE001
report.fail(section, f"company-context lint 점검 오류: {e}")
def check_venture_bootstrap_wiring(report):
"""8. venture-bootstrap 배선(P1: 신규 SoT/hook 실존 + plan + validation-map role-id 등록)."""
section = "8. venture-bootstrap 배선(P1)"
if HOOKS_DIR not in sys.path:
sys.path.insert(0, HOOKS_DIR)
try:
import yaml
missing = []
for p in ("org-os/01-company/founder-context.yaml",
"org-os/06-agent-work/venture-option-spec.yaml",
"org-os/06-agent-work/venture-validation-map.yaml",
".claude/hooks/lint_company_context.py",
".claude/hooks/commit_company_context.py"):
if not os.path.exists(os.path.join(REPO, p)):
missing.append(p)
if missing:
report.fail(section, "P1 신규 파일 누락: " + ", ".join(missing)); return
plans = yaml.safe_load(open(os.path.join(REPO, "org-os/06-agent-work/execution-plans.yaml")))["execution-plans"]["plans"]
if "venture-bootstrap" not in plans:
report.fail(section, "execution-plans 에 venture-bootstrap plan 없음"); return
fams = yaml.safe_load(open(os.path.join(REPO, "org-os/00-role-registry/capability-families.yaml")))["capability-families"]["families"]
reg = {str(r).upper() for fam in fams for r in (fam.get("member-role-ids") or [])} | {str(fam.get("lead-role-id")).upper() for fam in fams if fam.get("lead-role-id")}
m = yaml.safe_load(open(os.path.join(REPO, "org-os/06-agent-work/venture-validation-map.yaml")))["venture-validation-map"]
used = {x for g in m["gates"] for x in (g["primary"] + g["auditor"])} | set(m["opportunity-discovery-roles"]["diverge"] + m["opportunity-discovery-roles"]["contrarian"]) | {m["synthesis-owner"]}
unreg = sorted({u for u in used if str(u).upper() not in reg and not str(u).upper().startswith("HUMAN")})
if unreg:
report.fail(section, "venture-validation-map 미등록 role-id: " + ", ".join(unreg)); return
report.ok(section, "venture-bootstrap 배선 OK (파일·plan·role-id 등록 확인)")
except Exception as e: # noqa: BLE001
report.fail(section, f"venture-bootstrap 배선 점검 오류: {e}")
def check_design_direction_wiring(report):
"""9. design-direction 배선(P2: 신규 spec/hook/커맨드/agent 실존 + role 등록 + plan)."""
section = "9. design-direction 배선(P2)"
if HOOKS_DIR not in sys.path:
sys.path.insert(0, HOOKS_DIR)
try:
import yaml
missing = []
for p in ("org-os/06-agent-work/design-direction-spec.yaml",
".claude/hooks/lint_design_direction.py",
".claude/commands/design-direction.md",
".claude/commands/design-review.md",
".claude/agents/des-director.md",
".claude/agents/des-visual.md"):
if not os.path.exists(os.path.join(REPO, p)):
missing.append(p)
if missing:
report.fail(section, "P2 신규 파일 누락: " + ", ".join(missing)); return
txt = open(os.path.join(REPO, "org-os/00-role-registry/roles.yaml")).read()
unreg = [rid for rid in ("DES-DIRECTOR", "DES-VISUAL") if rid not in txt]
if unreg:
report.fail(section, "roles.yaml 미등록 role-id: " + ", ".join(unreg)); return
plans = yaml.safe_load(open(os.path.join(REPO, "org-os/06-agent-work/execution-plans.yaml")))["execution-plans"]["plans"]
if "design-direction" not in plans:
report.fail(section, "execution-plans 에 design-direction plan 없음"); return
report.ok(section, "design-direction 배선 OK (spec·hook·concrete agent·role·plan 확인)")
except Exception as e: # noqa: BLE001
report.fail(section, f"design-direction 배선 점검 오류: {e}")
def check_method_skill_wiring(report):
"""10. method-skill 배선(P3): registry 완전성·실존·참조해소·고아0·drift0·파일분리 정합."""
section = "10. method-skill 배선(P3)"
if HOOKS_DIR not in sys.path:
sys.path.insert(0, HOOKS_DIR)
try:
import glob as _glob
import subprocess as _sp
import yaml
from skill_refs import known_skill_names, parse_skills
reg_path = os.path.join(REPO, "org-os/00-role-registry/method-skill-registry.yaml")
if not os.path.exists(reg_path):
report.fail(section, "method-skill-registry.yaml 없음"); return
reg = yaml.safe_load(open(reg_path))["method-skill-registry"]
roles = reg["roles"]
gen_dir = os.path.join(REPO, reg["generated-dir"])
fams = yaml.safe_load(open(os.path.join(REPO, "org-os/00-role-registry/capability-families.yaml")))["capability-families"]["families"]
bound = set()
for f in fams:
bound |= set(f["member-role-ids"])
# 1. 완전성
miss = sorted(bound - set(roles))
if miss:
report.fail(section, "registry 미등록 역할: " + ", ".join(miss)); return
# 파일분리 정합(중복/누락/미include 0)
rwm_dir = os.path.join(REPO, "org-os/00-role-registry/role-working-methods")
idx = yaml.safe_load(open(os.path.join(rwm_dir, "index.yaml")))["role-method-contracts"]
merged, dup = set(), []
for inc in idx["includes"]:
for rid in (yaml.safe_load(open(os.path.join(rwm_dir, inc))) or {}).get("role-working-methods") or {}:
if rid in merged:
dup.append(rid)
merged.add(rid)
on_disk = {os.path.basename(p) for p in _glob.glob(os.path.join(rwm_dir, "*.yaml"))} - {"index.yaml"}
if dup or merged != bound or on_disk != set(idx["includes"]):
report.fail(section, f"role-working-methods 파일분리 불정합(dup={dup}, 누락={sorted(bound-merged)}, 파일={on_disk ^ set(idx['includes'])})"); return
# 2/4. 실존 + 고아
want = {r["method-skill"] for r in roles.values()}
miss_sk = sorted(s for s in want if not os.path.exists(os.path.join(gen_dir, s, "SKILL.md")))
if miss_sk:
report.fail(section, "생성 skill 파일 없음: " + ", ".join(miss_sk)); return
disk_sk = {os.path.basename(os.path.dirname(p)) for p in _glob.glob(os.path.join(gen_dir, "*", "SKILL.md"))
if os.path.basename(os.path.dirname(p)).endswith("-method")}
orphan = sorted(disk_sk - want)
if orphan:
report.fail(section, "고아 생성 skill: " + ", ".join(orphan)); return
# 3. 카드 skills: 참조 해소
known = known_skill_names(REPO)
unresolved = []
for a in _glob.glob(os.path.join(REPO, ".claude/agents/*.md")):
try:
fm = yaml.safe_load(open(a).read().split("---\n")[1]) or {}
except Exception: # noqa: BLE001
continue
for s in parse_skills(fm.get("skills")):
if s not in known:
unresolved.append(f"{os.path.basename(a)}:{s}")
if unresolved:
report.fail(section, "미해결 skills 참조: " + ", ".join(unresolved)); return
# 5. drift
rc = _sp.run([sys.executable, os.path.join(HOOKS_DIR, "gen_method_skills.py"), "--check"],
capture_output=True, text=True, env={**os.environ, "CLAUDE_PROJECT_DIR": REPO})
if rc.returncode != 0:
report.fail(section, "method-skill drift: " + (rc.stdout or rc.stderr).strip()); return
report.ok(section, f"method-skill 배선 OK ({len(roles)} roles · {len(disk_sk)} skills · 파일분리·참조·drift 정상)")
except Exception as e: # noqa: BLE001
report.fail(section, f"method-skill 배선 점검 오류: {e}")
def check_method_contract_wiring(report):
"""11. method-contract machinery(P3-B): policy engine·activation registry·capability-sections."""
section = "11. method-contract machinery(P3-B)"
if HOOKS_DIR not in sys.path:
sys.path.insert(0, HOOKS_DIR)
try:
import method_contracts as mc
# policy engine: 파일분리 병합 로드 + 현행 전부 v1(회귀 없음)
rm = mc.load_role_methods()
v2 = [r for r, e in rm.items() if (e.get("method-contract") or {}).get("version") == 2]
# activation registry: 로드 가능 + 미등록 → draft 기본
acts = mc.load_activations()
if not isinstance(acts, dict):
report.fail(section, "activation registry 로드 실패(dict 아님)"); return
if mc.resolve_activation("NO-ROLE", "no-method").get("status") != "draft":
report.fail(section, "resolve_activation 기본값이 draft 아님"); return
# active 인데 계약 profile/hash 불일치면 위험 — 정합 검사(현재 0개면 통과)
drift = []
for rid, rec in acts.items():
for mid, m in (rec.get("methods") or {}).items():
if m.get("status") != "active":
continue
prof = mc.resolve_method_profile(rid, mid, methods=rm)
if not prof:
drift.append(f"{rid}/{mid}(active인데 profile 없음)")
elif m.get("contract-sha256") and mc.canonical_contract_hash(prof) != m["contract-sha256"]:
drift.append(f"{rid}/{mid}(active hash≠계약 — 계약 변경 후 미재활성)")
if drift:
report.fail(section, "activation drift: " + ", ".join(drift)); return
# capability-sections manifest: 모든 section 해소 + section-sha256 계산 가능
skills = mc.load_capability_sections()
unresolved = []
n_sec = 0
for sk, entry in skills.items():
for sid in (entry.get("sections") or {}):
n_sec += 1
if mc.resolve_capability_section(sk, sid, skills=skills) is None:
unresolved.append(f"{sk}#{sid}")
if unresolved:
report.fail(section, "capability-section 미해소(헤딩 부재/파일 없음): " + ", ".join(unresolved)); return
# activation trusted CLI 실존
if not os.path.exists(os.path.join(HOOKS_DIR, "activate_method_contract.py")):
report.fail(section, "activate_method_contract.py 없음"); return
# migration-debt: 미해결 부채 surfacing(정직 대시보드 — 이행 미완을 숨기지 않는다)
try:
debt = mc.unresolved_debt()
except Exception: # noqa: BLE001
debt = []
n_active = sum(len((r.get("methods") or {})) for r in acts.values())
base = (f"v2 계약 {len(v2)}개·active {n_active}개·capability-section {n_sec}개 해소·activation CLI 존재")
if debt:
report.warn(section, f"계약 machinery OK·미해결 migration-debt {len(debt)}개(이행 진행중) — {base}")
else:
report.ok(section, f"계약 machinery OK ({base}·migration-debt 0)")
except Exception as e: # noqa: BLE001
report.fail(section, f"method-contract machinery 점검 오류: {e}")
def check_jsonl_integrity(report):
"""Fail closed when an append-only ledger contains a torn/malformed row.
Runtime readers intentionally skip malformed rows to remain query-safe. A
skipped decision or receipt must not be invisible operationally, therefore
doctor promotes corruption and duplicate immutable event ids to hard
failures.
"""
section = "12. append-only JSONL 원장 무결성"
if HOOKS_DIR not in sys.path:
sys.path.insert(0, HOOKS_DIR)
try:
import _workspace as workspace
root = workspace.work_root()
except Exception as exc: # check_workspace reports the primary failure.
report.fail(section, f"workspace 원장을 해석할 수 없음: {exc}")
return
paths = sorted(set(
glob.glob(os.path.join(root, "state", "**", "*.jsonl"), recursive=True)
+ glob.glob(os.path.join(root, "evidence", "**", "*.jsonl"), recursive=True)
))
if not paths:
report.ok(section, "검사할 JSONL 원장 없음(초기 workspace)")
return
id_keys = (
"workflow-event-id", "state-event-id", "artifact-event-id",
"acceptance-event-id", "registry-event-id", "usage-event-id",
"receipt_id", "tool_use_id",
)
failures = []
checked = 0
for path in paths:
seen = set()
try:
with open(path, encoding="utf-8") as fh:
for lineno, raw in enumerate(fh, 1):
if not raw.strip():
continue
checked += 1
try:
row = json.loads(raw)
except json.JSONDecodeError as exc:
failures.append(f"{os.path.relpath(path, root)}:{lineno} JSON 파싱 실패({exc.msg})")
continue
if not isinstance(row, dict):
failures.append(f"{os.path.relpath(path, root)}:{lineno} JSON object가 아님")
continue
identity = next(((key, str(row[key])) for key in id_keys if row.get(key)), None)
if identity and identity in seen:
failures.append(
f"{os.path.relpath(path, root)}:{lineno} 중복 immutable id "
f"{identity[0]}={identity[1]}"
)
if identity:
seen.add(identity)
except OSError as exc:
failures.append(f"{os.path.relpath(path, root)} 읽기 실패({exc})")
if failures:
for failure in failures[:20]:
report.fail(section, failure)
if len(failures) > 20:
report.fail(section, f"그 외 JSONL 무결성 오류 {len(failures) - 20}건")
else:
report.ok(section, f"JSONL {len(paths)}개 · non-empty row {checked}개 파싱/ID 무결성 정상")
def check_artifact_registry(report):
section = "13. compiled artifact registry"
compiler = os.path.join(HOOKS_DIR, "compile_artifact_registry.py")
try:
result = subprocess.run(
[sys.executable, compiler, "--check"], cwd=REPO,
capture_output=True, text=True, timeout=30,
)
except Exception as exc:
report.fail(section, f"artifact registry compiler 실행 실패: {exc}")
return
output = (result.stdout or result.stderr or "").strip()
if result.returncode == 0:
report.ok(section, output or "artifact registry invariants/drift 정상")
else:
report.fail(section, output or f"artifact registry check exit={result.returncode}")
def check_orgos_registry(report):
section = "14. compiled Org OS registries + architecture views"
compiler = os.path.join(HOOKS_DIR, "compile_orgos_registry.py")
try:
result = subprocess.run(
[sys.executable, compiler, "--check"], cwd=REPO,
capture_output=True, text=True, timeout=30,
)
except Exception as exc:
report.fail(section, f"Org OS registry compiler 실행 실패: {exc}")
return
output = (result.stdout or result.stderr or "").strip()
if result.returncode == 0:
report.ok(section, output or "Org OS registry invariants/drift 정상")
else:
report.fail(section, output or f"Org OS registry check exit={result.returncode}")
def check_experience_design_kernel(report):
section = "15. experience foundation + organization design kernel"
required = [
"org-os/08-design/principles.yaml",
"org-os/08-design/taste-profile.yaml",
"org-os/08-design/releases/index.yaml",
"org-os/08-design/design-engine-adapters.yaml",
".claude/commands/experience-foundation.md",
".claude/hooks/compile_design_system.py",
".claude/hooks/design_registry.py",
".claude/hooks/validate_design_engine_output.py",
".claude/hooks/first_draft_experiment.py",
"org-os/06-agent-work/first-draft-experiment-spec.yaml",
"hyeonworks/experiments/experience-foundation-ab/experiment.yaml",
]
missing = [path for path in required if not os.path.exists(os.path.join(REPO, path))]
if missing:
report.fail(section, "필수 경험/디자인 커널 파일 누락: " + ", ".join(missing))
return
try:
contracts = __import__("yaml").safe_load(open(
os.path.join(REPO, "org-os/06-agent-work/workflow-contracts.yaml"), encoding="utf-8"))
workflows = contracts["workflow-contracts"]["workflows"]
if "experience-foundation" not in workflows:
report.fail(section, "experience-foundation runtime workflow 없음")
return
result = subprocess.run(
[sys.executable, os.path.join(HOOKS_DIR, "compile_design_system.py"), "--check"],
cwd=REPO, capture_output=True, text=True, timeout=30)
output = (result.stdout or result.stderr or "").strip()
if result.returncode != 0:
report.fail(section, output or "organization design compiler drift")
return
report.ok(section, output or "experience/design kernel wiring 정상")
except Exception as exc:
report.fail(section, f"experience/design kernel 점검 오류: {exc}")
def main():
report = Report()
sections = [
"1. settings.json + hook 배선(C7)",
"2. 참조 hook 스크립트 실존",
"3. python / 의존성",
"4. workspace 해석",
"5. 커맨드→agent 참조 무결성(lint_refs.py)",
"6. SSOT 소비 현황(#13: 정책 YAML이 코드에 실제로 읽히나)",
"7. company-context 정합(lint_company_context)",
"8. venture-bootstrap 배선(P1)",
"9. design-direction 배선(P2)",
"10. method-skill 배선(P3)",
"11. method-contract machinery(P3-B)",
"12. append-only JSONL 원장 무결성",
"13. compiled artifact registry",
"14. compiled Org OS registries + architecture views",
"15. experience foundation + organization design kernel",
]
# 각 점검을 방어적으로 — 하나가 터져도 나머지는 계속.
try:
settings = load_settings(report)
referenced = check_wiring(report, settings)
check_referenced_scripts(report, referenced)
except Exception as e: # noqa: BLE001
report.fail(sections[0], f"설정/배선 점검 중 예기치 못한 오류: {e}")
for fn in (check_python_deps, check_workspace, check_lint_refs, check_ssot_consumption,
check_company_context_lint, check_venture_bootstrap_wiring, check_design_direction_wiring,
check_method_skill_wiring, check_method_contract_wiring, check_jsonl_integrity,
check_artifact_registry, check_orgos_registry, check_experience_design_kernel):
try:
fn(report)
except Exception as e: # noqa: BLE001
report.fail("3. python / 의존성", f"{fn.__name__} 중 오류: {e}")
report.render(sections)
return 1 if report.n_fail else 0
if __name__ == "__main__":
sys.exit(main())