Coverage for src/lilbee/wiki/lint.py: 100%

1"""Lint wiki pages for citation staleness, missing sources, and unmarked claims. 

2 

3Two modes: 

4- lightweight: runs automatically after sync, checks only pages whose sources changed 

5- full: manual ``lilbee wiki lint``, checks all wiki pages 

6""" 

7 

8from __future__ import annotations 

9 

10import logging 

11import threading 

12from dataclasses import dataclass, field 

13from enum import Enum 

14from pathlib import Path 

15 

16from lilbee.core.config import Config, cfg 

17from lilbee.core.security import PathTraversalError, validate_path_within 

18from lilbee.data.ingest import file_hash 

19from lilbee.data.store import CitationRecord, Store 

20from lilbee.wiki.citations import ( 

21 CitationStatus, 

22 find_unmarked_claims, 

23 verify_citation, 

24) 

25from lilbee.wiki.grammar import WIKI_LINK_RE 

26from lilbee.wiki.index import append_wiki_log 

27from lilbee.wiki.shared import ( 

28 WIKI_BUILD_LOCK, 

29 WIKI_CONTENT_SUBDIRS, 

30 WikiLogAction, 

31 WikiSubdir, 

32 parse_frontmatter, 

33) 

34 

35_ORPHAN_CANDIDATE_SUBDIRS: tuple[str, ...] = (WikiSubdir.CONCEPTS, WikiSubdir.ENTITIES) 

36 

37# Subdirs whose links don't count as "published" backlinks for orphan detection: 

38# a [[slug]] living only in a draft or an archived page must not exempt a live 

39# concept/entity page from the orphan flag. 

40_UNPUBLISHED_SUBDIRS: tuple[str, ...] = (WikiSubdir.DRAFTS, WikiSubdir.ARCHIVE) 

41 

42log = logging.getLogger(__name__) 

43 

44 

45class IssueSeverity(Enum): 

46 """Severity level for lint issues.""" 

47 

48 WARNING = "warning" 

49 ERROR = "error" 

50 

51 

52class IssueType(Enum): 

53 """Classification of lint findings, used by prune to filter programmatically.""" 

54 

55 PATH_TRAVERSAL = "path_traversal" 

56 SOURCE_MISSING = "source_missing" 

57 STALE_HASH = "stale_hash" 

58 EXCERPT_MISSING = "excerpt_missing" 

59 MODEL_CHANGED = "model_changed" 

60 UNMARKED_CLAIM = "unmarked_claim" 

61 ORPHAN = "orphan" 

62 

63 

64@dataclass(frozen=True) 

65class LintIssue: 

66 """A single lint finding on a wiki page.""" 

67 

68 wiki_source: str 

69 severity: IssueSeverity 

70 message: str 

71 issue_type: IssueType | None = None 

72 

73 def to_dict(self) -> dict[str, str]: 

74 """Serialize to a plain dict suitable for JSON output.""" 

75 return { 

76 "wiki_source": self.wiki_source, 

77 "severity": self.severity.value, 

78 "message": self.message, 

79 "issue_type": self.issue_type.value if self.issue_type else "", 

80 } 

81 

82 

83@dataclass 

84class LintReport: 

85 """Aggregated results from linting one or more wiki pages.""" 

86 

87 issues: list[LintIssue] = field(default_factory=list) 

88 

89 @property 

90 def error_count(self) -> int: 

91 return sum(1 for i in self.issues if i.severity == IssueSeverity.ERROR) 

92 

93 @property 

94 def warning_count(self) -> int: 

95 return sum(1 for i in self.issues if i.severity == IssueSeverity.WARNING) 

96 

97 

98def _lint_citation( 

99 rec: CitationRecord, 

100 store: Store, 

101) -> LintIssue | None: 

102 """Check a single citation record against the filesystem and the chunk store. 

103 Returns a LintIssue if the citation is stale or broken, None if valid. 

104 """ 

105 from lilbee.data.ingest.discovery import resolve_source_path_checked 

106 

107 wiki_source = rec["wiki_source"] 

108 # A registered root legitimately lives outside documents_dir, so containment 

109 # is checked against every allowed root: a key that climbs out of all of them 

110 # (a crafted ``../`` citation) is rejected before any file is read. 

111 source_path = resolve_source_path_checked(rec["source_filename"]) 

112 if source_path is None: 

113 return LintIssue( 

114 wiki_source=wiki_source, 

115 severity=IssueSeverity.ERROR, 

116 message=f"Source path escapes its root: {rec['source_filename']}", 

117 issue_type=IssueType.PATH_TRAVERSAL, 

118 ) 

119 

120 if not source_path.exists(): 

121 return LintIssue( 

122 wiki_source=wiki_source, 

123 severity=IssueSeverity.ERROR, 

124 message=f"Source deleted: {rec['source_filename']}", 

125 issue_type=IssueType.SOURCE_MISSING, 

126 ) 

127 

128 current_hash = file_hash(source_path) 

129 if current_hash != rec["source_hash"]: 

130 return LintIssue( 

131 wiki_source=wiki_source, 

132 severity=IssueSeverity.WARNING, 

133 message=f"Stale hash for {rec['source_filename']} (citation: {rec['citation_key']})", 

134 issue_type=IssueType.STALE_HASH, 

135 ) 

136 

137 return _lint_excerpt(rec, store) 

138 

139 

140def _lint_excerpt(rec: CitationRecord, store: Store) -> LintIssue | None: 

141 """Verify a citation's excerpt against the source's extracted chunks. 

142 

143 Excerpts are quoted from extracted text, not from the raw file: a PDF's 

144 bytes contain none of it. The caller has already established that the file 

145 is present and its hash current, so a source left with no chunks is one 

146 ``wiki_prune_raw`` cleared at publish time: the citation was verified at 

147 build and is not re-flagged. 

148 """ 

149 chunk_texts = [c.chunk for c in store.get_chunks_by_source(rec["source_filename"])] 

150 status = verify_citation(rec, chunk_texts) 

151 if status is CitationStatus.UNVERIFIABLE: 

152 log.debug( 

153 "No extracted text for %s; %s stands as verified at build time", 

154 rec["source_filename"], 

155 rec["citation_key"], 

156 ) 

157 return None 

158 if status is CitationStatus.EXCERPT_MISSING: 

159 return LintIssue( 

160 wiki_source=rec["wiki_source"], 

161 severity=IssueSeverity.WARNING, 

162 message=f"Excerpt not found in source for {rec['citation_key']}", 

163 issue_type=IssueType.EXCERPT_MISSING, 

164 ) 

165 return None 

166 

167 

168def _lint_model_changed(wiki_source: str, text: str, config: Config) -> LintIssue | None: 

169 """Flag pages whose generated_by model differs from the current chat model.""" 

170 generated_by = parse_frontmatter(text).get("generated_by", "") 

171 if not generated_by: 

172 return None 

173 if generated_by != config.chat_model: 

174 return LintIssue( 

175 wiki_source=wiki_source, 

176 severity=IssueSeverity.WARNING, 

177 issue_type=IssueType.MODEL_CHANGED, 

178 message=( 

179 f"model_changed: page generated by {generated_by!r}, " 

180 f"current model is {config.chat_model!r}" 

181 ), 

182 ) 

183 return None 

184 

185 

186def _lint_unmarked(wiki_source: str, text: str) -> list[LintIssue]: 

187 """Find unmarked claims in a wiki page.""" 

188 unmarked = find_unmarked_claims(text) 

189 return [ 

190 LintIssue( 

191 wiki_source=wiki_source, 

192 severity=IssueSeverity.WARNING, 

193 message=f"Unmarked claim: {line[:80]}", 

194 issue_type=IssueType.UNMARKED_CLAIM, 

195 ) 

196 for line in unmarked 

197 ] 

198 

199 

200def lint_wiki_page( 

201 wiki_source: str, 

202 store: Store, 

203 config: Config | None = None, 

204) -> list[LintIssue]: 

205 """Lint a single wiki page: check citations and unmarked claims.""" 

206 if config is None: 

207 config = cfg 

208 issues: list[LintIssue] = [] 

209 

210 citations = store.get_citations_for_wiki(wiki_source) 

211 for rec in citations: 

212 issue = _lint_citation(rec, store) 

213 if issue is not None: 

214 issues.append(issue) 

215 

216 wiki_root = config.data_root / config.wiki_dir 

217 # wiki_source is like "wiki/summaries/doc.md": strip the wiki_dir prefix 

218 relative = str(wiki_source).removeprefix(str(config.wiki_dir) + "/") 

219 wiki_path = wiki_root / relative 

220 # wiki_source reaches here straight from the CLI/MCP, so a traversal source 

221 # ("../../etc/passwd") would otherwise read and disclose an arbitrary file. 

222 try: 

223 validate_path_within(wiki_path, wiki_root) 

224 except PathTraversalError: 

225 return issues 

226 if wiki_path.exists(): 

227 text = wiki_path.read_text(encoding="utf-8", errors="replace") 

228 issues.extend(_lint_unmarked(wiki_source, text)) 

229 model_issue = _lint_model_changed(wiki_source, text, config) 

230 if model_issue is not None: 

231 issues.append(model_issue) 

232 

233 return issues 

234 

235 

236def lint_changed_sources( 

237 changed_sources: list[str], 

238 store: Store, 

239 config: Config | None = None, 

240) -> LintReport: 

241 """Lightweight lint for wiki pages citing changed or removed sources. 

242 

243 Callable from tools that already know the set of changed sources 

244 (e.g. a future `lilbee wiki check <source>` command); the sync 

245 pipeline uses `lilbee.wiki.ingest.incremental_update` instead, which runs full 

246 extraction rather than citation replay. 

247 """ 

248 if config is None: 

249 config = cfg 

250 report = LintReport() 

251 

252 seen_pages: set[str] = set() 

253 for source_name in changed_sources: 

254 citations = store.get_citations_for_source(source_name) 

255 for rec in citations: 

256 wiki_source = rec["wiki_source"] 

257 if wiki_source in seen_pages: 

258 continue 

259 seen_pages.add(wiki_source) 

260 report.issues.extend(lint_wiki_page(wiki_source, store, config)) 

261 

262 if report.issues: 

263 log.info( 

264 "Wiki lint: %d error(s), %d warning(s)", 

265 report.error_count, 

266 report.warning_count, 

267 ) 

268 return report 

269 

270 

271def lint_all( 

272 store: Store, 

273 config: Config | None = None, 

274 *, 

275 record_log: bool = True, 

276 cancel: threading.Event | None = None, 

277) -> LintReport: 

278 """Full lint: check every wiki page in the store. 

279 

280 ``record_log=False`` skips the audit-log append so a read-only status check 

281 can reuse this without mutating ``log.md``. 

282 

283 Each page costs a citation lookup, so a large wiki takes long enough to 

284 want stopping. Setting *cancel* ends the scan at the next page and reports 

285 what it found so far. 

286 """ 

287 if config is None: 

288 config = cfg 

289 report = LintReport() 

290 

291 wiki_root = config.data_root / config.wiki_dir 

292 if not wiki_root.exists(): 

293 return report 

294 

295 for subdir in WIKI_CONTENT_SUBDIRS: 

296 subdir_path = wiki_root / subdir 

297 if not subdir_path.is_dir(): 

298 continue 

299 for md_path in sorted(subdir_path.rglob("*.md")): 

300 if cancel is not None and cancel.is_set(): 

301 log.info("Wiki lint cancelled after %d issues", len(report.issues)) 

302 break 

303 relative = md_path.relative_to(wiki_root) 

304 wiki_source = f"{config.wiki_dir}/{relative.as_posix()}" 

305 report.issues.extend(lint_wiki_page(wiki_source, store, config)) 

306 

307 report.issues.extend(_lint_orphans(wiki_root, config)) 

308 if record_log: 

309 # log.md is shared with the builders; the append takes the same mutex. 

310 with WIKI_BUILD_LOCK: 

311 append_wiki_log( 

312 WikiLogAction.LINT, 

313 f"{report.error_count} error(s), {report.warning_count} warning(s)", 

314 config, 

315 ) 

316 return report 

317 

318 

319def _lint_orphans(wiki_root: Path, config: Config) -> list[LintIssue]: 

320 """Flag concept/entity pages that no other page links back to. 

321 

322 Single-pass over the wiki tree: we collect every inbound 

323 ``[[slug]]`` reference and the set of orphan candidates in one 

324 ``rglob`` walk, then subtract. The earlier two-pass version 

325 re-walked the tree to compute ``referenced`` and again to check 

326 candidates, which doubles the file-IO at build time. 

327 """ 

328 referenced: set[str] = set() 

329 candidates: list[Path] = [] 

330 candidate_roots = {wiki_root / sub for sub in _ORPHAN_CANDIDATE_SUBDIRS} 

331 unpublished_roots = {wiki_root / sub for sub in _UNPUBLISHED_SUBDIRS} 

332 for md_path in wiki_root.rglob("*.md"): 

333 # Only published pages contribute backlinks; a link from a draft or an 

334 # archived page must not keep a live concept/entity page off the orphan list. 

335 if not any(root in md_path.parents for root in unpublished_roots): 

336 text = md_path.read_text(encoding="utf-8", errors="replace") 

337 for match in WIKI_LINK_RE.finditer(text): 

338 slug = match.group(1).split("|", 1)[0].strip().lower() 

339 if slug: 

340 referenced.add(slug) 

341 if any(root in md_path.parents for root in candidate_roots): 

342 candidates.append(md_path) 

343 

344 issues: list[LintIssue] = [] 

345 for md_path in sorted(candidates): 

346 slug = md_path.stem.lower() 

347 if slug in referenced: 

348 continue 

349 relative = md_path.relative_to(wiki_root) 

350 wiki_source = f"{config.wiki_dir}/{relative.as_posix()}" 

351 issues.append( 

352 LintIssue( 

353 wiki_source=wiki_source, 

354 severity=IssueSeverity.WARNING, 

355 issue_type=IssueType.ORPHAN, 

356 message=f"Orphan: no inbound [[{slug}]] links from any other page", 

357 ) 

358 ) 

359 return issues