451 lines
18 KiB
Python
451 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""wiki_consistency_check.py — 문서 간 모순·동기화 결정론 검사기 (stdlib only).
|
|
|
|
consistency-contract (Single-Owner + Reference-Only) 의 결정론 레이어.
|
|
모순의 근원은 재진술(복제)이며, 다수의 모순은 'owner 문서의 정당한 진화 +
|
|
참조자의 무통보 낡음'으로 생긴다 — 따라서 (a) 참조의 기계 검증과
|
|
(b) 변경 시 역참조 전파 알림이 본 스크립트의 책임이다. 의미 대조
|
|
(요약 stale / 내용 모순)는 `wiki-consistency-auditor` (Layer 2) 의 몫.
|
|
|
|
검사 (전부 이진):
|
|
DANGLING_DECISION_REF [[feature-B]] D17 인데 B 의 결정 표에 D17 부재 (B 실존 시에만 — 부재는 BROKEN_LINK 몫)
|
|
BARE_DECISION_REF wikilink 없는 bare 슬러그 + D<n> 참조 (기계 추적 불가 — wikilink 화 필요)
|
|
BARE_OWNER_REF Coverage delegated 행의 owner 셀에 wikilink 없음
|
|
DUAL_OWNERSHIP 같은 관심사(정규화 exact)를 두 branch 가 covered-here 주장
|
|
DANGLING_SECTION_REF [[project-note]] §34 인데 해당 § 헤더 부재
|
|
|
|
모드:
|
|
--all vault 전수 검사 리포트 (exit 1 if findings)
|
|
--impact <slug|path> 해당 노트의 결정을 참조하는 문서 목록 (역참조 충격 분석)
|
|
--pre PreToolUse — projected 본문의 DANGLING_DECISION_REF 차단 (exit 2)
|
|
--post PostToolUse — DEM 행 편집 감지 시 참조자 목록 비차단 알림 (exit 2 = 모델에 정보 전달, 쓰기는 이미 완료)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
import wiki_rules
|
|
|
|
DEFAULT_ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
BRANCH_DIR = "raw/branch-notes"
|
|
PROJECT_DIR = "raw/project-notes"
|
|
|
|
# 결정 ID 정의 = 표 행의 첫 셀이 D<n> 로 시작 (DEM 등).
|
|
# 실코퍼스 변형 수용: `**D1**`, `D9 (2026-05-31 보강)`, `D3/D4` (복수 정의 셀).
|
|
DEM_ROW_RE = re.compile(r"^\|\s*\*{0,2}D\d+\b")
|
|
DEM_CELL_IDS_RE = re.compile(r"\bD(\d+)\b")
|
|
# 참조 윈도: 위키링크 종료 후 같은 줄 100자 내의 D<n> 토큰 (대문자만 — d2.naver 류 오탐 방지)
|
|
D_TOKEN_RE = re.compile(r"\bD(\d+)\b")
|
|
WIKILINK_RE = re.compile(r"\[\[([^\]|#]+?)(?:\.md)?(?:#[^\]|]*)?(?:\|[^\]]*)?\]\]")
|
|
BARE_SLUG_RE = re.compile(r"(?<![\[/\w-])((?:feature|fix|chore|experiment)-[a-z0-9-]{4,})")
|
|
SECTION_REF_RE = re.compile(r"§\s*(\d+)")
|
|
SECTION_DEF_RE = re.compile(r"^#{2,3}\s+(\d+)[.\s]")
|
|
COVERAGE_HEADER_RE = re.compile(r"^##+\s+.*coverage", re.I)
|
|
CONCERN_ID_PREFIX_RE = re.compile(r"^C\d+[a-z]?\s*:\s*")
|
|
REF_WINDOW = 100
|
|
|
|
|
|
def read(p: Path) -> str:
|
|
try:
|
|
return p.read_text(encoding="utf-8")
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def iter_lines_outside_fences(text: str):
|
|
"""(lineno, line) — fenced code block 내부(예시/템플릿) 제외."""
|
|
in_fence = False
|
|
for i, line in enumerate(text.splitlines(), start=1):
|
|
s = line.lstrip()
|
|
if s.startswith("```") or s.startswith("~~~"):
|
|
info = s.lstrip("`~").strip()
|
|
if in_fence:
|
|
if not info:
|
|
in_fence = False
|
|
else:
|
|
in_fence = True
|
|
continue
|
|
if not in_fence:
|
|
yield i, line
|
|
|
|
|
|
def dem_ids_from_text(text: str) -> set[str]:
|
|
"""표 행 첫 셀에서 정의된 D<n> 집합."""
|
|
ids: set[str] = set()
|
|
for _, line in iter_lines_outside_fences(text):
|
|
s = line.strip()
|
|
if not DEM_ROW_RE.match(s):
|
|
continue
|
|
first_cell = s.strip("|").split("|", 1)[0]
|
|
for m in DEM_CELL_IDS_RE.finditer(first_cell):
|
|
ids.add(f"D{m.group(1)}")
|
|
return ids
|
|
|
|
|
|
def decision_registry(root: Path) -> dict[str, set[str]]:
|
|
"""slug → {D1, D3, ...} (표 행 첫 셀 정의 기준). branch + project 노트."""
|
|
reg: dict[str, set[str]] = {}
|
|
for d in (BRANCH_DIR, PROJECT_DIR):
|
|
base = root / d
|
|
if not base.exists():
|
|
continue
|
|
for p in sorted(base.glob("*.md")):
|
|
reg[p.stem] = dem_ids_from_text(read(p))
|
|
return reg
|
|
|
|
|
|
def section_registry(root: Path) -> dict[str, set[str]]:
|
|
"""project-note slug → {§번호}."""
|
|
reg: dict[str, set[str]] = {}
|
|
base = root / PROJECT_DIR
|
|
if not base.exists():
|
|
return reg
|
|
for p in sorted(base.glob("*.md")):
|
|
nums = set()
|
|
for _, line in iter_lines_outside_fences(read(p)):
|
|
m = SECTION_DEF_RE.match(line)
|
|
if m:
|
|
nums.add(m.group(1))
|
|
reg[p.stem] = nums
|
|
return reg
|
|
|
|
|
|
def extract_refs(text: str, self_slug: str):
|
|
"""[(lineno, target_slug, d_id|None, kind)] — kind ∈ {wikilink, bare, section}."""
|
|
refs = []
|
|
for lineno, line in iter_lines_outside_fences(text):
|
|
spans = [] # wikilink 가 점유한 (start, end) — bare 매칭에서 제외
|
|
for m in WIKILINK_RE.finditer(line):
|
|
target = m.group(1).strip()
|
|
slug = target.rsplit("/", 1)[-1]
|
|
spans.append((m.start(), m.end()))
|
|
window = line[m.end(): m.end() + REF_WINDOW]
|
|
nxt = WIKILINK_RE.search(window)
|
|
if nxt:
|
|
window = window[: nxt.start()]
|
|
if slug == self_slug:
|
|
continue
|
|
is_branch = target.startswith(f"{BRANCH_DIR}/") or slug.startswith(
|
|
("feature-", "fix-", "chore-", "experiment-"))
|
|
is_project = target.startswith(f"{PROJECT_DIR}/")
|
|
if is_branch:
|
|
for d in D_TOKEN_RE.finditer(window):
|
|
refs.append((lineno, slug, f"D{d.group(1)}", "wikilink"))
|
|
if is_project:
|
|
for s in SECTION_REF_RE.finditer(window):
|
|
refs.append((lineno, slug, f"§{s.group(1)}", "section"))
|
|
for m in BARE_SLUG_RE.finditer(line):
|
|
if any(a <= m.start() < b for a, b in spans):
|
|
continue
|
|
slug = m.group(1)
|
|
if slug == self_slug:
|
|
continue
|
|
window = line[m.end(): m.end() + REF_WINDOW]
|
|
for d in D_TOKEN_RE.finditer(window):
|
|
refs.append((lineno, slug, f"D{d.group(1)}", "bare"))
|
|
break # bare 는 행당 1건만 보고 (노이즈 억제)
|
|
return refs
|
|
|
|
|
|
def coverage_rows(text: str):
|
|
"""Coverage 류 섹션의 표 행 → [(lineno, concern, status_cell, owner_cell)]."""
|
|
rows = []
|
|
lines = text.splitlines()
|
|
in_cov = False
|
|
for i, line in enumerate(lines, start=1):
|
|
if line.startswith("##"):
|
|
in_cov = bool(COVERAGE_HEADER_RE.match(line))
|
|
continue
|
|
if not in_cov or not line.strip().startswith("|"):
|
|
continue
|
|
cells = [c.strip() for c in line.strip().strip("|").split("|")]
|
|
if len(cells) < 3 or all(re.fullmatch(r":?-{3,}:?", c) for c in cells if c):
|
|
continue
|
|
if any(h in cells[0] for h in ("관심사", "concern", "Concern")):
|
|
continue # 헤더
|
|
rows.append((i, cells[0], cells[1] if len(cells) > 1 else "",
|
|
cells[2] if len(cells) > 2 else ""))
|
|
return rows
|
|
|
|
|
|
def normalize_concern(c: str) -> str:
|
|
c = CONCERN_ID_PREFIX_RE.sub("", c.strip())
|
|
return re.sub(r"[\s`*\-_/():]+", "", c).lower()
|
|
|
|
|
|
def check_file_refs(rel: str, text: str, dreg: dict, sreg: dict) -> list[tuple]:
|
|
"""(code, lineno, msg) — 참조 무결성 검사 (파일 단위).
|
|
|
|
귀속 모호성 규칙: 외부 링크 후방 윈도의 D<n> 이 *인용자 자신의* 결정 표에도
|
|
존재하면 자기-결정 언급일 수 있다 (실코퍼스: "X 에 의존 — 우회(D13)" 의 D13 이
|
|
인용자 자신의 D13). 모호 → 침묵 (의미 귀속은 Layer 2 wiki-consistency-auditor 몫).
|
|
"""
|
|
out = []
|
|
self_slug = Path(rel).stem
|
|
own_ids = dreg.get(self_slug, set()) | dem_ids_from_text(text)
|
|
for lineno, slug, ref_id, kind in extract_refs(text, self_slug):
|
|
if kind == "bare":
|
|
out.append(("BARE_DECISION_REF", lineno,
|
|
f"bare 참조 `{slug}` {ref_id} — 기계 추적을 위해 `[[{BRANCH_DIR}/{slug}]] {ref_id}` 로"))
|
|
continue
|
|
if kind == "section":
|
|
if slug in sreg and ref_id.lstrip("§") not in sreg[slug]:
|
|
out.append(("DANGLING_SECTION_REF", lineno,
|
|
f"[[{slug}]] {ref_id} — 해당 § 헤더 부재"))
|
|
continue
|
|
if slug in dreg and ref_id not in dreg[slug] and ref_id not in own_ids:
|
|
out.append(("DANGLING_DECISION_REF", lineno,
|
|
f"[[{slug}]] {ref_id} — `{slug}` 의 결정 표에 {ref_id} 없음"
|
|
f" (보유: {', '.join(sorted(dreg[slug])[:8]) or '없음'}…)"))
|
|
# slug not in dreg → 노트 부재: structure lint 의 BROKEN_LINK 몫 (중복 보고 안 함)
|
|
return out
|
|
|
|
|
|
def check_coverage(rel: str, text: str) -> list[tuple]:
|
|
out = []
|
|
for lineno, concern, status, owner in coverage_rows(text):
|
|
if "delegated" in status and owner and "—" not in owner[:2]:
|
|
if "[[" not in owner and BARE_SLUG_RE.search(owner):
|
|
out.append(("BARE_OWNER_REF", lineno,
|
|
f"delegated 행 '{concern[:40]}' 의 owner 가 bare 이름 — wikilink 필요"))
|
|
return out
|
|
|
|
|
|
def check_dual_ownership(root: Path) -> list[tuple]:
|
|
"""covered-here 관심사 정규화 exact 중복 → (code, 0, msg)."""
|
|
owners: dict[str, list[str]] = {}
|
|
base = root / BRANCH_DIR
|
|
if not base.exists():
|
|
return []
|
|
for p in sorted(base.glob("*.md")):
|
|
for _, concern, status, _ in coverage_rows(read(p)):
|
|
if "covered-here" in status:
|
|
key = normalize_concern(concern)
|
|
if key:
|
|
owners.setdefault(key, []).append(p.stem)
|
|
out = []
|
|
for key, who in sorted(owners.items()):
|
|
uniq = sorted(set(who))
|
|
if len(uniq) > 1:
|
|
out.append(("DUAL_OWNERSHIP", 0,
|
|
f"관심사 '{key[:50]}' 를 {len(uniq)}개 branch 가 covered-here 주장: {', '.join(uniq)}"))
|
|
return out
|
|
|
|
|
|
def referrers_of(root: Path, slug: str) -> list[tuple]:
|
|
"""slug 의 결정을 참조하는 문서 목록 [(rel, lineno, d_id)]."""
|
|
out = []
|
|
for d in (BRANCH_DIR, PROJECT_DIR, "wiki"):
|
|
base = root / d
|
|
if not base.exists():
|
|
continue
|
|
for p in sorted(base.rglob("*.md")):
|
|
rel = p.relative_to(root).as_posix()
|
|
if p.stem == slug:
|
|
continue
|
|
for lineno, tgt, ref_id, kind in extract_refs(read(p), p.stem):
|
|
if tgt == slug:
|
|
out.append((rel, lineno, ref_id))
|
|
return out
|
|
|
|
|
|
def _rel(p: Path | None, root: Path) -> str:
|
|
if p is None:
|
|
return ""
|
|
try:
|
|
return p.resolve().relative_to(root.resolve()).as_posix()
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
# ---------- 실행 모드 ----------
|
|
|
|
def run_all(root: Path) -> int:
|
|
dreg = decision_registry(root)
|
|
sreg = section_registry(root)
|
|
findings = check_dual_ownership(root)
|
|
by_code: dict[str, int] = {}
|
|
n_files = 0
|
|
for d in (BRANCH_DIR, PROJECT_DIR):
|
|
base = root / d
|
|
if not base.exists():
|
|
continue
|
|
for p in sorted(base.glob("*.md")):
|
|
n_files += 1
|
|
rel = p.relative_to(root).as_posix()
|
|
text = read(p)
|
|
fs = check_file_refs(rel, text, dreg, sreg) + check_coverage(rel, text)
|
|
for code, ln, msg in fs:
|
|
findings.append((code, ln, f"{rel}:{ln} {msg}" if ln else f"{rel} {msg}"))
|
|
for code, _, msg in findings:
|
|
by_code[code] = by_code.get(code, 0) + 1
|
|
print(f"[{code}] {msg}")
|
|
print(f"\n== consistency: 파일 {n_files} / findings {len(findings)} ==")
|
|
for code, n in sorted(by_code.items(), key=lambda x: -x[1]):
|
|
print(f" {n:4d} {code}")
|
|
return 1 if findings else 0
|
|
|
|
|
|
def find_dem_row(path: Path, d_id: str) -> tuple[int, str] | None:
|
|
"""owner 노트에서 d_id 를 정의하는 표 행 (lineno, line)."""
|
|
for lineno, line in iter_lines_outside_fences(read(path)):
|
|
s = line.strip()
|
|
if DEM_ROW_RE.match(s):
|
|
first_cell = s.strip("|").split("|", 1)[0]
|
|
if d_id in {f"D{m.group(1)}" for m in DEM_CELL_IDS_RE.finditer(first_cell)}:
|
|
return lineno, line
|
|
return None
|
|
|
|
|
|
def run_packets(root: Path, target: str | None) -> int:
|
|
"""T0 발췌 (P-tiering): 참조 엣지 양쪽의 ±맥락 줄을 결정론 추출 — 모델 토큰 0.
|
|
/sync 의 의미 판정(auditor/opus)이 corpus 대신 이 팩킷만 소비한다."""
|
|
files = []
|
|
for d in (BRANCH_DIR, PROJECT_DIR):
|
|
base = root / d
|
|
if base.exists():
|
|
files += sorted(base.glob("*.md"))
|
|
if target:
|
|
slug = Path(target).stem
|
|
files = [p for p in files if p.stem == slug] or files # citing 파일 스코프
|
|
files = [p for p in files if p.stem == slug]
|
|
n = 0
|
|
print("# Consistency Edge Packets (결정론 추출 — 의미 판정 입력)")
|
|
for p in files:
|
|
rel = p.relative_to(root).as_posix()
|
|
text = read(p)
|
|
lines = text.splitlines()
|
|
for lineno, slug, ref_id, kind in extract_refs(text, p.stem):
|
|
if kind != "wikilink":
|
|
continue
|
|
owner = root / BRANCH_DIR / f"{slug}.md"
|
|
if not owner.exists():
|
|
owner = root / PROJECT_DIR / f"{slug}.md"
|
|
if not owner.exists():
|
|
continue
|
|
n += 1
|
|
print(f"\n## Edge {n}: {rel}:{lineno} → [[{slug}]] {ref_id}")
|
|
print("### citing 측 (±2줄)")
|
|
for i in range(max(1, lineno - 2), min(len(lines), lineno + 2) + 1):
|
|
print(f" {rel}:{i}: {lines[i - 1][:300]}")
|
|
if ref_id.startswith("D"):
|
|
row = find_dem_row(owner, ref_id)
|
|
print("### owner 측 (D-row)")
|
|
if row:
|
|
print(f" {owner.relative_to(root).as_posix()}:{row[0]}: {row[1][:500]}")
|
|
else:
|
|
print(f" (D-row 미발견 — DANGLING 후보, --all 로 확인)")
|
|
print(f"\n== packets: {n} edges ==")
|
|
return 0
|
|
|
|
|
|
def run_impact(root: Path, target: str) -> int:
|
|
slug = Path(target).stem
|
|
refs = referrers_of(root, slug)
|
|
if not refs:
|
|
print(f"참조자 없음: {slug} 의 결정을 인용하는 문서가 없다")
|
|
return 0
|
|
print(f"== `{slug}` 의 결정을 참조하는 문서 {len(set(r[0] for r in refs))}개 / 참조 {len(refs)}건 ==")
|
|
for rel, ln, d in refs:
|
|
print(f" {rel}:{ln} → {d}")
|
|
return 0
|
|
|
|
|
|
def run_pre(event: dict, root: Path) -> int:
|
|
"""PreToolUse: 새로 쓰는 본문의 DANGLING 참조 차단 (타깃 노트 실존 시에만)."""
|
|
inp = wiki_rules.tool_input(event)
|
|
p = wiki_rules.target_path(inp)
|
|
rel = _rel(p, root)
|
|
if not rel or not (rel.startswith(BRANCH_DIR) or rel.startswith(PROJECT_DIR)):
|
|
return 0
|
|
text = wiki_rules.projected_content(p, inp)
|
|
if "[[" not in text:
|
|
return 0
|
|
dreg = decision_registry(root)
|
|
sreg = section_registry(root)
|
|
# 자기 자신의 projected 결정 표를 registry 에 반영 (자기 D 정의 동시 추가 케이스)
|
|
dreg[Path(rel).stem] = dem_ids_from_text(text)
|
|
bad = [f for f in check_file_refs(rel, text, dreg, sreg)
|
|
if f[0] in ("DANGLING_DECISION_REF", "DANGLING_SECTION_REF")]
|
|
if bad:
|
|
print(f"✗ wiki-consistency (pre): {rel} — 깨진 결정 참조 {len(bad)}건 → 쓰기 차단",
|
|
file=sys.stderr)
|
|
for code, ln, msg in bad[:8]:
|
|
print(f" [{code}]:{ln} {msg}", file=sys.stderr)
|
|
print(" owner 노트의 실제 Decision ID 를 확인하거나, 결정이 아직 없으면 owner 노트에 먼저 기록하세요"
|
|
" (rules/consistency-contract.md).", file=sys.stderr)
|
|
return 2
|
|
return 0
|
|
|
|
|
|
DEM_EDIT_SIGNAL_RE = re.compile(r"\|\s*\*{0,2}D\d+\*{0,2}\s*\||Decision Evidence Map")
|
|
|
|
|
|
def run_post(event: dict, root: Path) -> int:
|
|
"""PostToolUse: 결정 표를 건드린 편집이면 역참조 충격 알림 (비차단 — 쓰기는 완료됨)."""
|
|
inp = wiki_rules.tool_input(event)
|
|
p = wiki_rules.target_path(inp)
|
|
rel = _rel(p, root)
|
|
if not rel or not rel.startswith(BRANCH_DIR):
|
|
return 0
|
|
touched = ""
|
|
for k in ("old_string", "new_string", "content"):
|
|
v = inp.get(k)
|
|
if isinstance(v, str):
|
|
touched += v + "\n"
|
|
if not DEM_EDIT_SIGNAL_RE.search(touched):
|
|
return 0
|
|
slug = Path(rel).stem
|
|
refs = referrers_of(root, slug)
|
|
if not refs:
|
|
return 0
|
|
docs = sorted(set(r[0] for r in refs))
|
|
print(f"⚠ 역참조 충격 알림 (비차단 — 쓰기 완료됨): `{slug}` 의 결정 표를 수정했고, "
|
|
f"이 노트의 결정을 참조하는 문서 {len(docs)}개가 있다:", file=sys.stderr)
|
|
for rel2, ln, d in refs[:10]:
|
|
print(f" {rel2}:{ln} → {d}", file=sys.stderr)
|
|
if len(refs) > 10:
|
|
print(f" … 외 {len(refs) - 10}건", file=sys.stderr)
|
|
print(" 변경이 D-row 의 의미를 바꿨다면 참조 요약이 낡았을 수 있다 — 같은 세션에서 갱신하거나 `/sync` 로 대조"
|
|
" (rules/consistency-contract.md §전파).", file=sys.stderr)
|
|
return 2 # PostToolUse exit 2 = 모델에 stderr 전달 (이미 완료된 쓰기를 막지 않음)
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser(description="문서 간 모순·동기화 결정론 검사기")
|
|
ap.add_argument("--all", action="store_true")
|
|
ap.add_argument("--impact", metavar="SLUG_OR_PATH")
|
|
ap.add_argument("--packets", nargs="?", const="", metavar="SLUG",
|
|
help="참조 엣지 양쪽 ±맥락 결정론 추출 (T0 발췌 — /sync 의미 판정 입력). SLUG 생략 시 전체")
|
|
ap.add_argument("--pre", action="store_true")
|
|
ap.add_argument("--post", action="store_true")
|
|
ap.add_argument("--root", default=str(DEFAULT_ROOT))
|
|
args = ap.parse_args()
|
|
root = Path(args.root).resolve()
|
|
|
|
if args.packets is not None:
|
|
sys.exit(run_packets(root, args.packets or None))
|
|
|
|
if args.pre or args.post:
|
|
import json
|
|
try:
|
|
raw = sys.stdin.read()
|
|
event = json.loads(raw) if raw.strip() else {}
|
|
except Exception:
|
|
sys.exit(0)
|
|
sys.exit(run_pre(event, root) if args.pre else run_post(event, root))
|
|
if args.impact:
|
|
sys.exit(run_impact(root, args.impact))
|
|
if args.all:
|
|
sys.exit(run_all(root))
|
|
ap.error("--all / --impact / --pre / --post 중 하나 필요")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|