371 lines
15 KiB
Python
371 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Fail-closed, read-only quality gate for staged wiki document writes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from dataclasses import dataclass
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
from typing import Any, Callable, Iterable, Mapping
|
|
|
|
import moc_indexer
|
|
from contract_markdown import parse_frontmatter
|
|
import template_renderer
|
|
|
|
|
|
DEFAULT_ROOT = Path(__file__).resolve().parents[2]
|
|
SECTION_ID_RE = re.compile(r"^\s*<!--\s*section-id:\s*([a-z0-9][a-z0-9-]*)\s*-->\s*$", re.MULTILINE)
|
|
UNRESOLVED_RE = re.compile(r"\{\{[^{}\n]+\}\}")
|
|
TRANSPORT_RE = re.compile(
|
|
r"(?:</?content>|</?file(?:\s[^>]*)?>|^(?:<<<<<<<(?:\s.*)?|=======|>>>>>>>(?:\s.*)?)$)",
|
|
re.MULTILINE,
|
|
)
|
|
BRANCH_SECTION_ORDER = (
|
|
"branch-parent",
|
|
"branch-contract-packet",
|
|
"inherited-project-decisions",
|
|
"branch-local-decisions",
|
|
"declared-overrides",
|
|
"branch-goal",
|
|
"branch-scope",
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class QualityExtension:
|
|
"""A fail-closed external gate evaluated against the staged repository.
|
|
|
|
Typed contracts, projections, and semantic certificates can plug into the
|
|
common gate without creating imports from quality_gate back into those
|
|
independently versioned runtimes.
|
|
"""
|
|
|
|
name: str
|
|
runner: Callable[[Path], Mapping[str, Any]]
|
|
required: bool = True
|
|
|
|
|
|
class QualityGateError(RuntimeError):
|
|
"""A schema, I/O, or checker-loading failure prevented a quality decision."""
|
|
|
|
|
|
def _module(name: str, path: Path) -> Any:
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
if spec is None or spec.loader is None:
|
|
raise QualityGateError(f"checker cannot be loaded: {path}")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def _resolved_paths(root: Path, values: Iterable[str | Path]) -> list[Path]:
|
|
resolved: set[Path] = set()
|
|
for value in values:
|
|
path = Path(value)
|
|
path = path if path.is_absolute() else root / path
|
|
path = path.resolve(strict=True)
|
|
try:
|
|
path.relative_to(root)
|
|
except ValueError as exc:
|
|
raise QualityGateError(f"path escapes staged root: {path}") from exc
|
|
if not path.is_file():
|
|
raise QualityGateError(f"touched path is not a file: {path}")
|
|
resolved.add(path)
|
|
if not resolved:
|
|
raise QualityGateError("at least one touched path is required")
|
|
return sorted(resolved)
|
|
|
|
|
|
def _finding(
|
|
code: str,
|
|
path: str,
|
|
message: str,
|
|
line: int = 0,
|
|
*,
|
|
check: str = "",
|
|
) -> dict[str, Any]:
|
|
result: dict[str, Any] = {"code": code, "path": path, "line": line, "message": message}
|
|
if check:
|
|
result["check"] = check
|
|
return result
|
|
|
|
|
|
def _extension_result(
|
|
staged_root: Path,
|
|
extension: QualityExtension,
|
|
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
|
if not re.fullmatch(r"[a-z][a-z0-9-]*", extension.name):
|
|
raise QualityGateError(f"invalid extension name: {extension.name!r}")
|
|
raw = extension.runner(staged_root)
|
|
if not isinstance(raw, Mapping):
|
|
raise QualityGateError(f"extension {extension.name} returned a non-object")
|
|
status = raw.get("status")
|
|
if status not in {"PASS", "FAIL", "SKIP"}:
|
|
raise QualityGateError(f"extension {extension.name} returned invalid status: {status!r}")
|
|
raw_findings = raw.get("findings", [])
|
|
if not isinstance(raw_findings, list) or any(not isinstance(item, Mapping) for item in raw_findings):
|
|
raise QualityGateError(f"extension {extension.name} findings must be an array of objects")
|
|
findings: list[dict[str, Any]] = []
|
|
for item in raw_findings:
|
|
findings.append(
|
|
_finding(
|
|
str(item.get("code", "EXTERNAL_CHECK_FAILED")),
|
|
str(item.get("path", "<repository>")),
|
|
str(item.get("message", item.get("code", "external check failed"))),
|
|
int(item.get("line", 0) or 0),
|
|
check=extension.name,
|
|
)
|
|
)
|
|
if status == "FAIL" and not findings:
|
|
findings.append(
|
|
_finding(
|
|
"EXTERNAL_CHECK_FAILED",
|
|
"<repository>",
|
|
f"{extension.name} returned FAIL without findings",
|
|
check=extension.name,
|
|
)
|
|
)
|
|
if status == "SKIP" and extension.required:
|
|
findings.append(
|
|
_finding(
|
|
"REQUIRED_EXTENSION_SKIPPED",
|
|
"<repository>",
|
|
f"required extension was skipped: {extension.name}",
|
|
check=extension.name,
|
|
)
|
|
)
|
|
effective_status = "FAIL" if findings else status
|
|
return {
|
|
"name": extension.name,
|
|
"status": effective_status,
|
|
"finding_count": len(findings),
|
|
"schema_version": raw.get("schema_version"),
|
|
"required": extension.required,
|
|
}, findings
|
|
|
|
|
|
def scan_hygiene(paths: Iterable[Path], *, root: Path | None = None) -> list[dict[str, Any]]:
|
|
"""Return transport-wrapper findings for the supplied text files."""
|
|
base = root.resolve() if root is not None else None
|
|
findings: list[dict[str, Any]] = []
|
|
for path in sorted(set(paths)):
|
|
text = path.read_text(encoding="utf-8")
|
|
rel = path.relative_to(base).as_posix() if base is not None else path.as_posix()
|
|
for match in TRANSPORT_RE.finditer(text):
|
|
findings.append(
|
|
_finding(
|
|
"SOURCE_HYGIENE_VIOLATION",
|
|
rel,
|
|
f"transport wrapper token is forbidden: {match.group(0)[:80]!r}",
|
|
text.count("\n", 0, match.start()) + 1,
|
|
)
|
|
)
|
|
return findings
|
|
|
|
|
|
def run(
|
|
root: Path,
|
|
touched_paths: Iterable[str | Path],
|
|
*,
|
|
structure_paths: Iterable[str | Path] | None = None,
|
|
template_root: Path | None = None,
|
|
include_graph: bool = True,
|
|
require_moc_convergence: bool = True,
|
|
extensions: Iterable[QualityExtension] = (),
|
|
) -> dict[str, Any]:
|
|
"""Validate staged bytes without modifying them.
|
|
|
|
The caller is responsible for committing only after this function returns
|
|
``status=PASS``. Operational failures raise :class:`QualityGateError` so a
|
|
gateway cannot accidentally treat an incomplete check as a quality finding.
|
|
"""
|
|
try:
|
|
staged_root = root.resolve(strict=True)
|
|
templates = (template_root or DEFAULT_ROOT).resolve(strict=True)
|
|
touched = _resolved_paths(staged_root, touched_paths)
|
|
structure = _resolved_paths(staged_root, structure_paths if structure_paths is not None else touched_paths)
|
|
before = {path: hashlib.sha256(path.read_bytes()).hexdigest() for path in touched}
|
|
findings: list[dict[str, Any]] = []
|
|
checks: list[dict[str, Any]] = []
|
|
hygiene = scan_hygiene(touched, root=staged_root)
|
|
for item in hygiene:
|
|
item["check"] = "source-hygiene"
|
|
findings.extend(hygiene)
|
|
checks.append({"name": "source-hygiene", "status": "FAIL" if hygiene else "PASS", "finding_count": len(hygiene)})
|
|
|
|
document_findings: list[dict[str, Any]] = []
|
|
|
|
for path in touched:
|
|
rel = path.relative_to(staged_root).as_posix()
|
|
text = path.read_text(encoding="utf-8")
|
|
for match in UNRESOLVED_RE.finditer(text):
|
|
document_findings.append(
|
|
_finding(
|
|
"UNRESOLVED_PLACEHOLDER",
|
|
rel,
|
|
match.group(0),
|
|
text.count("\n", 0, match.start()) + 1,
|
|
)
|
|
)
|
|
section_ids = SECTION_ID_RE.findall(text)
|
|
duplicates = sorted({item for item in section_ids if section_ids.count(item) > 1})
|
|
for section_id in duplicates:
|
|
document_findings.append(_finding("DUPLICATE_SECTION_ID", rel, section_id))
|
|
if "contract_packet" in parse_frontmatter(text) or "branch-contract-packet" in section_ids:
|
|
observed = [item for item in section_ids if item in BRANCH_SECTION_ORDER]
|
|
if tuple(observed) != BRANCH_SECTION_ORDER:
|
|
document_findings.append(
|
|
_finding(
|
|
"SECTION_ID_ORDER",
|
|
rel,
|
|
f"expected {list(BRANCH_SECTION_ORDER)}, observed {observed}",
|
|
)
|
|
)
|
|
try:
|
|
template_renderer.generated_region(text)
|
|
except template_renderer.TemplateRenderError as exc:
|
|
document_findings.append(_finding(exc.code, rel, str(exc)))
|
|
|
|
# 한국어 문체·자연스러움 검사는 이 하네스의 책임이 아니다.
|
|
# 별도 하네스 im-not-ai(`/humanize-korean`)가 문서 작성이 끝난 뒤 일괄 처리한다.
|
|
|
|
for item in document_findings:
|
|
item["check"] = "document-schema"
|
|
findings.extend(document_findings)
|
|
checks.append({"name": "document-schema", "status": "FAIL" if document_findings else "PASS", "finding_count": len(document_findings)})
|
|
|
|
structure_lint = _module(
|
|
"wiki_structure_lint_quality_gate",
|
|
templates / ".claude/hooks/wiki_structure_lint.py",
|
|
)
|
|
by_st, by_file = structure_lint.build_template_index(templates)
|
|
vault_paths, vault_bases = structure_lint.build_vault_index(staged_root)
|
|
cache: dict[Path, str] = {}
|
|
structure_findings: list[dict[str, Any]] = []
|
|
for path in structure:
|
|
rel = path.relative_to(staged_root).as_posix()
|
|
mode = structure_lint.classify(rel, staged_root)
|
|
lint_findings, _source_type = structure_lint.lint_file(
|
|
path,
|
|
staged_root,
|
|
by_st,
|
|
by_file,
|
|
vault_paths,
|
|
vault_bases,
|
|
cache,
|
|
mode=mode,
|
|
)
|
|
structure_findings.extend(_finding(code, rel, message, line, check="structure") for code, line, message in lint_findings)
|
|
findings.extend(structure_findings)
|
|
checks.append({"name": "structure", "status": "FAIL" if structure_findings else "PASS", "finding_count": len(structure_findings)})
|
|
|
|
if require_moc_convergence:
|
|
moc_findings: list[dict[str, Any]] = []
|
|
moc_updates, _moc_stats = moc_indexer.build_updates(staged_root)
|
|
if moc_updates:
|
|
for path in sorted(moc_updates):
|
|
moc_findings.append(
|
|
_finding(
|
|
"MOC_NOT_CONVERGED",
|
|
path.relative_to(staged_root).as_posix(),
|
|
"generated reverse view differs from canonical child edges",
|
|
check="moc",
|
|
)
|
|
)
|
|
findings.extend(moc_findings)
|
|
checks.append({"name": "moc", "status": "FAIL" if moc_findings else "PASS", "finding_count": len(moc_findings)})
|
|
else:
|
|
checks.append({"name": "moc", "status": "SKIP", "finding_count": 0})
|
|
|
|
if include_graph:
|
|
graph = _module(
|
|
"wiki_graph_contract_check_quality_gate",
|
|
templates / ".claude/hooks/wiki_graph_contract_check.py",
|
|
)
|
|
graph_findings, _warnings, _stats = graph.scan(staged_root, include_expected_edges=True)
|
|
normalized_graph = [_finding(code, rel, message, line, check="graph") for code, rel, line, message in graph_findings]
|
|
findings.extend(normalized_graph)
|
|
checks.append({"name": "graph", "status": "FAIL" if normalized_graph else "PASS", "finding_count": len(normalized_graph)})
|
|
else:
|
|
checks.append({"name": "graph", "status": "SKIP", "finding_count": 0})
|
|
|
|
seen_extensions: set[str] = set()
|
|
for extension in extensions:
|
|
if not isinstance(extension, QualityExtension):
|
|
raise QualityGateError("extensions must contain QualityExtension values")
|
|
if extension.name in seen_extensions:
|
|
raise QualityGateError(f"duplicate extension name: {extension.name}")
|
|
seen_extensions.add(extension.name)
|
|
check_result, extension_findings = _extension_result(staged_root, extension)
|
|
checks.append(check_result)
|
|
findings.extend(extension_findings)
|
|
|
|
after = {path: hashlib.sha256(path.read_bytes()).hexdigest() for path in touched}
|
|
for path in touched:
|
|
if before[path] != after[path]:
|
|
findings.append(
|
|
_finding(
|
|
"TOUCHED_PATH_MUTATED_DURING_GATE",
|
|
path.relative_to(staged_root).as_posix(),
|
|
"a read-only checker changed staged bytes",
|
|
check="touched-paths",
|
|
)
|
|
)
|
|
|
|
mutation_count = sum(1 for item in findings if item.get("check") == "touched-paths")
|
|
checks.append({"name": "touched-paths", "status": "FAIL" if mutation_count else "PASS", "finding_count": mutation_count})
|
|
|
|
findings.sort(key=lambda item: (item["path"], item["line"], item["code"], item["message"]))
|
|
return {
|
|
"schema_version": "quality-gate-result/v1",
|
|
"status": "PASS" if not findings else "FAIL",
|
|
"findings": findings,
|
|
"checked_paths": [path.relative_to(staged_root).as_posix() for path in touched],
|
|
"touched_paths": [path.relative_to(staged_root).as_posix() for path in touched],
|
|
"checks": checks,
|
|
}
|
|
except QualityGateError:
|
|
raise
|
|
except (OSError, UnicodeError, ValueError, ImportError) as exc:
|
|
raise QualityGateError(str(exc)) from exc
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--root", type=Path, default=DEFAULT_ROOT)
|
|
parser.add_argument("--path", action="append", required=True)
|
|
parser.add_argument(
|
|
"--branch-scope",
|
|
action="store_true",
|
|
help="validate project/branch MOC through the graph gate without the R2 all-document relation index",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
result = run(args.root, args.path, require_moc_convergence=not args.branch_scope)
|
|
json.dump(result, sys.stdout, ensure_ascii=False, indent=2, sort_keys=True)
|
|
sys.stdout.write("\n")
|
|
return 0 if result["status"] == "PASS" else 1
|
|
except QualityGateError as exc:
|
|
json.dump(
|
|
{
|
|
"schema_version": "quality-gate-result/v1",
|
|
"status": "ERROR",
|
|
"errors": [{"code": "QUALITY_GATE_ERROR", "message": str(exc)}],
|
|
},
|
|
sys.stdout,
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
sort_keys=True,
|
|
)
|
|
sys.stdout.write("\n")
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|