Michał Kopeć
474630e2bb
Graft keeps a derived, disposable code graph in sync with a content hash rather than a calendar, and keeps a protected block on every regenerated node. This KB is the opposite kind of store — durable, curated, built from material that cannot be regenerated — but several of Graft's mechanisms port cleanly, and two of them close real gaps here. Schema 1.5 is additive: every 1.4 page remains valid. 1. `## Crux` — verbatim source excerpts alongside the synthesis. A summary can drift silently; a quote either still matches its source or it does not. Lets `ckb-retrieve` ground an answer without a round-trip to the archive, and makes drift mechanically detectable. 2. `## Notes` — human-authored and protected everywhere. Closes a real gap: `ckb-index-external` regenerates connector pages wholesale, so an annotation written there was previously destroyed on the next refresh. 3. `source_fingerprint`/`source_checked` — a digest of the material a page was built from. Freshness by date says a page has aged; a fingerprint says whether its evidence moved. Most valuable for connector-backed libs, where documents change with no notice. 4. `lint_report.py --quick` — a deterministic one-line session-start signal, wired into Rule E next to the existing `git status` check. 5. In-degree as a rank-fusion signal in `ckb-retrieve`, weighted below 1.0: centrality is a prior, not evidence. 6. Blast radius — a new `ckb-ingest` step walking the graph backwards from touched entities to find what the incoming material contradicts, before writing anything. Ingest was additive-first, which is how a wiki accumulates two pages that quietly disagree. 7. Edge vocabulary in `wiki/graph/index.md` rewritten as a question per verb, and completed: `part_of` was written by `ckb-code-map` but never declared. Added `produces`, `configures`, `validates`, `implements`. Lint gains checks 12 (fingerprint drift), 13 (crux verbatimness) and 14 (the protected-Notes rule), verified against a synthetic fixture covering stale digests, missing sources, fabricated quotes and paraphrased evidence. Not adopted: the gitignored regenerable store, the MCP server and CLI daemon, tree-sitter parsing, statusline hooks, telemetry. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
793 lines
31 KiB
Python
793 lines
31 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright 2026 Michał Kopeć
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
"""Deterministic half of the ckb-lint health check: detect, never fix.
|
|
|
|
Usage:
|
|
python3 lint_report.py [--scope wiki|libs|all] [--today YYYY-MM-DD] [--json] [--quick]
|
|
|
|
Run from the repository root (the directory containing wiki/ and libs/).
|
|
|
|
This script is strictly READ-ONLY. It writes no files, moves nothing, and runs
|
|
no git commands — it prints findings for the agent to act on. The mechanical
|
|
checks (conformance, freshness, retention/decay candidates, orphans, graph
|
|
consistency, index/log consistency, source.yaml validity, source fingerprints,
|
|
crux verbatimness) live here; the
|
|
judgment calls the ckb-lint skill owns (whether two pages genuinely supersede
|
|
one another, ambiguous orphans, error-book entries, auto-fix vs. report) stay
|
|
with the model. Decision records get their own structural checks here — status
|
|
vocabulary, required dates, two-sided supersession links — since those are
|
|
mechanical; deciding that a new decision *replaces* an old one is not.
|
|
|
|
Exit code is 0 when nothing was found and 1 when there are findings.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sys
|
|
from datetime import date, datetime
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path.cwd()
|
|
WIKI = REPO_ROOT / "wiki"
|
|
LIBS = REPO_ROOT / "libs"
|
|
|
|
RESERVED = {"index.md", "log.md"}
|
|
DEFAULT_REFRESH_DAYS = 30
|
|
DECISION_STATUSES = {"proposed", "accepted", "rejected", "superseded", "reversed"}
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# tiny YAML readers (stdlib only — this template ships without dependencies)
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def parse_frontmatter(text):
|
|
"""Return (dict, body, error). `error` is a string when the frontmatter is
|
|
present but unparseable, else None. Flat `key: value` pairs only."""
|
|
if not text.startswith("---\n"):
|
|
return {}, text, None
|
|
end = text.find("\n---\n", 4)
|
|
if end == -1:
|
|
return {}, text, "frontmatter opened with `---` but never closed"
|
|
raw, body = text[4:end], text[end + 5 :]
|
|
fm = {}
|
|
for lineno, line in enumerate(raw.split("\n"), start=2):
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#"):
|
|
continue
|
|
if line[:1].isspace():
|
|
return {}, body, f"line {lineno}: nested/indented YAML is not supported in page frontmatter"
|
|
if ":" not in stripped:
|
|
return {}, body, f"line {lineno}: not a `key: value` pair: {stripped!r}"
|
|
key, _, value = stripped.partition(":")
|
|
fm[key.strip()] = unquote(value.strip())
|
|
return fm, body, None
|
|
|
|
|
|
def unquote(value):
|
|
if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
|
|
return value[1:-1]
|
|
return value
|
|
|
|
|
|
def parse_simple_yaml(text):
|
|
"""Two-level `key: value` / `key:` + indented block YAML, enough for
|
|
source.yaml and source.local.yaml. Returns a dict; nested blocks become
|
|
nested dicts."""
|
|
out, current = {}, None
|
|
for line in text.split("\n"):
|
|
if not line.strip() or line.strip().startswith("#"):
|
|
continue
|
|
indented = line[:1].isspace()
|
|
stripped = line.strip()
|
|
if ":" not in stripped:
|
|
continue
|
|
key, _, value = stripped.partition(":")
|
|
key, value = key.strip(), unquote(value.split(" #", 1)[0].strip())
|
|
if indented:
|
|
if isinstance(current, dict):
|
|
current[key] = value
|
|
continue
|
|
if value == "":
|
|
current = {}
|
|
out[key] = current
|
|
else:
|
|
out[key] = value
|
|
current = None
|
|
# `key:` with nothing indented under it is an empty scalar, not a block.
|
|
return {k: ("" if v == {} else v) for k, v in out.items()}
|
|
|
|
|
|
def as_float(value):
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def as_int(value):
|
|
try:
|
|
return int(str(value).strip())
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def as_date(value):
|
|
try:
|
|
return datetime.strptime(str(value).strip(), "%Y-%m-%d").date()
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# tree model
|
|
# --------------------------------------------------------------------------
|
|
|
|
LINK_RE = re.compile(r"\[[^\]]*\]\(([^)]+)\)")
|
|
WIKILINK_RE = re.compile(r"\[\[([^\]]+)\]\]")
|
|
SOURCE_BULLET_RE = re.compile(r"^\s*[-*]\s+`([^`]+)`(.*)$", re.M)
|
|
FINGERPRINT_RE = re.compile(r"\b(sha256|etag|mtime):([^\s,)]+)")
|
|
ATTRIBUTION_RE = re.compile(r"^\s*[\u2014-]\s*`([^`]+)`", re.M)
|
|
# Text-ish sources a quote can actually be checked against byte-for-byte.
|
|
# Anything else (pdf, docx, audio) is fingerprinted but never quote-verified.
|
|
QUOTABLE_SUFFIXES = {".md", ".txt", ".eml", ".csv", ".json", ".yaml", ".yml", ".html", ".rst", ".log"}
|
|
|
|
|
|
class Doc:
|
|
def __init__(self, root, path):
|
|
self.path = path
|
|
self.rel = path.relative_to(root)
|
|
text = path.read_text(encoding="utf-8", errors="replace")
|
|
self.fm, self.body, self.fm_error = parse_frontmatter(text)
|
|
self.reserved = path.name in RESERVED
|
|
|
|
def links(self):
|
|
"""Every intra-tree link target as a path relative to the tree root."""
|
|
out = set()
|
|
for target in LINK_RE.findall(self.body):
|
|
target = target.split("#", 1)[0].strip()
|
|
if not target or re.match(r"^[a-z][a-z0-9+.-]*:", target):
|
|
continue
|
|
if target.startswith("/wiki/"):
|
|
out.add(target[len("/wiki/") :])
|
|
elif target.startswith("/"):
|
|
out.add(target.lstrip("/"))
|
|
else:
|
|
try:
|
|
resolved = (self.rel.parent / target).as_posix()
|
|
except ValueError:
|
|
continue
|
|
parts = []
|
|
for part in resolved.split("/"):
|
|
if part == "..":
|
|
if parts:
|
|
parts.pop()
|
|
elif part not in (".", ""):
|
|
parts.append(part)
|
|
out.add("/".join(parts))
|
|
return out
|
|
|
|
def wikilinks(self):
|
|
return {w.split("|", 1)[-1].strip() for w in WIKILINK_RE.findall(self.body)}
|
|
|
|
|
|
def read_tree(root):
|
|
docs = []
|
|
for path in sorted(root.rglob("*.md")):
|
|
if any(part.startswith(".") for part in path.relative_to(root).parts):
|
|
continue
|
|
docs.append(Doc(root, path))
|
|
return docs
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# checks (each returns a list of finding strings)
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def check_conformance(docs):
|
|
findings = []
|
|
for doc in docs:
|
|
if doc.reserved:
|
|
continue
|
|
if doc.fm_error:
|
|
findings.append(f"{doc.rel}: unparseable frontmatter — {doc.fm_error}")
|
|
elif not doc.fm:
|
|
findings.append(f"{doc.rel}: no frontmatter")
|
|
elif not doc.fm.get("type", "").strip():
|
|
findings.append(f"{doc.rel}: missing or empty `type`")
|
|
return findings
|
|
|
|
|
|
def check_freshness(docs, today):
|
|
findings = []
|
|
for doc in docs:
|
|
if doc.reserved or doc.fm_error:
|
|
continue
|
|
window = as_int(doc.fm.get("freshness_window_days"))
|
|
updated = as_date(doc.fm.get("last_updated"))
|
|
if window is None or updated is None:
|
|
if doc.fm and doc.fm.get("last_updated") and updated is None:
|
|
findings.append(f"{doc.rel}: `last_updated` is not a YYYY-MM-DD date: {doc.fm['last_updated']!r}")
|
|
continue
|
|
age = (today - updated).days
|
|
if age > window:
|
|
findings.append(f"{doc.rel}: stale — {age}d since last_updated, window is {window}d ({age - window}d over)")
|
|
return findings
|
|
|
|
|
|
def check_confidence(docs):
|
|
findings = []
|
|
for doc in docs:
|
|
if doc.reserved or doc.fm_error or "confidence" not in doc.fm:
|
|
continue
|
|
value = as_float(doc.fm["confidence"])
|
|
if value is None:
|
|
findings.append(f"{doc.rel}: `confidence` is not a number: {doc.fm['confidence']!r}")
|
|
elif value < 0.3:
|
|
findings.append(f"{doc.rel}: confidence {value} is below 0.3 — flag for re-review")
|
|
return findings
|
|
|
|
|
|
def check_retention(docs, today):
|
|
findings = []
|
|
for doc in docs:
|
|
if doc.reserved or doc.fm_error:
|
|
continue
|
|
if doc.fm.get("retention", "").strip().lower() != "low":
|
|
continue
|
|
window = as_int(doc.fm.get("freshness_window_days"))
|
|
updated = as_date(doc.fm.get("last_updated"))
|
|
if window is None or updated is None:
|
|
continue
|
|
age = (today - updated).days
|
|
if age > 2 * window:
|
|
findings.append(
|
|
f"{doc.rel}: archive candidate — retention: low, {age}d old, 2x window is {2 * window}d"
|
|
)
|
|
return findings
|
|
|
|
|
|
def check_decisions(docs, root, today):
|
|
"""Decision-record specific rules (see the ckb-decide skill)."""
|
|
findings = []
|
|
decisions = [d for d in docs if not d.reserved and not d.fm_error
|
|
and d.fm.get("type", "").strip().lower() == "decision"]
|
|
by_path = {d.rel.as_posix(): d for d in decisions}
|
|
seen_numbers = {}
|
|
|
|
for doc in sorted(decisions, key=lambda d: d.rel.as_posix()):
|
|
rel = doc.rel.as_posix()
|
|
status = doc.fm.get("status", "").strip().lower()
|
|
if not status:
|
|
findings.append(f"{rel}: decision has no `status`")
|
|
elif status not in DECISION_STATUSES:
|
|
findings.append(
|
|
f"{rel}: `status: {status}` is not one of {', '.join(sorted(DECISION_STATUSES))}"
|
|
)
|
|
|
|
decided_on = as_date(doc.fm.get("decided_on"))
|
|
if status in ("accepted", "rejected", "reversed"):
|
|
if not doc.fm.get("decided_on"):
|
|
findings.append(f"{rel}: `status: {status}` but no `decided_on` date")
|
|
elif decided_on is None:
|
|
findings.append(f"{rel}: `decided_on` is not a YYYY-MM-DD date: {doc.fm['decided_on']!r}")
|
|
if decided_on and decided_on > today:
|
|
findings.append(f"{rel}: `decided_on` is in the future: {decided_on.isoformat()}")
|
|
if not doc.fm.get("decided_by", "").strip():
|
|
findings.append(f"{rel}: no `decided_by` — record `unknown` rather than omitting it")
|
|
|
|
review = doc.fm.get("review_on")
|
|
if review:
|
|
review_date = as_date(review)
|
|
if review_date is None:
|
|
findings.append(f"{rel}: `review_on` is not a YYYY-MM-DD date: {review!r}")
|
|
elif review_date <= today:
|
|
findings.append(
|
|
f"{rel}: due for review — `review_on: {review_date.isoformat()}` "
|
|
f"passed {(today - review_date).days}d ago"
|
|
)
|
|
|
|
# Supersession links must resolve and must be reciprocal.
|
|
for field, mirror in (("supersedes", "superseded_by"), ("superseded_by", "supersedes")):
|
|
target = doc.fm.get(field, "").strip()
|
|
if not target:
|
|
continue
|
|
key = target[len("/wiki/") :] if target.startswith("/wiki/") else target.lstrip("/")
|
|
other = by_path.get(key)
|
|
if other is None:
|
|
if not (root / key).is_file():
|
|
findings.append(f"{rel}: `{field}: {target}` does not resolve to a page")
|
|
continue
|
|
back = other.fm.get(mirror, "").strip()
|
|
back_key = back[len("/wiki/") :] if back.startswith("/wiki/") else back.lstrip("/")
|
|
if back_key != rel:
|
|
findings.append(
|
|
f"{rel}: `{field}` points at {key}, but that page's `{mirror}` "
|
|
f"does not point back ({back or 'unset'}) — supersession must be two-sided"
|
|
)
|
|
|
|
if status in ("superseded", "reversed") and not doc.fm.get("superseded_by", "").strip():
|
|
findings.append(f"{rel}: `status: {status}` but no `superseded_by` naming what replaced it")
|
|
if doc.fm.get("superseded_by", "").strip() and status not in ("superseded", "reversed"):
|
|
findings.append(
|
|
f"{rel}: has `superseded_by` but `status: {status or 'unset'}` — "
|
|
"expected `superseded` or `reversed`"
|
|
)
|
|
|
|
for target in [t.strip() for t in doc.fm.get("affects", "").split(",") if t.strip()]:
|
|
key = target[len("/wiki/") :] if target.startswith("/wiki/") else target.lstrip("/")
|
|
if not (root / key).is_file():
|
|
findings.append(f"{rel}: `affects` entry does not resolve to a page: {target}")
|
|
|
|
m = re.match(r"^(\d{4})-", doc.rel.name)
|
|
if m:
|
|
seen_numbers.setdefault(m.group(1), []).append(rel)
|
|
else:
|
|
findings.append(f"{rel}: decision filename does not start with a four-digit number")
|
|
|
|
for number, paths in sorted(seen_numbers.items()):
|
|
if len(paths) > 1:
|
|
findings.append(f"decision number {number} used by more than one page: {', '.join(sorted(paths))}")
|
|
return findings
|
|
|
|
|
|
def check_orphans(docs):
|
|
inbound = set()
|
|
titles = {}
|
|
for doc in docs:
|
|
rel = doc.rel.as_posix()
|
|
titles[rel] = rel
|
|
stem = doc.rel.with_suffix("").as_posix()
|
|
titles[stem] = rel
|
|
for doc in docs:
|
|
for target in doc.links():
|
|
if target in titles:
|
|
inbound.add(titles[target])
|
|
for wl in doc.wikilinks():
|
|
key = wl[:-3] if wl.endswith(".md") else wl
|
|
if key in titles:
|
|
inbound.add(titles[key])
|
|
findings = []
|
|
for doc in docs:
|
|
rel = doc.rel.as_posix()
|
|
if doc.reserved or rel in inbound:
|
|
continue
|
|
findings.append(f"{rel}: no inbound links from anywhere in the tree")
|
|
return findings
|
|
|
|
|
|
def check_graph(root, docs):
|
|
findings = []
|
|
edges_path = root / "graph" / "edges.json"
|
|
if not edges_path.is_file():
|
|
return findings
|
|
try:
|
|
data = json.loads(edges_path.read_text(encoding="utf-8"))
|
|
except json.JSONDecodeError as exc:
|
|
return [f"graph/edges.json: invalid JSON — {exc}"]
|
|
edges = data.get("edges", data) if isinstance(data, dict) else data
|
|
if not isinstance(edges, list):
|
|
return ["graph/edges.json: `edges` is not a list"]
|
|
known = {doc.rel.as_posix() for doc in docs}
|
|
known |= {doc.rel.with_suffix("").as_posix() for doc in docs}
|
|
for i, edge in enumerate(edges):
|
|
if not isinstance(edge, dict):
|
|
findings.append(f"graph/edges.json: edge {i} is not an object")
|
|
continue
|
|
for side in ("from", "to"):
|
|
value = str(edge.get(side, "")).strip()
|
|
if not value:
|
|
findings.append(f"graph/edges.json: edge {i} has no `{side}`")
|
|
continue
|
|
# Edge paths are project-root-absolute per Rule C
|
|
# (`/wiki/entities/foo.md`), while `known` holds paths relative to
|
|
# wiki/. Strip the prefix the same way the decision-record check
|
|
# above already does, or every conformant edge reads as broken.
|
|
stripped = value[len("/wiki/"):] if value.startswith("/wiki/") else value.lstrip("/")
|
|
candidates = {
|
|
value,
|
|
value.removesuffix(".md"),
|
|
stripped,
|
|
stripped.removesuffix(".md"),
|
|
f"entities/{value}",
|
|
f"entities/{value}".removesuffix(".md"),
|
|
}
|
|
if not candidates & known:
|
|
findings.append(f"graph/edges.json: edge {i} `{side}: {value}` does not resolve to a page")
|
|
return findings
|
|
|
|
|
|
def check_index_and_logs(root, docs):
|
|
findings = []
|
|
by_dir = {}
|
|
for doc in docs:
|
|
by_dir.setdefault(doc.rel.parent, []).append(doc)
|
|
|
|
for directory, entries in sorted(by_dir.items()):
|
|
pages = [d for d in entries if not d.reserved]
|
|
index = next((d for d in entries if d.path.name == "index.md"), None)
|
|
if not pages:
|
|
continue
|
|
if index is None:
|
|
findings.append(f"{(directory / 'index.md').as_posix()}: missing — directory holds {len(pages)} page(s)")
|
|
continue
|
|
listed = {Path(t).name for t in index.links()} | {
|
|
(w if w.endswith(".md") else w + ".md").split("/")[-1] for w in index.wikilinks()
|
|
}
|
|
for page in sorted(pages, key=lambda d: d.rel.as_posix()):
|
|
if page.path.name not in listed:
|
|
findings.append(f"{index.rel.as_posix()}: does not list {page.rel.as_posix()}")
|
|
|
|
# A change must have exactly one home log (Recursive Index & Log Convention).
|
|
logs = [d for d in docs if d.path.name == "log.md"]
|
|
root_log = next((d for d in logs if d.rel == Path("log.md")), None)
|
|
if root_log is not None and len(logs) > 1:
|
|
root_entries = log_entry_keys(root_log.body)
|
|
for sub in logs:
|
|
if sub is root_log:
|
|
continue
|
|
for key in sorted(log_entry_keys(sub.body) & root_entries):
|
|
findings.append(
|
|
f"log.md and {sub.rel.as_posix()}: same change recorded in both — {key}"
|
|
)
|
|
return findings
|
|
|
|
|
|
LOG_HEADER_RE = re.compile(r"^##\s*\[?(\d{4}-\d{2}-\d{2}(?:[ T]\d{2}:\d{2})?)\]?")
|
|
|
|
|
|
def log_entry_keys(body):
|
|
"""Timestamp + affected-files pairs, used to spot a change logged twice."""
|
|
keys, stamp = set(), None
|
|
for line in body.split("\n"):
|
|
m = LOG_HEADER_RE.match(line.strip())
|
|
if m:
|
|
stamp = m.group(1)
|
|
continue
|
|
f = re.match(r"^-\s*\*\*File Affected:?\*\*:?\s*(.+)$", line.strip())
|
|
if f and stamp:
|
|
keys.add(f"{stamp} {f.group(1).strip()}")
|
|
return keys
|
|
|
|
|
|
def check_sources(today):
|
|
"""source.yaml validity plus overdue-index reporting for connector libs."""
|
|
findings, connectors = [], []
|
|
if not LIBS.is_dir():
|
|
return findings, connectors
|
|
for lib in sorted(p for p in LIBS.iterdir() if p.is_dir()):
|
|
source = lib / "source.yaml"
|
|
if not source.is_file():
|
|
continue # git-copy lib — never touched by lint
|
|
connectors.append(lib)
|
|
cfg = parse_simple_yaml(source.read_text(encoding="utf-8"))
|
|
name = lib.name
|
|
for key in ("connector", "location"):
|
|
if not str(cfg.get(key, "")).strip():
|
|
findings.append(f"libs/{name}/source.yaml: `{key}` is missing or empty")
|
|
index_block = cfg.get("index")
|
|
if isinstance(index_block, dict):
|
|
for key in ("store", "location"):
|
|
if not str(index_block.get(key, "")).strip():
|
|
findings.append(f"libs/{name}/source.yaml: `index.{key}` is missing or empty")
|
|
refresh = cfg.get("refresh_interval_days")
|
|
interval = DEFAULT_REFRESH_DAYS
|
|
if refresh is not None and not isinstance(refresh, dict):
|
|
parsed = as_int(refresh)
|
|
if parsed is None or parsed <= 0:
|
|
findings.append(f"libs/{name}/source.yaml: `refresh_interval_days` is not a positive integer: {refresh!r}")
|
|
else:
|
|
interval = parsed
|
|
|
|
generated = {"index.md", "log.md", "entities", "graph", "source.yaml", "source.local.yaml"}
|
|
stray = [
|
|
p.name
|
|
for p in sorted(lib.iterdir())
|
|
if p.name not in generated and not p.name.startswith(".")
|
|
]
|
|
if stray:
|
|
findings.append(
|
|
f"libs/{name}/: has both a source.yaml and non-index content ({', '.join(stray)}) — "
|
|
"ambiguous configuration for the user to resolve"
|
|
)
|
|
|
|
newest = None
|
|
for md in lib.rglob("*.md"):
|
|
fm, _, err = parse_frontmatter(md.read_text(encoding="utf-8", errors="replace"))
|
|
if err:
|
|
continue
|
|
updated = as_date(fm.get("last_updated"))
|
|
if updated and (newest is None or updated > newest):
|
|
newest = updated
|
|
if newest is None:
|
|
findings.append(f"libs/{name}/: no generated index yet — run 'index external sources'")
|
|
else:
|
|
age = (today - newest).days
|
|
if age > interval:
|
|
findings.append(
|
|
f"libs/{name}/: index is {age}d old against a {interval}d refresh interval "
|
|
f"({age - interval}d overdue) — suggest 'index external sources'"
|
|
)
|
|
return findings, connectors
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# driver
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def sections(body):
|
|
"""Split a page body into {heading: text} for `##`-level headings."""
|
|
out, current, buf = {}, None, []
|
|
for line in body.split("\n"):
|
|
match = re.match(r"^##\s+(.+?)\s*$", line)
|
|
if match:
|
|
if current is not None:
|
|
out[current] = "\n".join(buf)
|
|
current, buf = match.group(1).strip(), []
|
|
elif current is not None:
|
|
buf.append(line)
|
|
if current is not None:
|
|
out[current] = "\n".join(buf)
|
|
return out
|
|
|
|
|
|
def digest(path):
|
|
"""First 8 hex chars of the file's sha256, or None if unreadable."""
|
|
try:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()[:8]
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
def source_bullets(body):
|
|
"""[(cited path, kind, value)] from a page's `## Sources` section."""
|
|
section = sections(body).get("Sources")
|
|
if section is None:
|
|
return None
|
|
out = []
|
|
for cited, rest in SOURCE_BULLET_RE.findall(section):
|
|
fp = FINGERPRINT_RE.search(rest)
|
|
out.append((cited.strip(), fp.group(1) if fp else None, fp.group(2) if fp else None))
|
|
return out
|
|
|
|
|
|
def check_evidence(docs, root):
|
|
"""Check 12 — `## Sources` fingerprints against the files they cite."""
|
|
findings = []
|
|
for doc in docs:
|
|
if doc.reserved or doc.fm_error:
|
|
continue
|
|
bullets = source_bullets(doc.body)
|
|
if bullets is None:
|
|
continue
|
|
if not bullets:
|
|
findings.append(f"{doc.rel}: `## Sources` section is empty")
|
|
continue
|
|
for cited, kind, value in bullets:
|
|
if re.match(r"^[a-z][a-z0-9+.-]*://", cited):
|
|
continue # a URL: nothing local to hash
|
|
target = (REPO_ROOT / cited.lstrip("/")).resolve()
|
|
if not target.is_file():
|
|
target = (root / cited.lstrip("/")).resolve()
|
|
if not target.is_file():
|
|
findings.append(f"{doc.rel}: cited source `{cited}` does not exist")
|
|
continue
|
|
if kind is None:
|
|
findings.append(f"{doc.rel}: cited source `{cited}` has no fingerprint recorded")
|
|
continue
|
|
if kind != "sha256":
|
|
continue # etag/mtime come from a connector; nothing local to recompute
|
|
actual = digest(target)
|
|
if actual is None:
|
|
findings.append(f"{doc.rel}: cited source `{cited}` could not be read")
|
|
elif not value.lower().startswith(actual):
|
|
findings.append(
|
|
f"{doc.rel}: source `{cited}` CHANGED since this page was built "
|
|
f"(recorded sha256:{value}, now sha256:{actual})"
|
|
)
|
|
return findings
|
|
|
|
|
|
def check_crux(docs, root):
|
|
"""Check 13 — `## Crux` quotes are verbatim, attributed, and non-empty."""
|
|
findings = []
|
|
for doc in docs:
|
|
if doc.reserved or doc.fm_error:
|
|
continue
|
|
secs = sections(doc.body)
|
|
crux = secs.get("Crux")
|
|
if crux is None:
|
|
continue
|
|
quotes = [line[1:].strip() for line in crux.split("\n") if line.startswith(">")]
|
|
quoted = " ".join(q for q in quotes if q)
|
|
if not quoted:
|
|
findings.append(f"{doc.rel}: `## Crux` has no quoted lines (evidence sections must quote, not paraphrase)")
|
|
continue
|
|
if "Sources" not in secs:
|
|
findings.append(f"{doc.rel}: `## Crux` present but the page has no `## Sources` to attribute it to")
|
|
cited = ATTRIBUTION_RE.findall(crux)
|
|
if not cited:
|
|
findings.append(f"{doc.rel}: `## Crux` quote is not attributed to a source")
|
|
continue
|
|
for ref in cited:
|
|
ref = ref.strip().lstrip("/")
|
|
target = (REPO_ROOT / ref).resolve()
|
|
if not target.is_file():
|
|
target = (root / ref).resolve()
|
|
if not target.is_file() or target.suffix.lower() not in QUOTABLE_SUFFIXES:
|
|
continue
|
|
try:
|
|
haystack = " ".join(target.read_text(encoding="utf-8", errors="replace").split())
|
|
except OSError:
|
|
continue
|
|
for quote in quotes:
|
|
if not quote or len(quote) < 24:
|
|
continue # too short to match meaningfully
|
|
needle = " ".join(quote.split())
|
|
if needle not in haystack:
|
|
findings.append(
|
|
f"{doc.rel}: `## Crux` quote not found verbatim in `{ref}` — "
|
|
f"{needle[:60]!r}..."
|
|
)
|
|
return findings
|
|
|
|
|
|
CHECKS = [
|
|
("1 conformance", "conformance"),
|
|
("2 freshness", "freshness"),
|
|
("3 confidence", "confidence"),
|
|
("4 retention", "retention"),
|
|
("5 decisions", "decisions"),
|
|
("6 orphans", "orphans"),
|
|
("7 graph", "graph"),
|
|
("8 index/log", "index_log"),
|
|
("12 source fingerprints", "evidence"),
|
|
("13 crux", "crux"),
|
|
]
|
|
|
|
# The subset worth running at session start (Rule E): cheap, and each finding
|
|
# means something has actually changed rather than merely aged on a calendar.
|
|
QUICK_CHECKS = [("conformance", "malformed"), ("freshness", "past freshness window"),
|
|
("evidence", "source drift"), ("crux", "crux quotes diverged")]
|
|
|
|
|
|
def run_tree(root, today, label):
|
|
docs = read_tree(root)
|
|
return {
|
|
"label": label,
|
|
"documents": len(docs),
|
|
"conformance": check_conformance(docs),
|
|
"freshness": check_freshness(docs, today),
|
|
"confidence": check_confidence(docs),
|
|
"retention": check_retention(docs, today),
|
|
"decisions": check_decisions(docs, root, today),
|
|
"orphans": check_orphans(docs),
|
|
"graph": check_graph(root, docs),
|
|
"index_log": check_index_and_logs(root, docs),
|
|
"evidence": check_evidence(docs, root),
|
|
"crux": check_crux(docs, root),
|
|
"hubs": in_degree(root, docs),
|
|
}
|
|
|
|
|
|
def in_degree(root, docs):
|
|
"""Top pages by inbound edge count — not a finding, a retrieval signal.
|
|
|
|
`ckb-retrieve` fuses in-degree as one ranked list among several; surfacing
|
|
it here means the same number comes from one place rather than being
|
|
recomputed by eye at query time.
|
|
"""
|
|
edges_path = root / "graph" / "edges.json"
|
|
if not edges_path.is_file():
|
|
return []
|
|
try:
|
|
data = json.loads(edges_path.read_text(encoding="utf-8"))
|
|
except json.JSONDecodeError:
|
|
return []
|
|
edges = data.get("edges", data) if isinstance(data, dict) else data
|
|
if not isinstance(edges, list):
|
|
return []
|
|
counts = {}
|
|
for edge in edges:
|
|
if not isinstance(edge, dict):
|
|
continue
|
|
target = str(edge.get("to", "")).strip()
|
|
if target:
|
|
counts[target] = counts.get(target, 0) + 1
|
|
return sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))[:10]
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Read-only mechanical checks for ckb-lint.")
|
|
parser.add_argument("--scope", choices=("wiki", "libs", "all"), default="all")
|
|
parser.add_argument("--today", help="override today's date (YYYY-MM-DD) for reproducible runs")
|
|
parser.add_argument("--json", action="store_true", help="emit machine-readable JSON instead of text")
|
|
parser.add_argument("--quick", action="store_true",
|
|
help="one-line session-start summary (Rule E): conformance, freshness, and source drift only")
|
|
args = parser.parse_args()
|
|
|
|
today = as_date(args.today) if args.today else date.today()
|
|
if today is None:
|
|
print(f"error: --today is not a YYYY-MM-DD date: {args.today!r}", file=sys.stderr)
|
|
return 2
|
|
|
|
trees, source_findings = [], []
|
|
if args.scope in ("wiki", "all"):
|
|
if not WIKI.is_dir():
|
|
print(f"error: no wiki/ directory under {REPO_ROOT} — run from the repository root", file=sys.stderr)
|
|
return 2
|
|
trees.append(run_tree(WIKI, today, "wiki/"))
|
|
if args.scope in ("libs", "all"):
|
|
source_findings, connectors = check_sources(today)
|
|
for lib in connectors:
|
|
access = "read-only"
|
|
local = lib / "source.local.yaml"
|
|
if local.is_file():
|
|
cfg = parse_simple_yaml(local.read_text(encoding="utf-8"))
|
|
if str(cfg.get("access", "")).strip().lower() == "write":
|
|
access = "write"
|
|
tree = run_tree(lib, today, f"libs/{lib.name}/ (access: {access})")
|
|
tree["access"] = access
|
|
trees.append(tree)
|
|
|
|
total = sum(len(t[key]) for t in trees for _, key in CHECKS) + len(source_findings)
|
|
|
|
if args.quick:
|
|
counts = {key: sum(len(t[key]) for t in trees) for key, _ in QUICK_CHECKS}
|
|
flagged = [f"{counts[key]} {label}" for key, label in QUICK_CHECKS if counts[key]]
|
|
if not flagged:
|
|
print("ckb check: clean")
|
|
return 0
|
|
print("ckb check: " + ", ".join(flagged) + " — run \"lint\" for detail")
|
|
return 1
|
|
|
|
if args.json:
|
|
print(json.dumps({"today": today.isoformat(), "trees": trees, "sources": source_findings, "findings": total}, indent=2))
|
|
return 1 if total else 0
|
|
|
|
for tree in trees:
|
|
print(f"=== {tree['label']} — {tree['documents']} markdown file(s)")
|
|
for title, key in CHECKS:
|
|
items = tree[key]
|
|
if not items:
|
|
continue
|
|
print(f" [{title}] {len(items)} finding(s)")
|
|
for item in items:
|
|
print(f" - {item}")
|
|
if not any(tree[key] for _, key in CHECKS):
|
|
print(" clean")
|
|
if tree["hubs"]:
|
|
top = ", ".join(f"{name} ({n})" for name, n in tree["hubs"][:5])
|
|
print(f" [in-degree] most-referenced pages: {top}")
|
|
if args.scope in ("libs", "all"):
|
|
print("=== external sources (libs/*/source.yaml)")
|
|
if source_findings:
|
|
for item in source_findings:
|
|
print(f" - {item}")
|
|
else:
|
|
print(" clean")
|
|
print(f"total findings: {total}")
|
|
print("checks NOT covered here (model's job): whether two pages genuinely supersede "
|
|
"each other, 9 error-book, auto-fix vs. report")
|
|
print("note: a changed source fingerprint (check 12) or a diverged crux quote (check 13) "
|
|
"means the evidence moved, not that a page merely aged — read those first")
|
|
return 1 if total else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|