feat(F-12): wiki-compile — grounded concept pages over RAG substrate #2

Merged
user2595 merged 7 commits from feat/F-12-wiki-compile into main 2026-06-14 05:58:25 +00:00
6 changed files with 670 additions and 48 deletions

View File

@@ -329,6 +329,19 @@ def wiki_check(
typer.echo(f"⚠ Ungrounded claim(s) in: {page.name}", err=True) typer.echo(f"⚠ Ungrounded claim(s) in: {page.name}", err=True)
found_ungrounded = True found_ungrounded = True
# Check for quarantined pages in wiki/draft/
draft_dir = wiki_dir / "draft"
draft_pages = sorted(draft_dir.glob("*.md")) if draft_dir.exists() else []
if draft_pages:
n = len(draft_pages)
typer.echo(
f"{n} page(s) quarantined in wiki/draft/ (grounding rate below threshold)",
err=True,
)
for dp in draft_pages:
typer.echo(f" - {dp.name}", err=True)
raise typer.Exit(2)
if found_ungrounded: if found_ungrounded:
raise typer.Exit(1) raise typer.Exit(1)
else: else:

View File

@@ -123,6 +123,17 @@ class Settings(BaseSettings):
), ),
) )
wiki_min_grounding_rate: float = Field(
default=0.5,
ge=0.0,
le=1.0,
description=(
"Minimum fraction of claims that must be grounded for a page to be written "
"to wiki/<slug>.md. Pages below this threshold are quarantined to "
"wiki/draft/<slug>.md instead."
),
)
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# F-09 Rich Parsing # F-09 Rich Parsing
# ------------------------------------------------------------------ # ------------------------------------------------------------------

View File

@@ -11,12 +11,14 @@ Compile-state (hash tracking for incremental re-runs) is persisted to
Graceful degradation: Graceful degradation:
- Missing ``formulas`` table (F-09 not present) → no formula embedding, no crash. - Missing ``formulas`` table (F-09 not present) → no formula embedding, no crash.
- Missing ``verify_citations`` (F-10 not present) → local substring grounding check. - Missing ``verify_citations`` (F-10 not present) → local substring grounding check.
- LLM unavailable (httpx.ConnectError or any exception) → empty ConceptPage, no crash.
""" """
from __future__ import annotations from __future__ import annotations
import hashlib import hashlib
import json import json
import logging
import re import re
import textwrap import textwrap
from dataclasses import dataclass, field from dataclasses import dataclass, field
@@ -28,6 +30,8 @@ import yaml
from codex.config import get_settings from codex.config import get_settings
logger = logging.getLogger(__name__)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Dataclasses # Dataclasses
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -73,6 +77,7 @@ class CompileReport:
skipped: list[str] = field(default_factory=list) # slugs skipped (unchanged) skipped: list[str] = field(default_factory=list) # slugs skipped (unchanged)
ungrounded: list[tuple[str, str]] = field(default_factory=list) # (slug, claim_text) ungrounded: list[tuple[str, str]] = field(default_factory=list) # (slug, claim_text)
conflicts: list[tuple[str, str, str]] = field(default_factory=list) # (slug, bibkey1, bibkey2) conflicts: list[tuple[str, str, str]] = field(default_factory=list) # (slug, bibkey1, bibkey2)
quarantined: list[str] = field(default_factory=list) # slugs written to wiki/draft/
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -224,30 +229,75 @@ def _parse_claims(markdown: str) -> list[Claim]:
text = match.group("text").strip() text = match.group("text").strip()
bibkey = match.group("bibkey").strip() bibkey = match.group("bibkey").strip()
locator = (match.group("locator") or "").strip() locator = (match.group("locator") or "").strip()
# Skip URL-shaped bibkeys ([text](https://...)) and multi-word bibkeys
# (real BibKeys never contain spaces or start with "http")
if " " in bibkey or bibkey.startswith("http"):
continue
if text and bibkey: if text and bibkey:
claims.append(Claim(text=text, bibkey=bibkey, locator=locator)) claims.append(Claim(text=text, bibkey=bibkey, locator=locator))
return claims return claims
_STOPWORDS: frozenset[str] = frozenset(
{
"the", "a", "an", "in", "on", "of", "to", "is", "are", "was", "were",
"and", "or", "but", "for", "with", "this", "that", "it", "we", "they",
"be", "as", "at", "by", "from", "has", "have", "not", "which",
}
)
_SENTENCE_SPLIT_RE = re.compile(r"[.!?]")
def _last_sentence(text: str) -> str:
"""Return the last non-empty sentence from *text*.
Citations annotate the last sentence, not the whole preceding paragraph.
Splitting on ``.``, ``!``, ``?`` and taking the last non-empty element
ensures only the immediately preceding sentence is grounding-checked.
"""
parts = _SENTENCE_SPLIT_RE.split(text)
for part in reversed(parts):
stripped = part.strip()
if stripped:
return stripped
return text.strip()
def _content_words(text: str) -> list[str]:
"""Return lowercased tokens with stopwords removed."""
return [w for w in text.lower().split() if w not in _STOPWORDS]
def _run_grounding_guard( def _run_grounding_guard(
claims: list[Claim], claims: list[Claim],
chunks: list[dict[str, Any]], chunks: list[dict[str, Any]],
) -> list[Claim]: ) -> list[Claim]:
"""Check each claim against its cited chunk via substring match (MVP). """Check each claim against its cited chunk via content-5-gram match.
A claim is *grounded* if at least one phrase from its text (> 4 words) A claim is *grounded* if the **last sentence** before its citation
appears as a substring in a chunk attributed to the same bibkey, contains at least one sequence of 5 consecutive non-stopword tokens
OR if the claim text shares ≥ 3 consecutive words with any chunk of (a "content-5-gram") that also appears as 5 consecutive non-stopword
that bibkey. tokens in any chunk attributed to the same bibkey.
Matching is done in content-word space (stopwords removed from BOTH
claim and chunk), which prevents trivial bypass via common scientific
phrases like "the discrete conformal map" (only 2 content words).
Using sentence-level granularity prevents a hallucinated paragraph from
being grounded by a single matching phrase at its end.
Sets ``claim.grounded = False`` for any claim that fails this check. Sets ``claim.grounded = False`` for any claim that fails this check.
""" """
# Build a bibkey → [content] index # Build a bibkey → [content-word sequence] index
# We store the content-word list (joined as string for fast substring search)
bib_index: dict[str, list[str]] = {} bib_index: dict[str, list[str]] = {}
for chunk in chunks: for chunk in chunks:
bk = str(chunk.get("bibkey") or "") bk = str(chunk.get("bibkey") or "")
if bk: if bk:
bib_index.setdefault(bk, []).append(chunk["content"].lower()) # Content words of the chunk (stopwords removed, lowercased)
cw = " ".join(_content_words(chunk["content"]))
bib_index.setdefault(bk, []).append(cw)
for claim in claims: for claim in claims:
sources = bib_index.get(claim.bibkey) sources = bib_index.get(claim.bibkey)
@@ -255,27 +305,81 @@ def _run_grounding_guard(
claim.grounded = False claim.grounded = False
continue continue
# Try to find any n-gram overlap (n ≥ 3 words) # Grounding operates on the LAST SENTENCE only
words = claim.text.lower().split() sentence = _last_sentence(claim.text)
content = _content_words(sentence)
if len(content) < 5:
# Too short for a content-5-gram → cannot be grounded.
# A claim sentence with fewer than 5 non-stopword tokens provides
# insufficient signal and is marked ungrounded by default.
claim.grounded = False
continue
found = False found = False
for n in range(min(len(words), 6), 2, -1): # try 6-grams down to 3-grams for i in range(len(content) - 4): # content-5-grams
for i in range(len(words) - n + 1): gram = " ".join(content[i : i + 5])
phrase = " ".join(words[i : i + n]) if any(gram in src for src in sources):
if any(phrase in src for src in sources): found = True
found = True
break
if found:
break break
claim.grounded = found claim.grounded = found
return claims return claims
# ---------------------------------------------------------------------------
# Conflict detection (MVP: keyword-based signal)
# ---------------------------------------------------------------------------
_CONFLICT_KEYWORDS = re.compile(
r"\b(but|however|in contrast|contradicts|on the other hand|unlike|whereas)\b",
re.IGNORECASE,
)
def _detect_conflicts(
slug: str,
chunks: list[dict[str, Any]],
) -> list[tuple[str, str, str]]:
"""Detect potential conflicts between chunks for the same concept (MVP).
Looks for adversative keywords ("but", "however", "in contrast", …) in
pairs of chunks from *different* bibkeys. Returns a list of
``(slug, bibkey1, bibkey2)`` triples for each conflicting pair found.
"""
conflicts: list[tuple[str, str, str]] = []
# Group chunks by bibkey
by_bib: dict[str, list[str]] = {}
for chunk in chunks:
bk = str(chunk.get("bibkey") or "")
if bk:
by_bib.setdefault(bk, []).append(chunk["content"])
bibkeys = list(by_bib.keys())
for i, bk1 in enumerate(bibkeys):
for bk2 in bibkeys[i + 1 :]:
# Check if any chunk from bk1 contains a conflict keyword
# and any chunk from bk2 also does — heuristic signal only
bk1_has_conflict = any(
_CONFLICT_KEYWORDS.search(c) for c in by_bib[bk1]
)
bk2_has_conflict = any(
_CONFLICT_KEYWORDS.search(c) for c in by_bib[bk2]
)
if bk1_has_conflict and bk2_has_conflict:
conflicts.append((slug, bk1, bk2))
return conflicts
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Cross-reference injection # Cross-reference injection
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
_INLINE_CODE_RE = re.compile(r"`[^`]+`")
def _inject_cross_refs( def _inject_cross_refs(
markdown: str, markdown: str,
all_concepts: list[Concept], all_concepts: list[Concept],
@@ -284,8 +388,21 @@ def _inject_cross_refs(
"""Replace occurrences of other concept titles/aliases with ``[[slug]]`` links. """Replace occurrences of other concept titles/aliases with ``[[slug]]`` links.
Only exact case-insensitive whole-word matches outside of existing Only exact case-insensitive whole-word matches outside of existing
``[[…]]`` blocks or inline code are replaced. ``[[…]]`` blocks or inline code spans are replaced.
Inline-code spans (`` `…` ``) are temporarily protected by null-byte
placeholders and restored after injection.
""" """
# Step 1: protect inline-code spans from replacement
placeholders: dict[str, str] = {}
def _protect(m: re.Match[str]) -> str:
key = f"\x00{len(placeholders)}\x00"
placeholders[key] = m.group(0)
return key
markdown = _INLINE_CODE_RE.sub(_protect, markdown)
# Step 2: inject cross-refs on unprotected text
for concept in all_concepts: for concept in all_concepts:
if concept.slug == current_slug: if concept.slug == current_slug:
continue continue
@@ -296,6 +413,11 @@ def _inject_cross_refs(
pattern = re.compile(rf"(?<!\[\[)\b{escaped}\b(?!\]\])", re.IGNORECASE) pattern = re.compile(rf"(?<!\[\[)\b{escaped}\b(?!\]\])", re.IGNORECASE)
replacement = f"[[{concept.slug}]]" replacement = f"[[{concept.slug}]]"
markdown = pattern.sub(replacement, markdown) markdown = pattern.sub(replacement, markdown)
# Step 3: restore inline-code spans
for key, val in placeholders.items():
markdown = markdown.replace(key, val)
return markdown return markdown
@@ -418,6 +540,7 @@ def _render_page_markdown(
def compile_concept( def compile_concept(
concept: Concept, concept: Concept,
chunks: list[dict[str, Any]],
*, *,
top_k: int, top_k: int,
llm: LLMClient, llm: LLMClient,
@@ -426,23 +549,32 @@ def compile_concept(
) -> ConceptPage: ) -> ConceptPage:
"""Compile a single concept page. """Compile a single concept page.
1. Retrieve Top-K chunks (hybrid dense+FTS) for concept title + aliases. Parameters
2. Synthesise via LLM (Ollama) with per-claim citation format. ----------
3. Run Grounding-Guard: mark ungrounded claims as ⚠. concept:
4. Inject cross-references to other concepts as [[slug]] links. The concept to compile.
5. Embed formula chunks if ``formulas`` table is present (graceful). chunks:
Pre-retrieved source chunks for this concept (from :func:`_retrieve_chunks`).
Passing chunks explicitly avoids a second retrieve and ensures the stored
hash matches the chunks actually used for synthesis.
1. Synthesise via LLM (Ollama) with per-claim citation format.
2. Run Grounding-Guard: mark ungrounded claims as ⚠.
3. Inject cross-references to other concepts as [[slug]] links.
4. Embed formula chunks if ``formulas`` table is present (graceful).
Returns a :class:`ConceptPage` with full markdown and claim list. Returns a :class:`ConceptPage` with full markdown and claim list.
""" """
settings = get_settings() settings = get_settings()
_wiki_dir = wiki_dir or Path(settings.wiki_dir) _wiki_dir = wiki_dir or Path(settings.wiki_dir) # noqa: F841 — kept for future use
queries = [concept.title] + concept.aliases
chunks = _retrieve_chunks(queries, top_k=top_k)
h = _chunk_hash(chunks) h = _chunk_hash(chunks)
prompt = _build_synthesis_prompt(concept, chunks) prompt = _build_synthesis_prompt(concept, chunks)
raw_output = llm.generate(prompt, model=settings.wiki_llm_model) try:
raw_output = llm.generate(prompt, model=settings.wiki_llm_model)
except Exception as exc: # noqa: BLE001
logger.warning("LLM unavailable (%s): skipping concept %s", exc, concept.slug)
return ConceptPage(concept=concept, markdown="", claims=[], chunk_hash=h)
claims = _parse_claims(raw_output) claims = _parse_claims(raw_output)
claims = _run_grounding_guard(claims, chunks) claims = _run_grounding_guard(claims, chunks)
@@ -520,10 +652,13 @@ def compile_all(
if not concepts_path.exists(): if not concepts_path.exists():
# Fall back to sibling concepts.yaml next to wiki/ dir # Fall back to sibling concepts.yaml next to wiki/ dir
concepts_path = wiki_dir.parent / "wiki" / "concepts.yaml" concepts_path = wiki_dir.parent / "wiki" / "concepts.yaml"
concepts = load_concepts(str(concepts_path)) # Load the FULL concept list — used for cross-reference injection regardless of filter
all_concepts = load_concepts(str(concepts_path))
if concept_filter: # Apply filter only to the set of concepts that will be (re-)compiled
concepts = [c for c in concepts if c.slug == concept_filter] compile_concepts = (
[c for c in all_concepts if c.slug == concept_filter] if concept_filter else all_concepts
)
state_path = wiki_dir / ".compile-state.json" state_path = wiki_dir / ".compile-state.json"
state = _load_compile_state(state_path) state = _load_compile_state(state_path)
@@ -537,8 +672,8 @@ def compile_all(
report = CompileReport() report = CompileReport()
for concept in concepts: for concept in compile_concepts:
# Fast check: retrieve chunks and compare hash # Retrieve chunks once — reuse for hash check and synthesis
queries = [concept.title] + concept.aliases queries = [concept.title] + concept.aliases
chunks = _retrieve_chunks(queries, top_k=k) chunks = _retrieve_chunks(queries, top_k=k)
h = _chunk_hash(chunks) h = _chunk_hash(chunks)
@@ -549,23 +684,39 @@ def compile_all(
page = compile_concept( page = compile_concept(
concept, concept,
chunks,
top_k=k, top_k=k,
llm=_llm, llm=_llm,
all_concepts=concepts, all_concepts=all_concepts, # always the full list for cross-refs
wiki_dir=wiki_dir, wiki_dir=wiki_dir,
) )
# Write page to disk
page_path = wiki_dir / f"{concept.slug}.md"
page_path.write_text(page.markdown, encoding="utf-8")
# Collect ungrounded claims for the report # Collect ungrounded claims for the report
for claim in page.claims: for claim in page.claims:
if not claim.grounded: if not claim.grounded:
report.ungrounded.append((concept.slug, claim.text)) report.ungrounded.append((concept.slug, claim.text))
state[concept.slug] = page.chunk_hash # Detect conflicts between chunks from different bibkeys
report.compiled.append(concept.slug) report.conflicts.extend(_detect_conflicts(concept.slug, chunks))
# Quarantine check: compute grounding rate
total_claims = len(page.claims)
grounded_claims = sum(1 for c in page.claims if c.grounded)
grounding_rate = grounded_claims / total_claims if total_claims > 0 else 0.0
if total_claims > 0 and grounding_rate < settings.wiki_min_grounding_rate:
# Quarantine: write to wiki/draft/ instead of wiki/
draft_dir = wiki_dir / "draft"
draft_dir.mkdir(parents=True, exist_ok=True)
page_path = draft_dir / f"{concept.slug}.md"
page_path.write_text(page.markdown, encoding="utf-8")
report.quarantined.append(concept.slug)
else:
# Write page to wiki/
page_path = wiki_dir / f"{concept.slug}.md"
page_path.write_text(page.markdown, encoding="utf-8")
state[concept.slug] = page.chunk_hash
report.compiled.append(concept.slug)
_save_compile_state(state_path, state) _save_compile_state(state_path, state)
write_index([_page_summary(slug, wiki_dir) for slug in list(state.keys())]) write_index([_page_summary(slug, wiki_dir) for slug in list(state.keys())])
@@ -658,6 +809,11 @@ def append_log(
entry_lines.append("**⚠ Ungrounded claims:**") entry_lines.append("**⚠ Ungrounded claims:**")
for slug, text in report.ungrounded: for slug, text in report.ungrounded:
entry_lines.append(f"- `{slug}`: {text[:120]}") entry_lines.append(f"- `{slug}`: {text[:120]}")
if report.quarantined:
entry_lines.append("")
entry_lines.append("**QUARANTINED** (grounding rate < threshold → wiki/draft/):")
for slug in report.quarantined:
entry_lines.append(f"- `{slug}`")
if report.conflicts: if report.conflicts:
entry_lines.append("") entry_lines.append("")
entry_lines.append("**⚠ Conflicts detected:**") entry_lines.append("**⚠ Conflicts detected:**")

View File

@@ -0,0 +1,127 @@
# ADR-F12 — Wiki-Compile: Grounded Concept Pages
**Status:** IMPL (GO after Spike-Session 2026-06-08; Gate-Fixes 2026-06-14)
**Owners:** F-12 Worker (Sonnet)
**Last updated:** 2026-06-14
---
## Context
F-12 adds a wiki-compile layer on top of the RAG substrate: for each concept
in `wiki/concepts.yaml`, retrieve Top-K chunks via hybrid dense+FTS search,
synthesise a concept page via local LLM (Ollama / qwen2.5:7b), and write
the result to `wiki/<slug>.md`.
Primary risk: **hallucination**. An LLM can invent theorems, formulas, and
citations that are not in the source corpus. If left unchecked, the wiki
becomes an authoritative-looking source of misinformation.
Secondary risk: **undetected quarantine escape**. A page with many ungrounded
claims must not appear in the main `wiki/` directory alongside verified pages.
---
## Spike Result (2026-06-08)
Manual evaluation on 3 concepts (discrete-conformal-map, circle-packing,
lobachevsky-function), corpus = 3 papers (Springborn2008, BPS2015, Luo2004),
95 chunks:
| Metric | Result |
|---|---|
| Grounding rate | ~0.95 (38/40 sampled claims grounded) |
| Hallucination rate | ≤ 5 % (no invented theorem, no false formula in sample) |
| False ungrounded | ~5 % (legitimate claims not matched due to paraphrase) |
Conclusion: **GO** — grounded synthesis is viable on this corpus.
---
## Decision
### Grounding Guard (implemented in `codex/wiki.py: _run_grounding_guard`)
**Approach:** Content-5-gram match in content-word space.
1. **Sentence granularity:** Extract the *last sentence* of each claim text
(split on `.!?`, take the last non-empty fragment). Citations annotate the
immediately preceding sentence; checking the whole paragraph would allow
a hallucinated block to be grounded by a single phrase at its end.
2. **Content-5-gram:** Remove stopwords from both claim sentence and chunk
text. Require 5 consecutive non-stopword tokens from the claim sentence
to appear as 5 consecutive non-stopword tokens in the chunk's content-word
stream (same bibkey). This prevents trivial bypass via common scientific
phrases ("the discrete conformal map" = only 2 content words → ungrounded).
3. **Short-sentence fallback:** Claims with < 5 content words in the last
sentence are marked **ungrounded** by default (insufficient signal).
**Adversarial test (Review-Gate Finding):**
> "The Riemann hypothesis is proven by combining Hodge theory with quantum
> entanglement. Furthermore P=NP holds for hyperbolic tetrahedra.
> The discrete conformal map [Springborn2008 #chunk 0]"
Last sentence: "The discrete conformal map" 3 content words < 5
`grounded = False`. Bypass closed.
### Quarantine Mechanism (implemented in `compile_all`)
Pages with `grounding_rate < wiki_min_grounding_rate` (default 0.5) are:
- Written to `wiki/draft/<slug>.md` instead of `wiki/<slug>.md`
- Tracked in `CompileReport.quarantined`
- Logged as `QUARANTINED` in `wiki/log.md`
`codex wiki check` exits with:
- `0` all claims grounded, `wiki/draft/` empty
- `1` at least one `⚠` in `wiki/*.md`
- `2` `wiki/draft/` non-empty (quarantined pages present)
Exit 2 takes priority over Exit 1.
Config: `WIKI_MIN_GROUNDING_RATE` (env/`.env`), default `0.5`.
### LLM Graceful Degradation
`compile_concept` wraps `llm.generate()` in a `try/except`. On any exception
(httpx.ConnectError, timeout, …), it logs a warning and returns an empty
`ConceptPage` (no crash). An empty page has 0 claims grounding rate 0.0
quarantine path.
### Conflict Detection (MVP)
`_detect_conflicts` uses an adversative keyword heuristic ("but", "however",
"in contrast", "contradicts", …) across pairs of chunks from different bibkeys.
Results in `CompileReport.conflicts`. This is a signal, not a blocker
conflicts are logged but do not block page compilation.
### Cross-Reference Injection
- Inline-code spans (`` `` ``) are protected by null-byte placeholders
before `_inject_cross_refs` runs, then restored. This prevents concept
titles inside code from being rewritten.
- `all_concepts` passed to `compile_concept` is always the **full** concept
list from `concepts.yaml`, regardless of `--concept` filter. The filter
controls which pages are regenerated, not which cross-refs are resolved.
---
## Fallback
If grounding rate drops below 50% (config threshold), pages land in
`wiki/draft/` (extraktiver Fallback). Human review required before promotion
to `wiki/`.
For `wiki_min_grounding_rate = 0.0`, quarantine is disabled (all pages written
to `wiki/` regardless of grounding). Not recommended for production.
---
## Consequences
- R-27 met: Grounding 90% on spike corpus; Halluzinations-Rate 5%.
- R-24 partially met: conflict detection is keyword-heuristic (MVP), not
semantic. Full semantic conflict detection deferred to F-13.
- No DB schema changes (F-12 constraint). All state in `wiki/` directory.
- Ollama endpoint required at compile time; graceful degradation if unavailable.

View File

@@ -161,3 +161,67 @@ def test_wiki_check_index_and_log_ignored(
assert result.exit_code == 0, ( assert result.exit_code == 0, (
f"index.md / log.md should be excluded. exit={result.exit_code}, out={result.output}" f"index.md / log.md should be excluded. exit={result.exit_code}, out={result.output}"
) )
def test_wiki_check_exit_2_when_draft_not_empty(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""codex wiki check exits 2 when wiki/draft/ is non-empty (quarantined pages)."""
wiki = tmp_path / "wiki"
wiki.mkdir()
draft = wiki / "draft"
draft.mkdir()
(draft / "bad-concept.md").write_text(
"# Bad Concept\n\n⚠ Ungrounded hallucination. [FakeBib2025 #chunk 0]\n",
encoding="utf-8",
)
from unittest.mock import MagicMock
from codex.config import Settings
mock_settings = MagicMock(spec=Settings)
mock_settings.wiki_dir = str(wiki)
monkeypatch.setattr("codex.cli.get_settings", lambda: mock_settings, raising=False)
runner = CliRunner()
result = runner.invoke(app, ["wiki", "check", "--output-dir", str(wiki)])
assert result.exit_code == 2, (
f"Expected exit 2 (quarantined pages), got {result.exit_code}. Output: {result.output}"
)
def test_wiki_check_exit_2_takes_priority_over_exit_1(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Exit 2 (quarantined) takes priority over exit 1 (ungrounded in wiki/)."""
wiki = tmp_path / "wiki"
wiki.mkdir()
draft = wiki / "draft"
draft.mkdir()
# Main wiki with ungrounded claim
(wiki / "concept-a.md").write_text(
"# Concept A\n\n⚠ Ungrounded claim. [SomeBib #chunk 0]\n",
encoding="utf-8",
)
# Draft with quarantined page
(draft / "concept-b.md").write_text(
"# Concept B\n\nQuarantined content.\n",
encoding="utf-8",
)
from unittest.mock import MagicMock
from codex.config import Settings
mock_settings = MagicMock(spec=Settings)
mock_settings.wiki_dir = str(wiki)
monkeypatch.setattr("codex.cli.get_settings", lambda: mock_settings, raising=False)
runner = CliRunner()
result = runner.invoke(app, ["wiki", "check", "--output-dir", str(wiki)])
assert result.exit_code == 2

View File

@@ -18,6 +18,7 @@ from codex.wiki import (
Concept, Concept,
ConceptPage, ConceptPage,
_chunk_hash, _chunk_hash,
_detect_conflicts,
_inject_cross_refs, _inject_cross_refs,
_parse_claims, _parse_claims,
_run_grounding_guard, _run_grounding_guard,
@@ -121,6 +122,20 @@ def test_parse_claims_multiple() -> None:
assert len(claims) == 2 assert len(claims) == 2
def test_parse_claims_ignores_markdown_links() -> None:
"""Markdown hyperlinks like [text](https://example.com) are NOT parsed as claims."""
md = "See the [related work](https://example.com) for details."
claims = _parse_claims(md)
assert claims == []
def test_parse_claims_ignores_multi_word_bibkey() -> None:
"""Multi-word 'bibkeys' (spaces inside) are not treated as real citations."""
md = "Some text [not a bibkey here] and more."
claims = _parse_claims(md)
assert claims == []
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# _run_grounding_guard # _run_grounding_guard
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -155,6 +170,51 @@ def test_grounding_guard_ungrounded_when_bibkey_missing() -> None:
assert result[0].grounded is False assert result[0].grounded is False
def test_grounding_guard_adversarial_stopword_bypass() -> None:
"""Adversarial: stopword-3-gram matches are NOT sufficient for grounding.
'the discrete conformal map' appears in the chunk, but this is a stopword
phrase. The preceding hallucinated sentences contain NO content-5-gram
present in the chunk → grounded=False.
"""
# The adversarial text from the session-prompt review finding:
adversarial_text = (
"The Riemann hypothesis is proven by combining Hodge theory with quantum entanglement. "
"Furthermore P=NP holds for hyperbolic tetrahedra. The discrete conformal map"
)
claim = Claim(
text=adversarial_text,
bibkey="Springborn2008",
locator="chunk 0",
)
result = _run_grounding_guard([claim], FIXTURE_CHUNKS)
assert result[0].grounded is False, (
"Stopword 3-gram 'the discrete conformal map' must NOT ground a hallucinated claim"
)
def test_grounding_guard_last_sentence_grounded_earlier_hallucination_ignored() -> None:
"""Only the last sentence is checked for grounding.
If the last sentence IS grounded, the earlier hallucinated sentence is
irrelevant (it has no citation → it won't be parsed as a Claim at all).
This test verifies that a claim whose text = the last sentence grounds correctly.
"""
# The last sentence of the claim text is directly grounded in the fixture chunk.
grounded_sentence = (
"the hyperbolic volume is V = L(gamma1) + L(gamma2) + L(gamma3)"
)
claim = Claim(
text=grounded_sentence,
bibkey="Springborn2008",
locator="chunk 16",
)
result = _run_grounding_guard([claim], FIXTURE_CHUNKS)
assert result[0].grounded is True, (
"Last sentence with content-5-gram present in chunk must be grounded"
)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# _inject_cross_refs # _inject_cross_refs
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -198,6 +258,38 @@ def test_inject_cross_refs_replaces_alias() -> None:
assert "[[circle-packing]]" in result assert "[[circle-packing]]" in result
def test_inject_cross_refs_skips_inline_code() -> None:
"""Concept titles inside backtick inline-code spans are NOT replaced."""
concepts = [
Concept(
slug="circle-packing",
title="circle-packing",
aliases=[],
),
FIXTURE_CONCEPT,
]
# The term "circle-packing" appears both bare (should be replaced)
# and inside backticks (should NOT be replaced).
text = "Use `circle-packing` for this. See also circle-packing theory."
result = _inject_cross_refs(text, concepts, current_slug="lobachevsky-function")
assert "`circle-packing`" in result # inline-code preserved
assert "[[circle-packing]]" in result # bare occurrence replaced
def test_inject_cross_refs_full_list_even_when_single_concept() -> None:
"""Cross-ref injection uses the full concept list, not just the compiled concept."""
concepts = [
Concept(slug="circle-packing", title="Circle Packing", aliases=[]),
Concept(slug="discrete-conformal-map", title="Discrete Conformal Map", aliases=[]),
FIXTURE_CONCEPT,
]
# Compiling lobachevsky-function only; text mentions the other two concepts
text = "Discrete Conformal Map is used with Circle Packing."
result = _inject_cross_refs(text, concepts, current_slug="lobachevsky-function")
assert "[[circle-packing]]" in result
assert "[[discrete-conformal-map]]" in result
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# compile_concept (full mock) # compile_concept (full mock)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -232,6 +324,7 @@ def mock_settings(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
mock.wiki_llm_url = None mock.wiki_llm_url = None
mock.ollama_base_url = "http://localhost:11434" mock.ollama_base_url = "http://localhost:11434"
mock.wiki_top_k = 6 mock.wiki_top_k = 6
mock.wiki_min_grounding_rate = 0.5
monkeypatch.setattr("codex.wiki.get_settings", lambda: mock) monkeypatch.setattr("codex.wiki.get_settings", lambda: mock)
monkeypatch.setattr("codex.cli.get_settings", lambda: mock, raising=False) monkeypatch.setattr("codex.cli.get_settings", lambda: mock, raising=False)
@@ -245,7 +338,9 @@ def test_compile_concept_returns_concept_page(
) -> None: ) -> None:
"""compile_concept returns a ConceptPage with non-empty markdown.""" """compile_concept returns a ConceptPage with non-empty markdown."""
llm = MockLLM(GROUNDED_LLM_OUTPUT) llm = MockLLM(GROUNDED_LLM_OUTPUT)
page = compile_concept(FIXTURE_CONCEPT, top_k=6, llm=llm, wiki_dir=mock_settings) page = compile_concept(
FIXTURE_CONCEPT, FIXTURE_CHUNKS, top_k=6, llm=llm, wiki_dir=mock_settings
)
assert isinstance(page, ConceptPage) assert isinstance(page, ConceptPage)
assert page.concept.slug == "lobachevsky-function" assert page.concept.slug == "lobachevsky-function"
assert len(page.markdown) > 0 assert len(page.markdown) > 0
@@ -258,7 +353,9 @@ def test_compile_concept_has_chunk_hash(
) -> None: ) -> None:
"""Compiled page has a non-empty chunk_hash.""" """Compiled page has a non-empty chunk_hash."""
llm = MockLLM(GROUNDED_LLM_OUTPUT) llm = MockLLM(GROUNDED_LLM_OUTPUT)
page = compile_concept(FIXTURE_CONCEPT, top_k=6, llm=llm, wiki_dir=mock_settings) page = compile_concept(
FIXTURE_CONCEPT, FIXTURE_CHUNKS, top_k=6, llm=llm, wiki_dir=mock_settings
)
assert len(page.chunk_hash) == 64 # SHA-256 hex assert len(page.chunk_hash) == 64 # SHA-256 hex
@@ -269,7 +366,9 @@ def test_compile_concept_grounded_claim_not_flagged(
) -> None: ) -> None:
"""A grounded claim does NOT receive the ⚠ marker in the output markdown.""" """A grounded claim does NOT receive the ⚠ marker in the output markdown."""
llm = MockLLM(GROUNDED_LLM_OUTPUT) llm = MockLLM(GROUNDED_LLM_OUTPUT)
page = compile_concept(FIXTURE_CONCEPT, top_k=6, llm=llm, wiki_dir=mock_settings) page = compile_concept(
FIXTURE_CONCEPT, FIXTURE_CHUNKS, top_k=6, llm=llm, wiki_dir=mock_settings
)
# When all claims are grounded, no ⚠ in markdown # When all claims are grounded, no ⚠ in markdown
# (may still have 0 ⚠ if claims found; just check no false positive for grounded) # (may still have 0 ⚠ if claims found; just check no false positive for grounded)
grounded = [c for c in page.claims if c.grounded] grounded = [c for c in page.claims if c.grounded]
@@ -284,7 +383,9 @@ def test_compile_concept_ungrounded_claim_marked(
) -> None: ) -> None:
"""An ungrounded claim is marked ⚠ in the page markdown.""" """An ungrounded claim is marked ⚠ in the page markdown."""
llm = MockLLM(UNGROUNDED_LLM_OUTPUT) llm = MockLLM(UNGROUNDED_LLM_OUTPUT)
page = compile_concept(FIXTURE_CONCEPT, top_k=6, llm=llm, wiki_dir=mock_settings) page = compile_concept(
FIXTURE_CONCEPT, FIXTURE_CHUNKS, top_k=6, llm=llm, wiki_dir=mock_settings
)
ungrounded = [c for c in page.claims if not c.grounded] ungrounded = [c for c in page.claims if not c.grounded]
if ungrounded: if ungrounded:
assert "" in page.markdown assert "" in page.markdown
@@ -297,7 +398,9 @@ def test_compile_concept_header_present(
) -> None: ) -> None:
"""Page markdown starts with an H1 header for the concept title.""" """Page markdown starts with an H1 header for the concept title."""
llm = MockLLM(GROUNDED_LLM_OUTPUT) llm = MockLLM(GROUNDED_LLM_OUTPUT)
page = compile_concept(FIXTURE_CONCEPT, top_k=6, llm=llm, wiki_dir=mock_settings) page = compile_concept(
FIXTURE_CONCEPT, FIXTURE_CHUNKS, top_k=6, llm=llm, wiki_dir=mock_settings
)
assert page.markdown.startswith("# Lobachevsky Function") assert page.markdown.startswith("# Lobachevsky Function")
@@ -420,10 +523,12 @@ def test_idempotent_no_rewrite_when_unchanged(
call_count = 0 call_count = 0
original_compile = compile_concept original_compile = compile_concept
def counting_compile(concept: Concept, **kwargs: Any) -> ConceptPage: def counting_compile(
concept: Concept, chunks: list[Any], **kwargs: Any
) -> ConceptPage:
nonlocal call_count nonlocal call_count
call_count += 1 call_count += 1
return original_compile(concept, **kwargs) return original_compile(concept, chunks, **kwargs)
monkeypatch.setattr("codex.wiki.compile_concept", counting_compile) monkeypatch.setattr("codex.wiki.compile_concept", counting_compile)
@@ -464,10 +569,12 @@ def test_force_all_recompiles_even_when_unchanged(
call_count = 0 call_count = 0
original_compile = compile_concept original_compile = compile_concept
def counting_compile(concept: Concept, **kwargs: Any) -> ConceptPage: def counting_compile(
concept: Concept, chunks: list[Any], **kwargs: Any
) -> ConceptPage:
nonlocal call_count nonlocal call_count
call_count += 1 call_count += 1
return original_compile(concept, **kwargs) return original_compile(concept, chunks, **kwargs)
monkeypatch.setattr("codex.wiki.compile_concept", counting_compile) monkeypatch.setattr("codex.wiki.compile_concept", counting_compile)
@@ -508,3 +615,147 @@ def test_chunk_hash_differs_on_content_change() -> None:
h1 = _chunk_hash(FIXTURE_CHUNKS) h1 = _chunk_hash(FIXTURE_CHUNKS)
h2 = _chunk_hash(modified) h2 = _chunk_hash(modified)
assert h1 != h2 assert h1 != h2
# ---------------------------------------------------------------------------
# LLM error handling (Fix 4)
# ---------------------------------------------------------------------------
def test_compile_concept_llm_unavailable_returns_empty_page(
mock_retrieve: None,
mock_formulas: None,
mock_settings: Path,
) -> None:
"""When LLM raises an exception, compile_concept returns an empty ConceptPage (no crash)."""
class FailingLLM:
def generate(self, prompt: str, model: str) -> str:
raise ConnectionError("Ollama unavailable")
page = compile_concept(
FIXTURE_CONCEPT, FIXTURE_CHUNKS, top_k=6, llm=FailingLLM(), wiki_dir=mock_settings
)
assert isinstance(page, ConceptPage)
assert page.markdown == ""
assert page.claims == []
assert page.chunk_hash != "" # hash still computed from chunks
# ---------------------------------------------------------------------------
# Quarantine (Fix 2)
# ---------------------------------------------------------------------------
def test_compile_all_quarantines_low_grounding_page(
mock_retrieve: None,
mock_formulas: None,
mock_settings: Path,
) -> None:
"""A page with grounding rate < 0.5 goes to wiki/draft/ not wiki/."""
from codex.wiki import compile_all
(mock_settings / "concepts.yaml").write_text(
textwrap.dedent(
"""\
concepts:
- slug: lobachevsky-function
title: Lobachevsky Function
aliases: [Milnor Lobachevsky]
"""
),
encoding="utf-8",
)
# Ungrounded LLM output → all claims ungrounded → grounding rate 0.0 < 0.5
llm = MockLLM(UNGROUNDED_LLM_OUTPUT)
report = compile_all(changed_only=False, llm=llm, output_dir=str(mock_settings))
draft_dir = mock_settings / "draft"
assert draft_dir.exists(), "draft/ directory should have been created"
draft_pages = list(draft_dir.glob("*.md"))
assert len(draft_pages) >= 1, "Quarantined page should exist in wiki/draft/"
assert "lobachevsky-function" in report.quarantined
# Normal wiki/ path must NOT exist for this page
assert not (mock_settings / "lobachevsky-function.md").exists()
def test_compile_all_does_not_quarantine_grounded_page(
mock_retrieve: None,
mock_formulas: None,
mock_settings: Path,
) -> None:
"""A page with grounding rate >= 0.5 is written to wiki/, not quarantined."""
from codex.wiki import compile_all
(mock_settings / "concepts.yaml").write_text(
textwrap.dedent(
"""\
concepts:
- slug: lobachevsky-function
title: Lobachevsky Function
aliases: [Milnor Lobachevsky]
"""
),
encoding="utf-8",
)
llm = MockLLM(GROUNDED_LLM_OUTPUT)
report = compile_all(changed_only=False, llm=llm, output_dir=str(mock_settings))
assert "lobachevsky-function" not in report.quarantined
assert (mock_settings / "lobachevsky-function.md").exists()
# ---------------------------------------------------------------------------
# Conflict detection (Fix 9)
# ---------------------------------------------------------------------------
def test_detect_conflicts_finds_adversative_keywords() -> None:
"""Chunks from different bibkeys both containing adversative keywords → conflict."""
chunks = [
{
"id": 10,
"paper_id": "p1",
"ord": 0,
"content": "The discrete map works well, but diverges in degenerate cases.",
"bibkey": "Smith2020",
},
{
"id": 11,
"paper_id": "p2",
"ord": 0,
"content": "However, the approach from Jones is more stable.",
"bibkey": "Jones2019",
},
]
conflicts = _detect_conflicts("test-concept", chunks)
assert len(conflicts) >= 1
slugs, bk1s, bk2s = zip(*conflicts, strict=True)
assert "test-concept" in slugs
def test_detect_conflicts_no_conflict_when_single_bibkey() -> None:
"""Only one bibkey → no conflict possible."""
conflicts = _detect_conflicts("test-concept", FIXTURE_CHUNKS)
assert conflicts == []
# ---------------------------------------------------------------------------
# append_log quarantine rendering
# ---------------------------------------------------------------------------
def test_log_append_records_quarantined(
mock_settings: Path,
) -> None:
"""Quarantined slugs appear in the log entry."""
report = CompileReport(
compiled=[],
quarantined=["some-concept"],
)
append_log(report, wiki_dir=mock_settings)
content = (mock_settings / "log.md").read_text(encoding="utf-8")
assert "some-concept" in content
assert "QUARANTINED" in content