From 858783f7393fe67d4a301ec399d04ca486b9d33c Mon Sep 17 00:00:00 2001 From: Nelson Mfinda Date: Thu, 7 May 2026 22:02:37 +0100 Subject: [PATCH 1/3] feat(scraper): king-scrape audit subcommand for corpus drift --- .../adr/0013-corpus-drift-audit-subcommand.md | 82 ++++ CHANGELOG.md | 12 + src/king_context/scraper/audit.py | 448 ++++++++++++++++++ src/king_context/scraper/cli.py | 9 +- tests/test_scraper/test_audit.py | 369 +++++++++++++++ 5 files changed, 919 insertions(+), 1 deletion(-) create mode 100644 .king-context/adr/0013-corpus-drift-audit-subcommand.md create mode 100644 src/king_context/scraper/audit.py create mode 100644 tests/test_scraper/test_audit.py diff --git a/.king-context/adr/0013-corpus-drift-audit-subcommand.md b/.king-context/adr/0013-corpus-drift-audit-subcommand.md new file mode 100644 index 0000000..bc26dd1 --- /dev/null +++ b/.king-context/adr/0013-corpus-drift-audit-subcommand.md @@ -0,0 +1,82 @@ +--- +id: ADR-0013 +title: Read-only drift audit for indexed corpora via `king-scrape audit` +status: accepted +date: 2026-05-07 +areas: + - scraper + - corpus + - cli +supersedes: [] +superseded_by: [] +related: + - ADR-0009 + - ADR-0010 + - ADR-0012 +keywords: + - audit + - drift-detection + - http-head + - url-health + - discover-diff + - markdown-report + - read-only +tags: + - architecture + - scraper + - cli +--- + + +# ADR-0013: Read-only drift audit for indexed corpora via `king-scrape audit` + +## Context + +A corpus committed under `data/.json` is a snapshot. The upstream docs site keeps moving pages get renamed, deprecated, removed; new sections appear. The MCP server happily serves whatever was indexed, so a contributor has no easy way to ask *"is this corpus still aligned with the live docs, or do I need to re-scrape?"* without paying full pipeline cost (Firecrawl fetch + OpenRouter enrichment) just to find out. + +ADR-0012 established content hash provenance through the pipeline so that, eventually, an `--update` flow can answer that question precisely. But: + +- Today, many indexed corpora pre-date ADR-0012 and have no `_meta.content_hash`. +- A precise content-diff requires re-fetching every page, which is the expensive part the contributor is trying to avoid before deciding to refresh. +- The most actionable signal *which URLs are flat out broken* is cheap to obtain and useful on its own. + +There is a real gap between "do nothing" and "run the full incremental refresh that ADR-0014 will introduce." That gap is what this ADR fills. + +## Decision + +Introduce a `king-scrape audit ` subcommand whose only job is to read an indexed corpus and report drift signals as a Markdown file. The audit is **read-only**: it never mutates `data/.json`, never writes to the database, never spends LLM credits. It is safe to run on a schedule. + +The audit runs two passes: + +1. **URL health (always).** Walks every unique section URL in the corpus and issues an HTTP `HEAD` (falling back to `GET` when the server returns 405/501). Results are mapped to one of: `fresh` (2xx), `moved` (3xx, with `Location` captured), `broken` (404/410), or `unreachable` (timeout / network error / other 5xx). No provider key is required for this pass, it uses `httpx` directly with bounded concurrency (10). +2. **Discovery diff (optional, default-on).** Calls the configured `DiscoveryProvider` (Firecrawl, Crawl4AI) to re-map the upstream site, then diffs the URL set against the corpus. Reports `new_urls` (in fresh discover, not in corpus) and `orphan_urls` (in corpus, not in fresh discover). `--no-discover` skips this pass for environments without provider keys. + +A new module `src/king_context/scraper/audit.py` owns the dataclasses (`SectionAudit`, `CorpusAudit`), the async URL probe, the discovery diff, and a Markdown renderer (`render_report`) plus writer (`write_report`). The CLI dispatcher in `cli.py` delegates to `audit_main(argv)` when invoked as `king-scrape audit ...`. The existing `king-scrape ` flow is unchanged `audit` is a sibling subcommand, not a refactor of the existing parser. + +Reports land at `.king-context/audit/-.md`. The exit code is `0` for clean / advisory only audits, `2` when at least one section is broken, so a CI job can gate on it. + +## Alternatives Considered + +**Inline `--audit` flag on the existing `king-scrape ` invocation.** Rejected because the existing positional argument is a URL, not a name, and conflating the two surfaces would be confusing. Subcommand dispatch keeps the two flows clearly separate at the CLI level. + +**Standalone console script (`king-audit`).** Possible, but multiplies the bin surface for a feature that conceptually lives next to the scrape pipeline. The single dispatcher in `king-scrape` matches the way Git, Cargo, and `kctx` already organise their subcommands. + +**Full content-hash diff in this PR.** Considered and deferred. ADR-0012 introduced the per-section `_meta.content_hash`; using it here would force this PR to depend on ADR-0012 landing first, and would still be incomplete for pre-ADR-0012 corpora. Splitting URL health into its own change keeps each PR focused and lets contributors get the broken URL signal even on legacy corpora. ADR-0014 (incremental refresh) is the natural home for the content-hash diff because it has to compute fresh page hashes anyway as part of deciding what to re-enrich. + +**Mutate `data/.json` with status fields.** Rejected. Keeping the corpus pristine and writing audits to a separate location preserves the simple "read only fact" model of the indexed JSON. A contributor inspecting `data/*.json` should see only what `seed_data` reads, never audit metadata. + +## Consequences + +Contributors gain a one line, read only command to check whether a corpus is still aligned with upstream. The output is plain Markdown so it diffs nicely, can be committed if desired, or pasted into an issue. + +Because the audit never spends LLM credits and never mutates state, it is safe to attach to a CI cron, a pre merge check on corpus PRs, or a manual sanity run before considering a re-scrape. + +The dispatcher pattern (`if argv[1] == "audit": ...`) creates a precedent that ADR-0014's `update` subcommand will follow without further refactor. If the project ever grows enough subcommands to warrant `argparse` subparsers, the migration is a local change inside `cli.main` with no impact on the audit module. + +A small ongoing maintenance cost: `httpx` was already a dependency through the existing pipeline, so no new transitive dependencies. The audit's HEAD pass is bounded at 10 concurrent requests by default, appropriate for typical corpora (50-500 URLs) without hammering the upstream. + +## Links + +src/king_context/scraper/audit.py +src/king_context/scraper/cli.py +.king-context/adr/0012-content-hash-provenance-and-enrichment-cache.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 6e2fe67..27edbaa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- `king-scrape audit ` subcommand (ADR-0013). Walks the URLs of an + indexed corpus and reports which are reachable, moved (3xx with the + redirect target), broken (404/410), or unreachable. Optionally + re-runs discovery against the upstream and reports URLs added or + removed since the corpus was indexed (`--no-discover` to skip). + Read-only — never mutates the corpus file or the database. No LLM + cost. Markdown report lands at `.king-context/audit/-.md`. + Exit code is `0` on clean, `2` when at least one section is broken, + so it can gate a CI job. + ## [0.4.0] - 2026-05-06 ### Added diff --git a/src/king_context/scraper/audit.py b/src/king_context/scraper/audit.py new file mode 100644 index 0000000..2effc35 --- /dev/null +++ b/src/king_context/scraper/audit.py @@ -0,0 +1,448 @@ +"""Drift / staleness audit for an indexed corpus. + +Walks the URLs of an existing ``data/.json`` corpus and reports which +ones are still reachable, which moved (redirects), which are broken (404), +and optionally which URLs the upstream docs site has gained or lost +since the corpus was indexed. Pure read-only: never mutates the corpus +file or the database. + +Designed to run cheaply (HEAD requests, no LLM calls, no provider keys +required for the URL health pass) so contributors can re-run it on a +schedule without spending OpenRouter credits. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import re +import sys +from dataclasses import dataclass, field +from datetime import datetime, timezone +from importlib.metadata import PackageNotFoundError, version as _pkg_version +from pathlib import Path +from typing import Iterable +from urllib.parse import urldefrag, urlsplit, urlunsplit + +import httpx + +from king_context import PROJECT_ROOT + + +HTTP_TIMEOUT = 15.0 +DEFAULT_CONCURRENCY = 10 +THROTTLE_RETRY_CAP_SECONDS = 30 + + +def _scraper_version() -> str: + try: + return _pkg_version("king-context") + except PackageNotFoundError: + return "dev" + + +def _canonicalize(url: str) -> str: + """Drop fragment, lowercase scheme/host, strip trailing slash from path.""" + url, _ = urldefrag(url) + parts = urlsplit(url) + path = parts.path.rstrip("/") or "/" + return urlunsplit( + (parts.scheme.lower(), parts.netloc.lower(), path, parts.query, "") + ) + + +@dataclass +class SectionAudit: + url: str + title: str + status: str # fresh | broken | moved | throttled | auth_required | unreachable | unknown + final_url: str | None = None + status_code: int | None = None + error: str | None = None + + +@dataclass +class CorpusAudit: + name: str + base_url: str + audited_at: str + corpus_path: str + sections: list[SectionAudit] = field(default_factory=list) + new_urls: list[str] = field(default_factory=list) + orphan_urls: list[str] = field(default_factory=list) + discovery_skipped: bool = False + discovery_error: str | None = None + + def counts(self) -> dict[str, int]: + out = { + "fresh": 0, + "broken": 0, + "moved": 0, + "throttled": 0, + "auth_required": 0, + "unreachable": 0, + "unknown": 0, + } + for s in self.sections: + out[s.status] = out.get(s.status, 0) + 1 + return out + + +def _candidate_corpus_paths(name: str) -> list[Path]: + """Locations to search for the corpus JSON, in order.""" + return [ + PROJECT_ROOT / ".king-context" / "data" / f"{name}.json", + PROJECT_ROOT / "data" / f"{name}.json", + ] + + +def find_corpus(name: str) -> Path: + for path in _candidate_corpus_paths(name): + if path.exists(): + return path + searched = "\n ".join(str(p) for p in _candidate_corpus_paths(name)) + raise FileNotFoundError( + f"Corpus '{name}' not found. Searched:\n {searched}" + ) + + +def _load_corpus(corpus_path: Path) -> dict: + raw = corpus_path.read_text(encoding="utf-8") + try: + return json.loads(raw) + except json.JSONDecodeError as exc: + raise ValueError( + f"corpus at {corpus_path} is not valid JSON: {exc}" + ) from exc + + +def _unique_section_urls(corpus: dict) -> list[tuple[str, str]]: + """Return ``[(url, title), ...]`` deduped by canonical URL, preserving first-seen order.""" + seen: set[str] = set() + out: list[tuple[str, str]] = [] + for section in corpus.get("sections", []): + url = section.get("url") + if not url: + continue + canonical = _canonicalize(url) + if canonical in seen: + continue + seen.add(canonical) + out.append((url, section.get("title", ""))) + return out + + +def _classify(response: httpx.Response) -> tuple[str, str | None]: + """Map an httpx response to ``(status, final_url)``. + + Caller passes a response from ``follow_redirects=True``, so a populated + ``response.history`` means we landed on the chain endpoint after one or + more 3xx hops; ``response.url`` is then the final URL. + """ + if response.history: + return "moved", str(response.url) + code = response.status_code + if 200 <= code < 300: + return "fresh", None + if code in (401, 403): + return "auth_required", None + if code == 429: + return "throttled", None + if code in (404, 410): + return "broken", None + return "unreachable", None + + +def _retry_after_seconds(response: httpx.Response) -> float: + raw = response.headers.get("retry-after") + if not raw: + return 1.0 + try: + return min(THROTTLE_RETRY_CAP_SECONDS, max(1.0, float(raw))) + except ValueError: + return 1.0 + + +async def _check_url( + client: httpx.AsyncClient, + semaphore: asyncio.Semaphore, + url: str, + title: str, +) -> SectionAudit: + async with semaphore: + try: + response = await client.head(url, follow_redirects=True) + # Some servers return 405/501 on HEAD; retry as GET. + if response.status_code in (405, 501): + response = await client.get(url, follow_redirects=True) + # 429: respect Retry-After once, then accept the result. + if response.status_code == 429: + await asyncio.sleep(_retry_after_seconds(response)) + response = await client.head(url, follow_redirects=True) + if response.status_code in (405, 501): + response = await client.get(url, follow_redirects=True) + status, final_url = _classify(response) + return SectionAudit( + url=url, + title=title, + status=status, + final_url=final_url, + status_code=response.status_code, + ) + except httpx.TimeoutException: + return SectionAudit( + url=url, title=title, status="unreachable", error="timeout" + ) + except httpx.RequestError as exc: + return SectionAudit( + url=url, title=title, status="unreachable", error=str(exc) + ) + + +async def _check_section_urls( + section_urls: list[tuple[str, str]], + *, + concurrency: int = DEFAULT_CONCURRENCY, +) -> list[SectionAudit]: + semaphore = asyncio.Semaphore(concurrency) + async with httpx.AsyncClient( + timeout=HTTP_TIMEOUT, + headers={"User-Agent": f"king-scrape-audit/{_scraper_version()}"}, + ) as client: + tasks = [_check_url(client, semaphore, url, title) for url, title in section_urls] + return list(await asyncio.gather(*tasks)) + + +async def _discover_fresh_urls(base_url: str) -> list[str]: + # Lazy import so audit doesn't pull in scraper-provider deps when --no-discover. + from king_context.scraper.config import load_config + from scraper_providers import get_discovery_provider, resolve_provider_name + + config = load_config() + provider_name = resolve_provider_name( + "discover", + explicit=config.scrape_discover_provider or config.scrape_provider, + ) + provider = get_discovery_provider(provider_name, config) + return list(await provider.discover_urls(base_url)) + + +def _diff_urls(corpus_urls: Iterable[str], fresh_urls: Iterable[str]) -> tuple[list[str], list[str]]: + """Diff two URL lists by canonical form, returning ``(new, orphan)``. + + Reports surface the *original* form from each side so contributors see what + they have, not a rewritten version they never indexed. + """ + corpus_by_canon: dict[str, str] = {} + for url in corpus_urls: + corpus_by_canon.setdefault(_canonicalize(url), url) + fresh_by_canon: dict[str, str] = {} + for url in fresh_urls: + fresh_by_canon.setdefault(_canonicalize(url), url) + + new = sorted(fresh_by_canon[c] for c in fresh_by_canon if c not in corpus_by_canon) + orphan = sorted(corpus_by_canon[c] for c in corpus_by_canon if c not in fresh_by_canon) + return new, orphan + + +async def audit_corpus( + corpus_path: Path, + *, + do_discover: bool = True, + concurrency: int = DEFAULT_CONCURRENCY, +) -> CorpusAudit: + corpus = _load_corpus(corpus_path) + section_urls = _unique_section_urls(corpus) + sections = await _check_section_urls(section_urls, concurrency=concurrency) + + result = CorpusAudit( + name=corpus.get("name", corpus_path.stem), + base_url=corpus.get("base_url", ""), + audited_at=datetime.now(timezone.utc).isoformat(), + corpus_path=str(corpus_path), + sections=sections, + ) + + if not do_discover: + result.discovery_skipped = True + return result + + if not result.base_url: + result.discovery_skipped = True + result.discovery_error = "corpus has no base_url" + return result + + try: + fresh = await _discover_fresh_urls(result.base_url) + result.new_urls, result.orphan_urls = _diff_urls( + (s.url for s in sections), fresh + ) + except (httpx.RequestError, RuntimeError, ImportError, OSError) as exc: + # Narrow catch: provider call may surface any of these. A bug-class + # exception (TypeError, AttributeError, etc.) deliberately propagates so + # broken code doesn't masquerade as a "discovery skipped" warning. + result.discovery_skipped = True + result.discovery_error = f"{type(exc).__name__}: {exc}" + + return result + + +def render_report(audit: CorpusAudit) -> str: + counts = audit.counts() + lines: list[str] = [ + f"# Audit -- {audit.name}", + "", + f"- **Audited at:** {audit.audited_at}", + f"- **Corpus:** `{audit.corpus_path}`", + f"- **Base URL:** {audit.base_url or '(none)'}", + f"- **Sections checked:** {len(audit.sections)}", + "", + "## Summary", + "", + f"- fresh: {counts['fresh']}", + f"- moved: {counts['moved']}", + f"- broken: {counts['broken']}", + f"- throttled: {counts['throttled']}", + f"- auth_required: {counts['auth_required']}", + f"- unreachable: {counts['unreachable']}", + ] + if audit.discovery_skipped: + reason = audit.discovery_error or "discovery skipped by flag" + lines.append(f"- discovery: skipped ({reason})") + else: + lines.append(f"- new upstream URLs: {len(audit.new_urls)}") + lines.append(f"- orphan corpus URLs: {len(audit.orphan_urls)}") + + by_status: dict[str, list[SectionAudit]] = {} + for section in audit.sections: + by_status.setdefault(section.status, []).append(section) + + for status in ("broken", "moved", "throttled", "auth_required", "unreachable"): + items = by_status.get(status, []) + if not items: + continue + lines += ["", f"## {status.replace('_', ' ').capitalize()} ({len(items)})", ""] + for s in items: + extras: list[str] = [] + if s.status_code is not None: + extras.append(f"HTTP {s.status_code}") + if s.final_url: + extras.append(f"-> {s.final_url}") + if s.error: + extras.append(s.error) + suffix = f" - {' | '.join(extras)}" if extras else "" + title = f" - {s.title}" if s.title else "" + lines.append(f"- [{s.url}]({s.url}){title}{suffix}") + + if not audit.discovery_skipped: + if audit.new_urls: + lines += ["", f"## New upstream URLs ({len(audit.new_urls)})", ""] + for url in audit.new_urls: + lines.append(f"- {url}") + if audit.orphan_urls: + lines += ["", f"## Orphan corpus URLs ({len(audit.orphan_urls)})", ""] + for url in audit.orphan_urls: + lines.append(f"- {url}") + + return "\n".join(lines) + "\n" + + +_SAFE_NAME_CHARS = re.compile(r"[^A-Za-z0-9._-]+") + + +def _sanitize_filename_component(value: str) -> str: + safe = _SAFE_NAME_CHARS.sub("-", value).strip("-") + return safe or "audit" + + +def _report_path(audit: CorpusAudit, report_dir: Path) -> Path: + stamp = audit.audited_at.replace(":", "").replace("-", "").replace(".", "") + name = _sanitize_filename_component(audit.name) + return report_dir / f"{name}-{stamp}.md" + + +def write_report(audit: CorpusAudit, report_dir: Path) -> Path: + report_dir.mkdir(parents=True, exist_ok=True) + path = _report_path(audit, report_dir) + path.write_text(render_report(audit), encoding="utf-8") + return path + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="king-scrape audit", + description="Audit an indexed corpus for broken, moved, throttled, or stale URLs.", + ) + parser.add_argument("name", help="Doc name (matches data/.json)") + parser.add_argument( + "--no-discover", + action="store_true", + help="Skip the upstream discover diff (faster, no provider key required).", + ) + parser.add_argument( + "--concurrency", + type=int, + default=DEFAULT_CONCURRENCY, + help=f"Max concurrent URL probes (default: {DEFAULT_CONCURRENCY}).", + ) + parser.add_argument( + "--report-dir", + type=Path, + default=PROJECT_ROOT / ".king-context" / "audit", + help="Where to write the Markdown report.", + ) + return parser + + +def _print_summary(audit: CorpusAudit, report_path: Path) -> None: + counts = audit.counts() + parts = [ + f"{counts['fresh']} fresh", + f"{counts['moved']} moved", + f"{counts['broken']} broken", + ] + if counts["throttled"]: + parts.append(f"{counts['throttled']} throttled") + if counts["auth_required"]: + parts.append(f"{counts['auth_required']} auth_required") + if counts["unreachable"]: + parts.append(f"{counts['unreachable']} unreachable") + summary = ", ".join(parts) + print(f"audit {audit.name}: {summary}", end="") + if not audit.discovery_skipped: + print(f" | +{len(audit.new_urls)} / -{len(audit.orphan_urls)}", end="") + print(f" | report: {report_path}") + + +def audit_main(argv: list[str]) -> int: + args = _build_parser().parse_args(argv) + try: + corpus_path = find_corpus(args.name) + except FileNotFoundError as exc: + print(str(exc), file=sys.stderr) + return 1 + + if args.concurrency < 1: + print("error: --concurrency must be >= 1", file=sys.stderr) + return 1 + + try: + audit = asyncio.run( + audit_corpus( + corpus_path, + do_discover=not args.no_discover, + concurrency=args.concurrency, + ) + ) + except ValueError as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + + report_path = write_report(audit, args.report_dir) + _print_summary(audit, report_path) + + counts = audit.counts() + if counts["broken"] > 0: + return 2 + return 0 diff --git a/src/king_context/scraper/cli.py b/src/king_context/scraper/cli.py index 5b04c3d..964d5f0 100644 --- a/src/king_context/scraper/cli.py +++ b/src/king_context/scraper/cli.py @@ -256,7 +256,10 @@ async def run_pipeline(args: argparse.Namespace, config: ScraperConfig) -> None: def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="king-scrape", - description="Scrape and index documentation for King Context", + description=( + "Scrape and index documentation for King Context. " + "Use `king-scrape audit ` to check an existing corpus for drift." + ), ) parser.add_argument("url", help="Base URL of the documentation to scrape") parser.add_argument( @@ -347,6 +350,10 @@ def _build_parser() -> argparse.ArgumentParser: def main() -> None: + if len(sys.argv) > 1 and sys.argv[1] == "audit": + from king_context.scraper.audit import audit_main + sys.exit(audit_main(sys.argv[2:])) + parser = _build_parser() args = parser.parse_args() config = load_config( diff --git a/tests/test_scraper/test_audit.py b/tests/test_scraper/test_audit.py new file mode 100644 index 0000000..ccca204 --- /dev/null +++ b/tests/test_scraper/test_audit.py @@ -0,0 +1,369 @@ +import asyncio +import json +from pathlib import Path +from unittest.mock import patch + +import httpx +import pytest + +from king_context.scraper import audit + + +def _make_corpus(tmp_path: Path, sections: list[dict] | None = None) -> Path: + corpus = { + "name": "demo", + "display_name": "Demo", + "version": "v1", + "base_url": "https://docs.example.com", + "sections": sections + if sections is not None + else [ + {"title": "First", "url": "https://docs.example.com/a"}, + {"title": "Second", "url": "https://docs.example.com/b"}, + {"title": "Third", "url": "https://docs.example.com/c"}, + ], + } + path = tmp_path / "demo.json" + path.write_text(json.dumps(corpus)) + return path + + +def _make_handler(by_url: dict[str, httpx.Response]): + def handler(request: httpx.Request) -> httpx.Response: + response = by_url.get(str(request.url)) + if response is None: + return httpx.Response(500) + return response + return handler + + +def test_unique_section_urls_dedupes_preserving_order(): + corpus = { + "sections": [ + {"url": "https://x/a", "title": "A"}, + {"url": "https://x/b", "title": "B"}, + {"url": "https://x/a", "title": "A again"}, # duplicate + {"title": "no-url"}, # skipped + ] + } + out = audit._unique_section_urls(corpus) + assert out == [("https://x/a", "A"), ("https://x/b", "B")] + + +def test_unique_section_urls_dedupes_fragments_and_slashes(): + corpus = { + "sections": [ + {"url": "https://docs.example.com/page", "title": "Page"}, + {"url": "https://docs.example.com/page/", "title": "Trailing slash"}, + {"url": "https://docs.example.com/page#install", "title": "Fragment"}, + {"url": "https://DOCS.EXAMPLE.COM/page", "title": "Caps"}, + {"url": "https://docs.example.com/other", "title": "Other"}, + ] + } + out = audit._unique_section_urls(corpus) + urls = [u for u, _ in out] + assert urls == ["https://docs.example.com/page", "https://docs.example.com/other"] + + +def test_canonicalize_strips_fragment_and_normalises_path(): + assert audit._canonicalize("https://x/A?q=1#frag") == "https://x/A?q=1" + assert audit._canonicalize("https://x/a/") == "https://x/a" + assert audit._canonicalize("HTTPS://X.COM/A") == "https://x.com/A" + assert audit._canonicalize("https://x.com/") == "https://x.com/" + + +def test_diff_urls(): + new, orphan = audit._diff_urls( + ["https://x/a", "https://x/b", "https://x/c"], + ["https://x/b", "https://x/c", "https://x/d"], + ) + assert new == ["https://x/d"] + assert orphan == ["https://x/a"] + + +def test_diff_urls_canonicalizes_before_comparing(): + new, orphan = audit._diff_urls( + ["https://x/a", "https://x/b/"], + ["https://x/b#anchor", "https://x/c"], + ) + # /b/ in corpus matches /b#anchor in fresh -> not new, not orphan + # /a only in corpus -> orphan; /c only in fresh -> new + assert new == ["https://x/c"] + assert orphan == ["https://x/a"] + + +def test_audit_corpus_marks_status_per_url(tmp_path: Path): + corpus_path = _make_corpus(tmp_path) + by_url = { + "https://docs.example.com/a": httpx.Response(200), + "https://docs.example.com/b": httpx.Response( + 301, headers={"location": "https://docs.example.com/b2"} + ), + "https://docs.example.com/c": httpx.Response(404), + } + transport = httpx.MockTransport(_make_handler(by_url)) + + async def fake_check(section_urls, *, concurrency=10): + async with httpx.AsyncClient(transport=transport) as client: + sem = asyncio.Semaphore(10) + return list(await asyncio.gather(*[ + audit._check_url(client, sem, url, title) + for url, title in section_urls + ])) + + with patch.object(audit, "_check_section_urls", side_effect=fake_check): + result = asyncio.run(audit.audit_corpus(corpus_path, do_discover=False)) + + assert result.discovery_skipped is True + by_status = {s.url: s.status for s in result.sections} + assert by_status["https://docs.example.com/a"] == "fresh" + assert by_status["https://docs.example.com/b"] == "moved" + assert by_status["https://docs.example.com/c"] == "broken" + + moved = next(s for s in result.sections if s.status == "moved") + assert moved.final_url == "https://docs.example.com/b2" + + +def test_audit_corpus_falls_back_to_get_when_head_unsupported(tmp_path: Path): + corpus_path = _make_corpus( + tmp_path, sections=[{"title": "A", "url": "https://docs.example.com/a"}] + ) + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "HEAD": + return httpx.Response(405) + return httpx.Response(200) + + transport = httpx.MockTransport(handler) + + async def fake_check(section_urls, *, concurrency=10): + async with httpx.AsyncClient(transport=transport) as client: + sem = asyncio.Semaphore(10) + return list(await asyncio.gather(*[ + audit._check_url(client, sem, url, title) + for url, title in section_urls + ])) + + with patch.object(audit, "_check_section_urls", side_effect=fake_check): + result = asyncio.run(audit.audit_corpus(corpus_path, do_discover=False)) + + assert result.sections[0].status == "fresh" + assert result.sections[0].status_code == 200 + + +def test_audit_corpus_runs_discovery_diff(tmp_path: Path): + corpus_path = _make_corpus(tmp_path) + + async def fake_check(section_urls, *, concurrency=10): + return [ + audit.SectionAudit(url=url, title=title, status="fresh", status_code=200) + for url, title in section_urls + ] + + async def fake_discover(base_url): + return [ + "https://docs.example.com/a", # in corpus + "https://docs.example.com/b", # in corpus + # /c missing → orphan + "https://docs.example.com/d", # new upstream URL + ] + + with patch.object(audit, "_check_section_urls", side_effect=fake_check), \ + patch.object(audit, "_discover_fresh_urls", side_effect=fake_discover): + result = asyncio.run(audit.audit_corpus(corpus_path, do_discover=True)) + + assert result.discovery_skipped is False + assert result.new_urls == ["https://docs.example.com/d"] + assert result.orphan_urls == ["https://docs.example.com/c"] + + +def test_audit_corpus_records_discovery_failure(tmp_path: Path): + corpus_path = _make_corpus(tmp_path) + + async def fake_check(section_urls, *, concurrency=10): + return [ + audit.SectionAudit(url=url, title=title, status="fresh") + for url, title in section_urls + ] + + async def fake_discover(base_url): + raise RuntimeError("provider unavailable") + + with patch.object(audit, "_check_section_urls", side_effect=fake_check), \ + patch.object(audit, "_discover_fresh_urls", side_effect=fake_discover): + result = asyncio.run(audit.audit_corpus(corpus_path, do_discover=True)) + + assert result.discovery_skipped is True + assert "provider unavailable" in (result.discovery_error or "") + + +def test_render_report_contains_summary_and_breakdowns(): + a = audit.CorpusAudit( + name="demo", + base_url="https://docs.example.com", + audited_at="2026-05-07T12:00:00+00:00", + corpus_path="/data/demo.json", + sections=[ + audit.SectionAudit(url="https://x/a", title="A", status="fresh", status_code=200), + audit.SectionAudit(url="https://x/b", title="B", status="broken", status_code=404), + audit.SectionAudit( + url="https://x/c", + title="C", + status="moved", + status_code=301, + final_url="https://x/c2", + ), + ], + new_urls=["https://x/d"], + orphan_urls=[], + ) + report = audit.render_report(a) + + assert "# Audit -- demo" in report + assert "fresh: 1" in report + assert "broken: 1" in report + assert "moved: 1" in report + assert "https://x/b" in report + assert "https://x/c2" in report + assert "https://x/d" in report + + +def test_audit_main_returns_2_when_broken_urls_present(tmp_path: Path): + corpus_path = _make_corpus(tmp_path) + audit_dir = tmp_path / "audit-out" + + async def fake_check(section_urls, *, concurrency=10): + return [ + audit.SectionAudit(url=url, title=title, status="broken", status_code=404) + for url, title in section_urls + ] + + with patch.object(audit, "find_corpus", return_value=corpus_path), \ + patch.object(audit, "_check_section_urls", side_effect=fake_check): + rc = audit.audit_main(["demo", "--no-discover", "--report-dir", str(audit_dir)]) + + assert rc == 2 + reports = list(audit_dir.glob("demo-*.md")) + assert len(reports) == 1 + + +def test_audit_main_returns_0_when_clean(tmp_path: Path): + corpus_path = _make_corpus(tmp_path) + audit_dir = tmp_path / "audit-out" + + async def fake_check(section_urls, *, concurrency=10): + return [ + audit.SectionAudit(url=url, title=title, status="fresh", status_code=200) + for url, title in section_urls + ] + + with patch.object(audit, "find_corpus", return_value=corpus_path), \ + patch.object(audit, "_check_section_urls", side_effect=fake_check): + rc = audit.audit_main(["demo", "--no-discover", "--report-dir", str(audit_dir)]) + + assert rc == 0 + + +def test_audit_main_returns_1_when_corpus_missing(tmp_path: Path, capsys): + rc = audit.audit_main(["does-not-exist", "--no-discover"]) + assert rc == 1 + err = capsys.readouterr().err + assert "not found" in err.lower() + + +def test_audit_main_returns_1_on_malformed_json(tmp_path: Path, capsys): + bad = tmp_path / "bad.json" + bad.write_text("{not valid") + audit_dir = tmp_path / "out" + + with patch.object(audit, "find_corpus", return_value=bad): + rc = audit.audit_main(["bad", "--no-discover", "--report-dir", str(audit_dir)]) + + assert rc == 1 + err = capsys.readouterr().err + assert "not valid json" in err.lower() + + +def test_classify_maps_auth_and_throttle_codes(): + assert audit._classify(httpx.Response(401))[0] == "auth_required" + assert audit._classify(httpx.Response(403))[0] == "auth_required" + assert audit._classify(httpx.Response(429))[0] == "throttled" + assert audit._classify(httpx.Response(404))[0] == "broken" + assert audit._classify(httpx.Response(410))[0] == "broken" + assert audit._classify(httpx.Response(500))[0] == "unreachable" + + +def test_classify_marks_followed_redirects_as_moved(): + final = httpx.Response(200, request=httpx.Request("HEAD", "https://x/final")) + final.history = [httpx.Response(301)] + status, final_url = audit._classify(final) + assert status == "moved" + assert final_url == "https://x/final" + + +def test_check_url_retries_on_429_then_resolves(tmp_path: Path): + corpus_path = _make_corpus( + tmp_path, sections=[{"title": "A", "url": "https://docs.example.com/a"}] + ) + + state = {"calls": 0} + + def handler(request: httpx.Request) -> httpx.Response: + state["calls"] += 1 + if state["calls"] == 1: + return httpx.Response(429, headers={"retry-after": "0"}) + return httpx.Response(200) + + transport = httpx.MockTransport(handler) + + async def fake_check(section_urls, *, concurrency=10): + async with httpx.AsyncClient(transport=transport) as client: + sem = asyncio.Semaphore(concurrency) + return list(await asyncio.gather(*[ + audit._check_url(client, sem, url, title) + for url, title in section_urls + ])) + + with patch.object(audit, "_check_section_urls", side_effect=fake_check): + result = asyncio.run(audit.audit_corpus(corpus_path, do_discover=False)) + + assert state["calls"] == 2 + assert result.sections[0].status == "fresh" + + +def test_render_report_has_no_emoji(): + result = audit.CorpusAudit( + name="demo", + base_url="https://x", + audited_at="2026-05-07T12:00:00+00:00", + corpus_path="/data/demo.json", + sections=[audit.SectionAudit(url="https://x/a", title="A", status="fresh")], + ) + report = audit.render_report(result) + # Quick sanity: each character in the report is in the ASCII range. + assert all(ord(ch) < 128 for ch in report), "report should be ASCII-only" + + +def test_audit_corpus_propagates_unexpected_errors(tmp_path: Path): + """Bug class exceptions should NOT be swallowed as 'discovery skipped'.""" + corpus_path = _make_corpus(tmp_path) + + async def fake_check(section_urls, *, concurrency=10): + return [ + audit.SectionAudit(url=url, title=title, status="fresh") + for url, title in section_urls + ] + + async def fake_discover(base_url): + raise TypeError("genuine bug, do not swallow") + + with patch.object(audit, "_check_section_urls", side_effect=fake_check), \ + patch.object(audit, "_discover_fresh_urls", side_effect=fake_discover): + with pytest.raises(TypeError): + asyncio.run(audit.audit_corpus(corpus_path, do_discover=True)) + + +def test_sanitize_filename_component_strips_unsafe_chars(): + assert audit._sanitize_filename_component("ok-name_1.0") == "ok-name_1.0" + assert audit._sanitize_filename_component("a/b\\c:d") == "a-b-c-d" + assert audit._sanitize_filename_component("///") == "audit" From 0f01e0121f39e94b8513667a82786b66defb4bfd Mon Sep 17 00:00:00 2001 From: Nelson Mfinda Date: Fri, 8 May 2026 17:34:33 +0100 Subject: [PATCH 2/3] fix(audit): classify redirect chains by final status; document subcommand --- .../adr/0013-corpus-drift-audit-subcommand.md | 16 +- CHANGELOG.md | 22 +-- docs/CLI_GUIDE.md | 50 ++++++ src/king_context/scraper/audit.py | 72 ++++++--- tests/test_scraper/test_audit.py | 152 +++++++++++++++++- 5 files changed, 274 insertions(+), 38 deletions(-) diff --git a/.king-context/adr/0013-corpus-drift-audit-subcommand.md b/.king-context/adr/0013-corpus-drift-audit-subcommand.md index bc26dd1..e231e38 100644 --- a/.king-context/adr/0013-corpus-drift-audit-subcommand.md +++ b/.king-context/adr/0013-corpus-drift-audit-subcommand.md @@ -1,6 +1,6 @@ --- id: ADR-0013 -title: Read-only drift audit for indexed corpora via `king-scrape audit` +title: Read only drift audit for indexed corpora via `king-scrape audit` status: accepted date: 2026-05-07 areas: @@ -28,28 +28,28 @@ tags: --- -# ADR-0013: Read-only drift audit for indexed corpora via `king-scrape audit` +# ADR-0013: Read only drift audit for indexed corpora via `king-scrape audit` ## Context -A corpus committed under `data/.json` is a snapshot. The upstream docs site keeps moving pages get renamed, deprecated, removed; new sections appear. The MCP server happily serves whatever was indexed, so a contributor has no easy way to ask *"is this corpus still aligned with the live docs, or do I need to re-scrape?"* without paying full pipeline cost (Firecrawl fetch + OpenRouter enrichment) just to find out. +A corpus committed under `data/.json` is a snapshot. The upstream docs site keeps moving: pages get renamed, deprecated, removed; new sections appear. The MCP server happily serves whatever was indexed, so a contributor has no easy way to ask *"is this corpus still aligned with the live docs, or do I need to rescrape?"* without paying full pipeline cost (Firecrawl fetch + OpenRouter enrichment) just to find out. ADR-0012 established content hash provenance through the pipeline so that, eventually, an `--update` flow can answer that question precisely. But: - Today, many indexed corpora pre-date ADR-0012 and have no `_meta.content_hash`. -- A precise content-diff requires re-fetching every page, which is the expensive part the contributor is trying to avoid before deciding to refresh. +- A precise content diff requires refetching every page, which is the expensive part the contributor is trying to avoid before deciding to refresh. - The most actionable signal *which URLs are flat out broken* is cheap to obtain and useful on its own. There is a real gap between "do nothing" and "run the full incremental refresh that ADR-0014 will introduce." That gap is what this ADR fills. ## Decision -Introduce a `king-scrape audit ` subcommand whose only job is to read an indexed corpus and report drift signals as a Markdown file. The audit is **read-only**: it never mutates `data/.json`, never writes to the database, never spends LLM credits. It is safe to run on a schedule. +Introduce a `king-scrape audit ` subcommand whose only job is to read an indexed corpus and report drift signals as a Markdown file. The audit is **read only**: it never mutates `data/.json`, never writes to the database, never spends LLM credits. It is safe to run on a schedule. The audit runs two passes: 1. **URL health (always).** Walks every unique section URL in the corpus and issues an HTTP `HEAD` (falling back to `GET` when the server returns 405/501). Results are mapped to one of: `fresh` (2xx), `moved` (3xx, with `Location` captured), `broken` (404/410), or `unreachable` (timeout / network error / other 5xx). No provider key is required for this pass, it uses `httpx` directly with bounded concurrency (10). -2. **Discovery diff (optional, default-on).** Calls the configured `DiscoveryProvider` (Firecrawl, Crawl4AI) to re-map the upstream site, then diffs the URL set against the corpus. Reports `new_urls` (in fresh discover, not in corpus) and `orphan_urls` (in corpus, not in fresh discover). `--no-discover` skips this pass for environments without provider keys. +2. **Discovery diff (optional, on by default).** Calls the configured `DiscoveryProvider` (Firecrawl, Crawl4AI) to remap the upstream site, then diffs the URL set against the corpus. Reports `new_urls` (in fresh discover, not in corpus) and `orphan_urls` (in corpus, not in fresh discover). `--no-discover` skips this pass for environments without provider keys. A new module `src/king_context/scraper/audit.py` owns the dataclasses (`SectionAudit`, `CorpusAudit`), the async URL probe, the discovery diff, and a Markdown renderer (`render_report`) plus writer (`write_report`). The CLI dispatcher in `cli.py` delegates to `audit_main(argv)` when invoked as `king-scrape audit ...`. The existing `king-scrape ` flow is unchanged `audit` is a sibling subcommand, not a refactor of the existing parser. @@ -61,7 +61,7 @@ Reports land at `.king-context/audit/-.md`. The exit code i **Standalone console script (`king-audit`).** Possible, but multiplies the bin surface for a feature that conceptually lives next to the scrape pipeline. The single dispatcher in `king-scrape` matches the way Git, Cargo, and `kctx` already organise their subcommands. -**Full content-hash diff in this PR.** Considered and deferred. ADR-0012 introduced the per-section `_meta.content_hash`; using it here would force this PR to depend on ADR-0012 landing first, and would still be incomplete for pre-ADR-0012 corpora. Splitting URL health into its own change keeps each PR focused and lets contributors get the broken URL signal even on legacy corpora. ADR-0014 (incremental refresh) is the natural home for the content-hash diff because it has to compute fresh page hashes anyway as part of deciding what to re-enrich. +**Full content-hash diff in this PR.** Considered and deferred. ADR-0012 introduced the per section `_meta.content_hash`; using it here would force this PR to depend on ADR-0012 landing first, and would still be incomplete for pre ADR-0012 corpora. Splitting URL health into its own change keeps each PR focused and lets contributors get the broken URL signal even on legacy corpora. ADR-0014 (incremental refresh) is the natural home for the content-hash diff because it has to compute fresh page hashes anyway as part of deciding what to reenrich. **Mutate `data/.json` with status fields.** Rejected. Keeping the corpus pristine and writing audits to a separate location preserves the simple "read only fact" model of the indexed JSON. A contributor inspecting `data/*.json` should see only what `seed_data` reads, never audit metadata. @@ -69,7 +69,7 @@ Reports land at `.king-context/audit/-.md`. The exit code i Contributors gain a one line, read only command to check whether a corpus is still aligned with upstream. The output is plain Markdown so it diffs nicely, can be committed if desired, or pasted into an issue. -Because the audit never spends LLM credits and never mutates state, it is safe to attach to a CI cron, a pre merge check on corpus PRs, or a manual sanity run before considering a re-scrape. +Because the audit never spends LLM credits and never mutates state, it is safe to attach to a CI cron, a pre merge check on corpus PRs, or a manual sanity run before considering a rescrape. The dispatcher pattern (`if argv[1] == "audit": ...`) creates a precedent that ADR-0014's `update` subcommand will follow without further refactor. If the project ever grows enough subcommands to warrant `argparse` subparsers, the migration is a local change inside `cli.main` with no impact on the audit module. diff --git a/CHANGELOG.md b/CHANGELOG.md index 27edbaa..9e28ebe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,15 +9,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- `king-scrape audit ` subcommand (ADR-0013). Walks the URLs of an - indexed corpus and reports which are reachable, moved (3xx with the - redirect target), broken (404/410), or unreachable. Optionally - re-runs discovery against the upstream and reports URLs added or - removed since the corpus was indexed (`--no-discover` to skip). - Read-only — never mutates the corpus file or the database. No LLM - cost. Markdown report lands at `.king-context/audit/-.md`. - Exit code is `0` on clean, `2` when at least one section is broken, - so it can gate a CI job. +- `king-scrape audit ` subcommand (ADR-0013). Walks the URLs of + an indexed corpus and classifies each by its final HTTP status: + `fresh` (2xx), `moved` (redirect chain ending in 2xx, with the final + URL captured), `broken` (final 404/410, including chains that + redirect into a dead page), `throttled` (429, retried once with + `Retry-After`), `auth_required` (401/403), or `unreachable`. URLs + are canonicalised (fragment / trailing slash / host case) before + dedupe and discovery diff. Optionally reruns discovery against the + upstream and reports URLs added or removed since the corpus was + indexed (`--no-discover` to skip). Read only, never mutates the + corpus file or the database. No LLM cost. Markdown report lands at + `.king-context/audit/-.md`. Exit code is `0` on clean, + `2` when at least one section is broken, so it can gate a CI job. ## [0.4.0] - 2026-05-06 diff --git a/docs/CLI_GUIDE.md b/docs/CLI_GUIDE.md index deed49f..d7ffc3b 100644 --- a/docs/CLI_GUIDE.md +++ b/docs/CLI_GUIDE.md @@ -14,6 +14,8 @@ King Context installs three command-line tools: - `kctx`: search, read, index, and validate local retrieval stores. - `king-scrape`: scrape a documentation site and export indexed sections. + Also exposes `king-scrape audit ` to check an indexed corpus + for broken, moved, or drifted URLs. - `king-research`: build and index a research corpus for a topic. The `kctx` command searches two content stores by default: @@ -483,6 +485,54 @@ Firecrawl is selected (default) but no API key is set. Add `FIRECRAWL_API_KEY=...` to `.env`, or switch to Crawl4AI with `--provider=crawl4ai`. +## Audit a corpus for drift + +Use `king-scrape audit ` to check whether an indexed corpus is still +aligned with its upstream source. The audit is read only: it never mutates +`data/.json` or the database, and the URL health pass needs no +provider key, so it is safe to run on a CI cron. + +```bash +king-scrape audit elevenlabs-api +``` + +Each section URL is classified by its **final** HTTP status: + +| Status | Meaning | +| --- | --- | +| `fresh` | 2xx | +| `moved` | redirect chain ending in 2xx; the final URL is captured | +| `broken` | 404 / 410, including chains that redirect into a dead page | +| `throttled` | 429; respects `Retry-After` (capped 30s), retries once | +| `auth_required` | 401 / 403 | +| `unreachable` | timeout / network error / other 5xx | + +URLs are canonicalised (fragment, trailing slash, host case stripped) before +dedupe and before the discovery diff so cosmetic variations do not inflate +the report. + +A Markdown report lands at `.king-context/audit/-.md` +(timestamp is UTC, microsecond precision, suffixed `Z`). +Exit code is `0` when no broken URLs are found, `2` when at least one is, so +the audit can gate a CI job. The URL health pass needs no provider key, so +the keyless form is the right shape for CI: + +```bash +# CI friendly: no provider key required, no upstream discovery diff +king-scrape audit my-corpus --no-discover || echo "drift detected, see report" +``` + +| Flag | Default | Description | +| --- | --- | --- | +| `--no-discover` | off | Skip the upstream discovery diff (faster, no provider key required). | +| `--concurrency N` | 10 | Max concurrent URL probes. | +| `--report-dir DIR` | `.king-context/audit/` | Where to write the report. | + +The optional discovery diff calls the configured `DiscoveryProvider` +(Firecrawl or Crawl4AI) to remap the upstream and lists URLs added or +removed since the corpus was indexed. Use `--no-discover` to skip the +provider step entirely. + ## LLM Provider Configuration `king-scrape` and `king-research` use OpenRouter by default: diff --git a/src/king_context/scraper/audit.py b/src/king_context/scraper/audit.py index 2effc35..70c1664 100644 --- a/src/king_context/scraper/audit.py +++ b/src/king_context/scraper/audit.py @@ -1,13 +1,13 @@ -"""Drift / staleness audit for an indexed corpus. +"""Drift and staleness audit for an indexed corpus. Walks the URLs of an existing ``data/.json`` corpus and reports which ones are still reachable, which moved (redirects), which are broken (404), and optionally which URLs the upstream docs site has gained or lost -since the corpus was indexed. Pure read-only: never mutates the corpus +since the corpus was indexed. Pure read only: never mutates the corpus file or the database. Designed to run cheaply (HEAD requests, no LLM calls, no provider keys -required for the URL health pass) so contributors can re-run it on a +required for the URL health pass) so contributors can rerun it on a schedule without spending OpenRouter credits. """ @@ -20,6 +20,7 @@ import sys from dataclasses import dataclass, field from datetime import datetime, timezone +from email.utils import parsedate_to_datetime from importlib.metadata import PackageNotFoundError, version as _pkg_version from pathlib import Path from typing import Iterable @@ -118,7 +119,7 @@ def _load_corpus(corpus_path: Path) -> dict: def _unique_section_urls(corpus: dict) -> list[tuple[str, str]]: - """Return ``[(url, title), ...]`` deduped by canonical URL, preserving first-seen order.""" + """Return ``[(url, title), ...]`` deduped by canonical URL, preserving first seen order.""" seen: set[str] = set() out: list[tuple[str, str]] = [] for section in corpus.get("sections", []): @@ -136,32 +137,50 @@ def _unique_section_urls(corpus: dict) -> list[tuple[str, str]]: def _classify(response: httpx.Response) -> tuple[str, str | None]: """Map an httpx response to ``(status, final_url)``. - Caller passes a response from ``follow_redirects=True``, so a populated - ``response.history`` means we landed on the chain endpoint after one or - more 3xx hops; ``response.url`` is then the final URL. + Classification is keyed off the *final* status code, not the presence of + redirects. A chain like ``301 -> 308 -> 404`` is broken, not moved. The + audit exists to surface that, so reporting it as moved would defeat the + point. The final URL is still captured for context whenever the chain + has any redirect history, regardless of the final status. """ - if response.history: - return "moved", str(response.url) + final_url = str(response.url) if response.history else None code = response.status_code if 200 <= code < 300: - return "fresh", None + return ("moved" if response.history else "fresh"), final_url if code in (401, 403): - return "auth_required", None + return "auth_required", final_url if code == 429: - return "throttled", None + return "throttled", final_url if code in (404, 410): - return "broken", None - return "unreachable", None + return "broken", final_url + return "unreachable", final_url def _retry_after_seconds(response: httpx.Response) -> float: + """Parse a ``Retry-After`` header into a sleep duration, capped to a sane max. + + The header may be either a delta in seconds (RFC 7231) or an HTTP date. + Unrecognised values (or a date in the past) fall back to one second so the + audit always makes forward progress. + """ raw = response.headers.get("retry-after") if not raw: return 1.0 + raw = raw.strip() try: return min(THROTTLE_RETRY_CAP_SECONDS, max(1.0, float(raw))) except ValueError: + pass + try: + target = parsedate_to_datetime(raw) + except (TypeError, ValueError): + return 1.0 + if target is None: return 1.0 + if target.tzinfo is None: + target = target.replace(tzinfo=timezone.utc) + delta = (target - datetime.now(timezone.utc)).total_seconds() + return min(THROTTLE_RETRY_CAP_SECONDS, max(1.0, delta)) async def _check_url( @@ -279,9 +298,9 @@ async def audit_corpus( (s.url for s in sections), fresh ) except (httpx.RequestError, RuntimeError, ImportError, OSError) as exc: - # Narrow catch: provider call may surface any of these. A bug-class - # exception (TypeError, AttributeError, etc.) deliberately propagates so - # broken code doesn't masquerade as a "discovery skipped" warning. + # Narrow catch: provider call may surface any of these. A bug class + # exception (TypeError, AttributeError, etc.) deliberately propagates + # so broken code does not masquerade as a "discovery skipped" warning. result.discovery_skipped = True result.discovery_error = f"{type(exc).__name__}: {exc}" @@ -291,7 +310,7 @@ async def audit_corpus( def render_report(audit: CorpusAudit) -> str: counts = audit.counts() lines: list[str] = [ - f"# Audit -- {audit.name}", + f"# Audit: {audit.name}", "", f"- **Audited at:** {audit.audited_at}", f"- **Corpus:** `{audit.corpus_path}`", @@ -357,7 +376,17 @@ def _sanitize_filename_component(value: str) -> str: def _report_path(audit: CorpusAudit, report_dir: Path) -> Path: - stamp = audit.audited_at.replace(":", "").replace("-", "").replace(".", "") + """Build the report path. Tolerates a malformed or naive ``audited_at``.""" + try: + dt = datetime.fromisoformat(audit.audited_at) + if dt.tzinfo is None: + # A naive timestamp would crash astimezone(); assume UTC and proceed. + dt = dt.replace(tzinfo=timezone.utc) + # Microsecond precision avoids collision when audits run in the same + # second; trailing Z strips the timezone offset for a shorter name. + stamp = dt.astimezone(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ") + except ValueError: + stamp = re.sub(r"[^A-Za-z0-9]", "", audit.audited_at) name = _sanitize_filename_component(audit.name) return report_dir / f"{name}-{stamp}.md" @@ -411,7 +440,10 @@ def _print_summary(audit: CorpusAudit, report_path: Path) -> None: summary = ", ".join(parts) print(f"audit {audit.name}: {summary}", end="") if not audit.discovery_skipped: - print(f" | +{len(audit.new_urls)} / -{len(audit.orphan_urls)}", end="") + print( + f" | new {len(audit.new_urls)} / orphan {len(audit.orphan_urls)}", + end="", + ) print(f" | report: {report_path}") diff --git a/tests/test_scraper/test_audit.py b/tests/test_scraper/test_audit.py index ccca204..ca22002 100644 --- a/tests/test_scraper/test_audit.py +++ b/tests/test_scraper/test_audit.py @@ -1,5 +1,6 @@ import asyncio import json +from datetime import datetime, timezone from pathlib import Path from unittest.mock import patch @@ -99,6 +100,7 @@ def test_audit_corpus_marks_status_per_url(tmp_path: Path): "https://docs.example.com/b": httpx.Response( 301, headers={"location": "https://docs.example.com/b2"} ), + "https://docs.example.com/b2": httpx.Response(200), "https://docs.example.com/c": httpx.Response(404), } transport = httpx.MockTransport(_make_handler(by_url)) @@ -219,7 +221,7 @@ def test_render_report_contains_summary_and_breakdowns(): ) report = audit.render_report(a) - assert "# Audit -- demo" in report + assert "# Audit: demo" in report assert "fresh: 1" in report assert "broken: 1" in report assert "moved: 1" in report @@ -301,6 +303,154 @@ def test_classify_marks_followed_redirects_as_moved(): assert final_url == "https://x/final" +def test_classify_redirect_chain_ending_in_404_is_broken(): + """301 -> 404 must classify as broken (not moved). Final URL preserved for context.""" + final = httpx.Response(404, request=httpx.Request("HEAD", "https://x/dead")) + final.history = [httpx.Response(308)] + status, final_url = audit._classify(final) + assert status == "broken" + assert final_url == "https://x/dead" + + +def test_classify_redirect_chain_ending_in_500_is_unreachable(): + final = httpx.Response(503, request=httpx.Request("HEAD", "https://x/down")) + final.history = [httpx.Response(301)] + status, final_url = audit._classify(final) + assert status == "unreachable" + assert final_url == "https://x/down" + + +def test_classify_redirect_chain_ending_in_401_is_auth_required(): + final = httpx.Response(401, request=httpx.Request("HEAD", "https://x/login")) + final.history = [httpx.Response(302)] + status, final_url = audit._classify(final) + assert status == "auth_required" + assert final_url == "https://x/login" + + +def test_classify_redirect_chain_ending_in_429_is_throttled(): + final = httpx.Response(429, request=httpx.Request("HEAD", "https://x/limited")) + final.history = [httpx.Response(307)] + status, final_url = audit._classify(final) + assert status == "throttled" + assert final_url == "https://x/limited" + + +def test_print_summary_uses_new_orphan_words(tmp_path: Path, capsys): + a = audit.CorpusAudit( + name="x", + base_url="https://x", + audited_at="2026-05-08T00:00:00+00:00", + corpus_path="/x.json", + sections=[audit.SectionAudit(url="https://x/a", title="A", status="fresh")], + new_urls=["https://x/new"], + orphan_urls=[], + ) + audit._print_summary(a, tmp_path / "report.md") + out = capsys.readouterr().out + assert "new 1" in out + assert "orphan 0" in out + assert "+1" not in out + assert "-0" not in out + + +def test_retry_after_parses_seconds(): + response = httpx.Response(429, headers={"retry-after": "5"}) + assert audit._retry_after_seconds(response) == 5.0 + + +def test_retry_after_caps_excessive_seconds(): + response = httpx.Response(429, headers={"retry-after": "9999"}) + assert audit._retry_after_seconds(response) == audit.THROTTLE_RETRY_CAP_SECONDS + + +def test_retry_after_parses_http_date(): + # An HTTP date a few seconds in the future should be honoured (and capped). + future = datetime.now(timezone.utc).replace(microsecond=0) + future = future.replace(year=future.year + 1) + http_date = future.strftime("%a, %d %b %Y %H:%M:%S GMT") + response = httpx.Response(429, headers={"retry-after": http_date}) + delay = audit._retry_after_seconds(response) + assert delay == audit.THROTTLE_RETRY_CAP_SECONDS # capped + + +def test_retry_after_falls_back_on_garbage(): + response = httpx.Response(429, headers={"retry-after": "not-a-date-or-number"}) + assert audit._retry_after_seconds(response) == 1.0 + + +def test_retry_after_past_http_date_falls_back_to_min(): + # A date in the past should not produce a negative or zero sleep. + response = httpx.Response(429, headers={"retry-after": "Wed, 21 Oct 2020 07:28:00 GMT"}) + assert audit._retry_after_seconds(response) == 1.0 + + +def test_report_path_handles_naive_isoformat(tmp_path: Path): + """Naive (no tz) audited_at must not crash; assumed UTC.""" + a = audit.CorpusAudit( + name="demo", + base_url="", + audited_at="2026-05-08T15:42:43.082839", # no offset + corpus_path="/x.json", + ) + path = audit._report_path(a, tmp_path) + assert path.name.endswith("Z.md") + assert "20260508T154243" in path.name + + +def test_report_path_keeps_microsecond_precision(tmp_path: Path): + """Two audits in the same second must not collide on filename.""" + base = audit.CorpusAudit( + name="demo", + base_url="", + audited_at="2026-05-08T15:42:43.082839+00:00", + corpus_path="/x.json", + ) + other = audit.CorpusAudit( + name="demo", + base_url="", + audited_at="2026-05-08T15:42:43.999999+00:00", + corpus_path="/x.json", + ) + p1 = audit._report_path(base, tmp_path) + p2 = audit._report_path(other, tmp_path) + assert p1 != p2 + # Filename ends with `Z`, no offset. + assert p1.name.endswith("Z.md") + assert "+0000" not in p1.name + + +def test_audit_main_returns_2_on_redirect_to_broken(tmp_path: Path): + """Live bug repro from elevenlabs corpus: 308 -> 404 must trigger CI gate.""" + corpus_path = _make_corpus( + tmp_path, sections=[{"title": "X", "url": "https://docs.example.com/old"}] + ) + audit_dir = tmp_path / "out" + + def handler(request: httpx.Request) -> httpx.Response: + if str(request.url) == "https://docs.example.com/old": + return httpx.Response( + 308, headers={"location": "https://docs.example.com/new"} + ) + return httpx.Response(404) + + transport = httpx.MockTransport(handler) + + async def fake_check(section_urls, *, concurrency=10): + async with httpx.AsyncClient(transport=transport) as client: + sem = asyncio.Semaphore(concurrency) + return list(await asyncio.gather(*[ + audit._check_url(client, sem, url, title) + for url, title in section_urls + ])) + + with patch.object(audit, "find_corpus", return_value=corpus_path), \ + patch.object(audit, "_check_section_urls", side_effect=fake_check): + rc = audit.audit_main(["demo", "--no-discover", "--report-dir", str(audit_dir)]) + + assert rc == 2 + + def test_check_url_retries_on_429_then_resolves(tmp_path: Path): corpus_path = _make_corpus( tmp_path, sections=[{"title": "A", "url": "https://docs.example.com/a"}] From e3c14c3fa61d60c6323075f1bf0cce67583a65d6 Mon Sep 17 00:00:00 2001 From: Nelson Mfinda Date: Fri, 8 May 2026 17:46:57 +0100 Subject: [PATCH 3/3] fix(audit): classify redirect chains by final status; document subcommand --- tests/test_scraper/test_audit.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tests/test_scraper/test_audit.py b/tests/test_scraper/test_audit.py index ca22002..e9f1d0f 100644 --- a/tests/test_scraper/test_audit.py +++ b/tests/test_scraper/test_audit.py @@ -350,8 +350,7 @@ def test_print_summary_uses_new_orphan_words(tmp_path: Path, capsys): out = capsys.readouterr().out assert "new 1" in out assert "orphan 0" in out - assert "+1" not in out - assert "-0" not in out + assert "+1 / -0" not in out def test_retry_after_parses_seconds():