Coverage for src/lilbee/wiki/batch.py: 100%
121 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 11:46 +0000
1"""Per-source batched-generation helpers and legacy concept-page archival.
3The batched build (one LLM call per source that emits sections for
4every pre-extracted entity plus 3-5 LLM-curated concepts) lives here:
5section-finalization, label matching, source hashing, and the page
6splitter that turns the model's response into per-section bodies.
7Also owns the one-time migration that archives legacy concept pages
8(written before per-source batched generation) and unwraps stale
9``[[archived-slug]]`` links.
10"""
12from __future__ import annotations
14import hashlib
15import logging
16import re
17from collections.abc import Callable, Sequence
18from datetime import UTC, datetime
19from pathlib import Path
21from lilbee.core.config import Config
22from lilbee.core.text import make_slug
23from lilbee.data.ingest import file_hash
24from lilbee.data.store import CitationRecord, SearchChunk, Store
25from lilbee.wiki.citations import (
26 ParsedCitation,
27 footnote_marker_keys,
28 render_citation_block,
29 scrub_unverified_markers,
30 strip_citation_block,
31 verify_citations,
32 wiki_sourced_count,
33)
34from lilbee.wiki.entity_extractor import EntityKind
35from lilbee.wiki.page import assemble_content, build_frontmatter
36from lilbee.wiki.persistence import (
37 delete_pending_marker_if_present,
38 divert_concept_collision,
39 persist_and_finalize,
40)
41from lilbee.wiki.quality import check_faithfulness
42from lilbee.wiki.shared import (
43 WIKI_CONTENT_SUBDIRS,
44 PageTarget,
45 WikiSubdir,
46 atomic_write_text,
47)
48from lilbee.wiki.stats import BuildStats
50log = logging.getLogger(__name__)
52# Sentinel file for the one-time legacy-concepts archival. Lives under
53# data_dir (NOT inside wiki/) so Obsidian sync and wiki tree-walkers
54# never surface it. The on-disk filename is preserved across renames so
55# upgrading installs do not re-run the migration.
56_LEGACY_CONCEPTS_MIGRATED_SENTINEL = ".phase-d-migrated"
58# Legacy wiki concepts that we move to archive/ as part of the one-time
59# migration. Matches wiki/<WikiSubdir.CONCEPTS>/*.md recursively.
60_ARCHIVE_CONCEPTS_SUBPATH = Path(WikiSubdir.ARCHIVE) / WikiSubdir.CONCEPTS
63def hash_existing_sources(source_names: list[str]) -> dict[str, str]:
64 """Hash each source file that still exists on disk (used for citation staleness)."""
65 from lilbee.data.ingest.discovery import resolve_source_path
67 out: dict[str, str] = {}
68 for name in source_names:
69 source_path = resolve_source_path(name)
70 if source_path.exists():
71 out[name] = file_hash(source_path)
72 return out
75def _first_label_where(
76 candidates: list[tuple[list[str], EntityKind]],
77 predicate: Callable[[str], bool],
78) -> tuple[EntityKind, str] | None:
79 for labels, kind in candidates:
80 for label in labels:
81 if predicate(label.lower()):
82 return (kind, label)
83 return None
86def match_label(
87 lowered_name: str,
88 candidates: Sequence[tuple[set[str], EntityKind]],
89) -> tuple[EntityKind, str] | None:
90 """Case-insensitive match of *lowered_name* against ordered *candidates*.
92 Each candidate is an ``(expected labels, kind)`` pair. Returns
93 ``(kind, original_label)`` on hit, ``None`` otherwise. Every candidate set
94 is tried for an exact match before any is tried for a substring match, so a
95 header naming one set's label exactly is not taken by another set's label
96 that merely contains it. Substring overlap in either direction accommodates
97 the LLM adding qualifiers ("Brake System (hydraulic)" vs "brake system").
98 Labels are ordered by length then alphabetically, so overlapping labels
99 ("Ford" and "Henry Ford") bind the same way on every run.
100 """
101 ordered = [
102 (sorted(expected, key=lambda label: (-len(label), label)), kind)
103 for expected, kind in candidates
104 ]
105 return _first_label_where(ordered, lambda low: low == lowered_name) or _first_label_where(
106 ordered, lambda low: bool(low) and (low in lowered_name or lowered_name in low)
107 )
110def short_source_hash(source: str) -> str:
111 """8-char sha256 digest of *source* (stable collision-marker suffix)."""
112 return hashlib.sha256(source.encode("utf-8")).hexdigest()[:8]
115def archive_legacy_concept_pages(
116 wiki_root: Path, data_dir: Path, store: Store, config: Config
117) -> None:
118 """One-time migration: archive legacy concept pages.
120 Runs idempotently, gated by ``{data_dir}/.phase-d-migrated``:
122 1. Delete each ``wiki/concepts/*.md`` page's chunk and citation
123 rows, then move the file to ``wiki/archive/concepts/``
124 preserving relative subpaths. Store cleanup comes first so an
125 interrupted migration leaves the page on disk rather than rows
126 serving a page nothing scans. Older concept pages stay readable
127 but drop out of retrieval and the active browse surface.
128 2. Unwrap stale ``[[archived-slug]]`` references across the
129 remaining pages so a reader clicking a link does not hit a
130 404. Archived slugs become plain text.
131 3. Write the sentinel so future builds skip this path.
133 Freshly LLM-curated concept pages written AFTER the sentinel exists
134 are never touched.
135 """
136 sentinel = data_dir / _LEGACY_CONCEPTS_MIGRATED_SENTINEL
137 if sentinel.exists():
138 return
139 concepts_dir = wiki_root / WikiSubdir.CONCEPTS
140 archive_dir = wiki_root / _ARCHIVE_CONCEPTS_SUBPATH
141 archived_slugs: list[str] = []
142 if concepts_dir.is_dir():
143 for src in sorted(concepts_dir.rglob("*.md")):
144 rel = src.relative_to(concepts_dir)
145 dest = archive_dir / rel
146 dest.parent.mkdir(parents=True, exist_ok=True)
147 slug = str(rel.with_suffix("")).replace("\\", "/")
148 wiki_source = f"{config.wiki_dir}/{WikiSubdir.CONCEPTS}/{slug}.md"
149 store.delete_by_source(wiki_source)
150 if not store.delete_citations_for_wiki(wiki_source):
151 # Sentinel unwritten: the next build retries the migration. Pages
152 # already moved this pass still need their inbound links unwrapped,
153 # since the retry only sees what is left in concepts/ and would
154 # never revisit them, leaving those links pointing at a 404.
155 log.warning("Citation delete failed for %s; migration will retry", wiki_source)
156 _unwrap_archived_links(wiki_root, archived_slugs)
157 return
158 src.replace(dest)
159 archived_slugs.append(slug)
161 if archived_slugs:
162 _unwrap_archived_links(wiki_root, archived_slugs)
164 data_dir.mkdir(parents=True, exist_ok=True)
165 sentinel.write_text(datetime.now(UTC).isoformat(), encoding="utf-8")
166 if archived_slugs:
167 log.info(
168 "Legacy-concepts migration: archived %d concept pages, sentinel written at %s",
169 len(archived_slugs),
170 sentinel,
171 )
174def _unwrap_archived_links(wiki_root: Path, archived_slugs: list[str]) -> None:
175 """Rewrite ``[[slug]]`` → ``slug`` (plain text) across remaining wiki pages.
177 The existing ``rewrite_links_across_wiki`` path is the wrong
178 tool here: it compiles an *additive* surface map, not a
179 removal pass. Walk the active wiki content subdirs once per
180 archived slug is acceptable because the archive count is
181 bounded (concepts that existed pre-migration). Pages whose body
182 did not change are not rewritten.
183 """
184 if not archived_slugs:
185 return
186 patterns = [(re.compile(r"\[\[" + re.escape(slug) + r"\]\]"), slug) for slug in archived_slugs]
187 for subdir in WIKI_CONTENT_SUBDIRS:
188 subdir_path = wiki_root / subdir
189 if not subdir_path.is_dir():
190 continue
191 for md_path in subdir_path.rglob("*.md"):
192 original = md_path.read_text(encoding="utf-8")
193 rewritten = original
194 for pattern, replacement in patterns:
195 rewritten = pattern.sub(replacement, rewritten)
196 if rewritten != original:
197 atomic_write_text(md_path, rewritten)
200def finalize_section(
201 *,
202 header_label: str,
203 kind: EntityKind,
204 body: str,
205 chunks: list[SearchChunk],
206 citation_resolver: Callable[[list[ParsedCitation]], list[CitationRecord]],
207 source_names: list[str],
208 store: Store,
209 config: Config,
210 source: str,
211 written_concept_slugs: dict[str, str],
212 drafts_dir: Path,
213 shared_parsed_citations: list[ParsedCitation],
214 scoring_chunks_by_label: dict[str, list[SearchChunk]],
215 stats: BuildStats | None = None,
216) -> Path | None:
217 """Citation-check, faithfulness-check, write one batched section.
219 Shared by entity and concept sections from the per-source batched
220 call. Returns the written page path, or ``None`` if the section
221 failed any gate (no citations, empty body, slug collision marker
222 handled via side channel). ``shared_parsed_citations`` is the
223 definition list parsed once over the whole response: every
224 section replays it so pages other than the last one still have
225 their footnotes resolved.
227 ``scoring_chunks_by_label`` maps an entity label to the chunks its
228 extraction refs named. Faithfulness scores against those rather
229 than the whole-source mean, so a section about one entity in a
230 multi-topic source is not compared to the document-wide centroid.
231 A label with no entry (concepts, or refs that fell outside the
232 budgeted chunks) scores against the full pool.
234 Citation counts and the section's outcome are recorded on *stats*.
235 """
236 stats = BuildStats.ensure(stats)
237 slug = make_slug(header_label)
238 if not slug:
239 log.info("Empty slug for batched section %r; skipping", header_label)
240 return None
242 # Only replay citation keys the section's prose references. Keys are read
243 # from the citation-stripped body: the response's trailing block lands in
244 # the last section, whose definitions would otherwise make it claim every
245 # citation in the response.
246 section_keys = footnote_marker_keys(strip_citation_block(body))
247 relevant = [c for c in shared_parsed_citations if c.citation_key in section_keys]
248 resolved = citation_resolver(relevant)
249 verified = verify_citations(resolved, chunks, header_label, config)
250 dropped = len(relevant) - wiki_sourced_count(resolved, config) - len(verified)
251 if not verified:
252 stats.record_citations(0, dropped)
253 log.info("No valid citations for batched section %s, skipping", header_label)
254 return None
256 score = check_faithfulness(
257 scoring_chunks_by_label.get(header_label, chunks), body, header_label, config
258 )
259 threshold = config.wiki_embedding_faithfulness_threshold
260 page_type = WikiSubdir.CONCEPTS if kind is EntityKind.CONCEPT else WikiSubdir.ENTITIES
261 subdir = page_type if score >= threshold else WikiSubdir.DRAFTS
262 if subdir == WikiSubdir.DRAFTS:
263 log.info(
264 "Batched section %s scored %.2f (< %.2f), sending to drafts",
265 header_label,
266 score,
267 threshold,
268 )
270 clean_body = scrub_unverified_markers(strip_citation_block(body), verified)
271 frontmatter = build_frontmatter(config, source_names, score, chunks=chunks)
272 citation_block = render_citation_block(verified)
273 full_content = assemble_content(frontmatter, clean_body, citation_block)
275 # Recorded before the collision return: the section's footnotes were parsed and
276 # verified either way, and the verify rate counts every outcome that got that far.
277 stats.record_citations(len(verified), dropped)
279 # Concept collision: the second source proposing a slug loses and writes to a
280 # drafts collision marker; the winning source's page stays untouched. This
281 # applies whether the section publishes or is routed to drafts -- a below-
282 # threshold concept still claims the slug, and two such drafts would otherwise
283 # overwrite each other at drafts/<slug>.md.
284 if kind is EntityKind.CONCEPT:
285 first_source = written_concept_slugs.get(slug)
286 if first_source is not None and first_source != source:
287 stats.record_pending_marker()
288 divert_concept_collision(
289 slug=slug,
290 source=source,
291 first_source=first_source,
292 content=full_content,
293 drafts_dir=drafts_dir,
294 origin_subdir=page_type,
295 )
296 return None
297 written_concept_slugs.setdefault(slug, source)
299 # Successful regen of a previously-PENDING slug: remove the old
300 # marker so the drafts surface no longer lists it.
301 delete_pending_marker_if_present(drafts_dir, slug)
303 wiki_root = config.data_root / config.wiki_dir
304 target = PageTarget(
305 wiki_root=wiki_root,
306 subdir=subdir,
307 slug=slug,
308 wiki_source=f"{config.wiki_dir}/{subdir}/{slug}.md",
309 page_type=page_type,
310 label=header_label,
311 )
312 page_path = persist_and_finalize(
313 full_content, target, verified, source_names, store, config, stats=stats
314 )
315 log.info(
316 "Generated batched page for %s -> %s (score=%.2f, citations=%d)",
317 header_label,
318 target.subdir,
319 score,
320 len(verified),
321 )
322 return page_path