Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 41 additions & 2 deletions src/vouch/health.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

from . import index_db
from .audit import count_events, verify_chain
from .models import Claim, ClaimStatus, Entity, Page, ProposalKind, ProposalStatus
from .models import Claim, ClaimStatus, Entity, Page, ProposalKind, ProposalStatus, Source
from .storage import KBStore, _yaml_load, sha256_hex
from .verify import verify_all

Expand Down Expand Up @@ -121,9 +121,48 @@ def _load_claims_for_lint(store: KBStore) -> tuple[list[Claim], list[Finding]]:
return valid, findings


def _collect_sources_for_lint(store: KBStore) -> tuple[set[str], list[Finding]]:
"""Source ids for citation checks, with unreadable metas surfaced.

`storage.list_sources()` deliberately skips files `_load_or_skip`
can't parse so bulk listings survive one bad artifact; for lint that
silence is the bug — a corrupt meta.yaml (e.g. a mojibake title
carrying a raw control character that pyyaml rejects) must become an
`unreadable_source` finding. The source id is its directory name, so
it still counts as present for citation checks: the claim's citation
isn't broken, the meta is, and reporting `broken_citation` on top
would point the repair at the wrong file."""
present: set[str] = set()
findings: list[Finding] = []
sources_dir = store.kb_dir / "sources"
if not sources_dir.is_dir():
return present, findings
for sdir in sorted(sources_dir.iterdir()):
meta = sdir / "meta.yaml"
if not meta.exists():
continue
sid = sdir.name
present.add(sid)
try:
Source.model_validate(_yaml_load(meta.read_text(encoding="utf-8")))
except Exception as e:
findings.append(
Finding(
"error",
"unreadable_source",
f"source {sid} ({meta}) could not be loaded: {e} — "
"fix the meta.yaml by hand, or delete the source "
"directory and re-register it",
[sid],
)
)
Comment on lines +148 to +158
return present, findings


def lint(store: KBStore, *, stale_after_days: int = 180) -> HealthReport:
claims, findings = _load_claims_for_lint(store)
sources_present = {s.id for s in store.list_sources()}
sources_present, source_findings = _collect_sources_for_lint(store)
findings.extend(source_findings)
evidence_present = {e.id for e in store.list_evidence()}

for c in claims:
Expand Down
32 changes: 32 additions & 0 deletions tests/test_health.py
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,38 @@ def test_lint_surfaces_legacy_uncited_claim_yaml_without_crashing(
assert all(f.severity != "error" for f in good_findings), good_findings


def test_lint_surfaces_unreadable_source_meta(store: KBStore) -> None:
"""A source meta.yaml can carry a raw C1 control character — e.g. a
hand edit or an external writer under a mismatched locale landing a
utf-8 em dash double-decoded into latin-1 mojibake. pyyaml's reader
rejects the character outright, and storage._load_or_skip
deliberately skips the file so bulk listings survive. For lint that
skip is the bug: the corrupt meta silently vanishes from the sweep,
and any claim citing the source is misreported as broken_citation.
Lint must surface the file as an `unreadable_source` error and keep
citation checks honest — the source id is its directory name,
readable even when the yaml is not."""
src = store.put_source(b"e")
store.put_claim(Claim(id="c1", text="t", evidence=[src.id]))
meta = store.kb_dir / "sources" / src.id / "meta.yaml"
meta.write_text(
meta.read_text(encoding="utf-8") + "title: conversation \u00e2\u0080\u0094 vouch\n",
encoding="utf-8",
)

report = health.lint(store)

unreadable = [f for f in report.findings if f.code == "unreadable_source"]
assert unreadable, [f.code for f in report.findings]
assert src.id in unreadable[0].object_ids
assert report.ok is False # unreadable_source is severity=error
# The source directory still exists — the citation isn't broken, the
# meta is. Reporting broken_citation here would send the user hunting
# for a missing source that is right there on disk.
codes = {f.code for f in report.findings}
assert "broken_citation" not in codes


def test_list_claims_filtered_by_status(store: KBStore) -> None:
src = store.put_source(b"e")
store.put_claim(Claim(id="c1", text="x", evidence=[src.id],
Expand Down
Loading