diff --git a/Makefile b/Makefile index 58408fb..08860b5 100644 --- a/Makefile +++ b/Makefile @@ -31,7 +31,8 @@ SEARCH_Q ?= computer ingest-grok ingest-grok-media \ ingest-self ingest-git ingest-hg \ verify search stats test docs chain-check chain-check-shards \ - falsify burn bootstrap-crawler test-crawler clean clean-db clean-data help + falsify burn bootstrap-crawler test-crawler crawl-ingest recrawl-check \ + clean clean-db clean-data help all: bootstrap fetch-cur ingest-cur verify stats ## bootstrap → fetch cur → ingest cur → verify → stats @@ -321,6 +322,22 @@ bootstrap-crawler: bootstrap ## install [crawler] extras into the venv test-crawler: bootstrap-crawler ## run only the lifted crawler tests $(VENV)/bin/pytest -q tests/crawler +# Bridge target: BFS a seed URL, ingest discovered pages into a shard, +# capture ETag/Last-Modified for each so recrawl-check can do conditional +# HEAD requests later. URL is required; DEPTH and MAX have safe defaults. +CRAWL_DEPTH ?= 2 +CRAWL_MAX ?= 20 +crawl-ingest: bootstrap-crawler ## crawl URL=https://x.com [DEPTH=2 MAX=20] and ingest into $(DB) + @if [ -z "$(URL)" ]; then echo "usage: make crawl-ingest URL=https://example.com [DEPTH=2 MAX=20]" >&2; exit 2; fi + $(ABORIST) --db $(DB) crawl --seed-url "$(URL)" --depth $(CRAWL_DEPTH) --max-pages $(CRAWL_MAX) --ingest + +# Fast freshness probe: conditional HEAD per doc, classify fresh/stale/gone. +# Send only If-None-Match + If-Modified-Since headers — server returns 304 +# with no body when content unchanged. +RECRAWL_LIMIT ?= 100 +recrawl-check: bootstrap-crawler ## conditional HEAD on ingested docs [DOMAIN=foo.com LIMIT=100] + $(ABORIST) --db $(DB) crawler recrawl-check $(if $(DOMAIN),--domain $(DOMAIN),) --limit $(RECRAWL_LIMIT) + DOT_SRCS := $(wildcard docs/diagrams/*.dot) DOT_PNGS := $(DOT_SRCS:.dot=.png) diff --git a/aborist/cli.py b/aborist/cli.py index 1398825..c9f612f 100644 --- a/aborist/cli.py +++ b/aborist/cli.py @@ -2102,6 +2102,78 @@ def _cmd_mesh_pull(args: argparse.Namespace) -> int: return 0 +def _cmd_crawl(args: argparse.Namespace) -> int: + """BFS-discover same-domain URLs from a seed and optionally ingest. + + Two modes: + + - default: print discovered URLs to stdout (one per line). Compose + with `aborist ingest --source html` if you want to feed them + through the standard ingest path manually. + - ``--ingest``: run the discovery + ingest path in a single shot, + capturing ETag + Last-Modified per page so a future + ``crawler recrawl-check`` can do conditional HEADs. + """ + try: + from aborist.sources.crawler.bridge import crawl_seed, ingest_crawled + except ImportError as e: + print(f"error: {e}", file=sys.stderr) + return 2 + + urls = crawl_seed( + args.seed_url, max_depth=args.depth, max_pages=args.max_pages + ) + if not args.ingest: + for u in urls: + print(u) + return 0 + + conn = connect(args.db) + try: + result = ingest_crawled(conn, urls) + finally: + conn.close() + print(json.dumps( + { + "status": "crawled_and_ingested", + "seed": args.seed_url, + "depth": args.depth, + "max_pages": args.max_pages, + "discovered": len(urls), + **result, + }, + indent=2, + )) + return 0 + + +def _cmd_crawler_recrawl_check(args: argparse.Namespace) -> int: + """Send conditional HEAD requests for ingested documents. + + Reports each as fresh (304), stale (200, body changed), gone + (404/410), or unreachable. Updates `document_http_meta.last_status` + and `last_checked_at` so consecutive runs target the oldest checks + first. + """ + try: + from aborist.sources.crawler.bridge import recrawl_check + except ImportError as e: + print(f"error: {e}", file=sys.stderr) + return 2 + + conn = connect(args.db) + try: + result = recrawl_check( + conn, + domain=args.domain, + limit=args.limit, + ) + finally: + conn.close() + print(json.dumps(result, indent=2)) + return 0 + + def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="aborist", @@ -2651,6 +2723,56 @@ def build_parser() -> argparse.ArgumentParser: mesh_pull.add_argument("--peer", required=True, help="peer URL, e.g. http://other.example.com:8400") mesh_pull.set_defaults(func=_cmd_mesh_pull) + crawl_cmd = sub.add_parser( + "crawl", + help=( + "BFS-discover same-domain URLs from a seed; optionally ingest " + "and store ETag/Last-Modified for cheap recrawl-checks " + "(requires aborist[crawler] extras)" + ), + ) + crawl_cmd.add_argument("--seed-url", dest="seed_url", required=True) + crawl_cmd.add_argument("--depth", type=int, default=2, help="max BFS depth (default: 2)") + crawl_cmd.add_argument( + "--max-pages", + dest="max_pages", + type=int, + default=20, + help="cap discovery at N URLs (default: 20)", + ) + crawl_cmd.add_argument( + "--ingest", + action="store_true", + help="ingest the discovered pages into --db (default: print URL list only)", + ) + crawl_cmd.set_defaults(func=_cmd_crawl) + + crawler_cmd = sub.add_parser( + "crawler", + help="crawler maintenance verbs (recrawl-check, ...)", + ) + crawler_sub = crawler_cmd.add_subparsers(dest="crawler_op", required=True) + + recrawl_check_cmd = crawler_sub.add_parser( + "recrawl-check", + help=( + "send conditional HEAD requests for ingested documents and " + "classify each as fresh/stale/gone/unreachable" + ), + ) + recrawl_check_cmd.add_argument( + "--domain", + default=None, + help="restrict to documents whose URI contains this domain", + ) + recrawl_check_cmd.add_argument( + "--limit", + type=int, + default=100, + help="check at most N documents (oldest checks first; default: 100)", + ) + recrawl_check_cmd.set_defaults(func=_cmd_crawler_recrawl_check) + return p diff --git a/aborist/sources/crawler/bridge.py b/aborist/sources/crawler/bridge.py new file mode 100644 index 0000000..969a079 --- /dev/null +++ b/aborist/sources/crawler/bridge.py @@ -0,0 +1,316 @@ +"""Bridge: aborist sources/crawler ↔ aborist ingest pipeline. + +Two operations: + +1. **crawl_seed(seed_url, depth, max_pages)** — BFS-discovers same-domain + URLs from a seed, respecting robots.txt + crawl delays. Returns a + list of URLs ready for ingest. + +2. **ingest_crawled(conn, urls)** — fetches each URL with httpx, + captures ETag + Last-Modified, parses HTML, runs through the + standard ingest_source path so the documents land in the same + tables as any other source. After ingest, writes one row per + document into ``document_http_meta`` so a future recrawl-check can + send conditional HEAD requests. + +Off by default — both operations require ``aborist[crawler]`` extras. +""" + +from __future__ import annotations + +import asyncio +import time +import urllib.parse +import urllib.robotparser +from collections import deque +from pathlib import Path +from typing import Iterable, Iterator + +import httpx + +from aborist.document import Document +from aborist.ingest import ingest_source +from aborist.sources.crawler.async_web_fetcher import AsyncWebFetcher +from aborist.sources.html_page import USER_AGENT, parse_html +from aborist.store import transaction + + +def _same_domain(a: str, b: str) -> bool: + """True iff both URLs share an exact hostname (no subdomain crossover).""" + return urllib.parse.urlparse(a).netloc == urllib.parse.urlparse(b).netloc + + +def _normalize(url: str) -> str: + """Canonical form for dedup: scheme + netloc + path + query, no fragment.""" + parsed = urllib.parse.urlparse(url) + cleaned = parsed._replace(fragment="") + return urllib.parse.urlunparse(cleaned) + + +async def _crawl_seed_async( + seed_url: str, + *, + max_depth: int, + max_pages: int, + fetcher: AsyncWebFetcher | None = None, +) -> list[str]: + """BFS from `seed_url`, staying on the same hostname. + + Discovery only — fetches each page enough to extract links, no + aborist-side ingestion happens here. Returns a deduplicated list + of URLs in BFS order, capped at `max_pages`. + """ + fetcher = fetcher or AsyncWebFetcher(user_agent=USER_AGENT) + seen: set[str] = set() + discovered: list[str] = [] + queue: deque[tuple[str, int]] = deque([(_normalize(seed_url), 0)]) + + while queue and len(discovered) < max_pages: + url, depth = queue.popleft() + if url in seen: + continue + seen.add(url) + + try: + html, links = await fetcher.fetch_webpage(url) + except Exception: + continue + if html is None: + continue + + discovered.append(url) + if depth >= max_depth: + continue + + for link in links or []: + link_url = link[0] if isinstance(link, tuple) else link + if not isinstance(link_url, str): + continue + absolute = urllib.parse.urljoin(url, link_url) + normalized = _normalize(absolute) + if normalized in seen: + continue + if not _same_domain(seed_url, normalized): + continue + if not normalized.startswith(("http://", "https://")): + continue + queue.append((normalized, depth + 1)) + + return discovered + + +def crawl_seed( + seed_url: str, + *, + max_depth: int = 2, + max_pages: int = 20, +) -> list[str]: + """Sync wrapper around the async crawler. Returns same-domain URLs.""" + return asyncio.run( + _crawl_seed_async(seed_url, max_depth=max_depth, max_pages=max_pages) + ) + + +class _CrawledHtmlSource: + """Source that fetches each URL synchronously, captures HTTP metadata, + parses HTML, and yields one Document. The metadata is recorded in a + side dict so the caller can write it to ``document_http_meta`` after + ingest commits the document_root. + """ + + source_type = "html" + + def __init__(self, urls: Iterable[str], *, timeout: float = 30.0): + self.urls = list(urls) + self.timeout = timeout + # Filled in during iter_documents — keyed by document_uri (request + # URL pre-redirect) so the caller can map document_root → metadata + # via the documents.document_uri column after ingest. + self.http_meta: dict[str, dict[str, object]] = {} + + def iter_documents(self) -> Iterator[Document]: + with httpx.Client( + headers={"User-Agent": USER_AGENT}, + timeout=self.timeout, + follow_redirects=True, + ) as client: + for url in self.urls: + try: + resp = client.get(url) + resp.raise_for_status() + except httpx.HTTPError: + continue + ctype = resp.headers.get("content-type", "").lower() + if "html" not in ctype and "xml" not in ctype: + continue + doc = parse_html(str(resp.url), resp.text, self.source_type) + if doc is None: + continue + self.http_meta[doc.uri] = { + "etag": resp.headers.get("etag"), + "last_modified": resp.headers.get("last-modified"), + "last_fetched_at": int(time.time()), + } + yield doc + + +def ingest_crawled( + conn, + urls: Iterable[str], + *, + timeout: float = 30.0, +) -> dict: + """Fetch each URL, ingest into `conn`, record ETag + Last-Modified. + + Returns a summary dict with counts and the list of (uri, document_root) + pairs written. The caller's connection must be writable; metadata is + persisted in the same connection so an interrupted run leaves either + the document AND its http_meta or neither (FK guarantees this on + cascade delete; the upsert happens inside the same transaction + boundary as the audit event). + """ + src = _CrawledHtmlSource(urls, timeout=timeout) + stats = ingest_source(conn, src) + written: list[dict] = [] + if not src.http_meta: + return { + "seen": stats.seen, + "inserted": stats.inserted, + "http_meta_written": 0, + "documents": [], + } + + # Map document_uri → document_root via the documents table (URI is + # the post-redirect URL the source set on the Document, which is what + # parse_html stored). The map is stable as long as URIs are unique + # within the corpus, which is enforced at ingest time. + uri_list = list(src.http_meta.keys()) + placeholders = ",".join(["?"] * len(uri_list)) + rows = conn.execute( + f"SELECT document_root, document_uri FROM documents " + f"WHERE document_uri IN ({placeholders})", + uri_list, + ).fetchall() + + with transaction(conn): + for r in rows: + meta = src.http_meta.get(r["document_uri"]) + if meta is None: + continue + conn.execute( + "INSERT OR REPLACE INTO document_http_meta " + "(document_root, etag, last_modified, last_fetched_at, " + " last_status, last_checked_at) " + "VALUES (?, ?, ?, ?, 200, ?)", + ( + r["document_root"], + meta["etag"], + meta["last_modified"], + meta["last_fetched_at"], + meta["last_fetched_at"], + ), + ) + written.append( + { + "document_root": r["document_root"], + "document_uri": r["document_uri"], + } + ) + + return { + "seen": stats.seen, + "inserted": stats.inserted, + "http_meta_written": len(written), + "documents": written, + } + + +def recrawl_check( + conn, + *, + domain: str | None = None, + limit: int = 100, + timeout: float = 10.0, +) -> dict: + """Send conditional HEAD requests for ingested documents and classify + each as fresh/stale/gone/unreachable. + + Fast path: ETag → If-None-Match. Server returns 304 with no body = + skip. Slow path: full re-fetch only if the server says 200. We do + NOT re-fetch the body in this verb — that's recrawl's job. We just + answer "needs recrawl: yes/no" without doing the recrawl. + """ + sql = ( + "SELECT d.document_root, d.document_uri, m.etag, m.last_modified " + "FROM documents d " + "JOIN document_http_meta m ON m.document_root = d.document_root" + ) + args: list[object] = [] + if domain: + sql += " WHERE d.document_uri LIKE ?" + args.append(f"%//{domain}%") + sql += " ORDER BY m.last_checked_at ASC NULLS FIRST LIMIT ?" + args.append(limit) + + rows = conn.execute(sql, args).fetchall() + + fresh, stale, gone, unreachable = 0, 0, 0, 0 + items: list[dict] = [] + now = int(time.time()) + + with httpx.Client( + headers={"User-Agent": USER_AGENT}, + timeout=timeout, + follow_redirects=True, + ) as client: + for r in rows: + headers: dict[str, str] = {} + if r["etag"]: + headers["If-None-Match"] = r["etag"] + if r["last_modified"]: + headers["If-Modified-Since"] = r["last_modified"] + + try: + resp = client.head(r["document_uri"], headers=headers) + status = resp.status_code + except httpx.HTTPError: + status = 0 # network failure / DNS / TLS + + if status == 304: + fresh += 1 + verdict = "fresh" + elif status in (200,): + stale += 1 + verdict = "stale" + elif status in (404, 410): + gone += 1 + verdict = "gone" + else: + unreachable += 1 + verdict = "unreachable" + + with transaction(conn): + conn.execute( + "UPDATE document_http_meta " + "SET last_status = ?, last_checked_at = ? " + "WHERE document_root = ?", + (status, now, r["document_root"]), + ) + + items.append( + { + "document_root": r["document_root"], + "document_uri": r["document_uri"], + "status": status, + "verdict": verdict, + } + ) + + return { + "examined": len(rows), + "fresh": fresh, + "stale": stale, + "gone": gone, + "unreachable": unreachable, + "items": items, + } diff --git a/aborist/store.py b/aborist/store.py index 61f8980..395c351 100644 --- a/aborist/store.py +++ b/aborist/store.py @@ -63,6 +63,22 @@ CREATE TABLE IF NOT EXISTS documents ( CREATE INDEX IF NOT EXISTS idx_documents_uri ON documents(document_uri); CREATE INDEX IF NOT EXISTS idx_documents_kind ON documents(kind); +-- Per-document HTTP metadata for cheap recrawl-checks. ETag and +-- Last-Modified come from the response headers at ingest time and let a +-- recrawl pass send conditional HEAD requests (If-None-Match / +-- If-Modified-Since); a 304 lets us skip the body fetch entirely. +-- last_status / last_checked_at record the most recent recheck so +-- operators can audit "when was this URL last verified." +CREATE TABLE IF NOT EXISTS document_http_meta ( + document_root TEXT PRIMARY KEY, + etag TEXT, + last_modified TEXT, + last_fetched_at INTEGER NOT NULL, + last_status INTEGER, + last_checked_at INTEGER, + FOREIGN KEY (document_root) REFERENCES documents(document_root) ON DELETE CASCADE +); + -- Chunks with tier-based reversible eviction. -- content nullable: cold tier evicts content but retains leaf_hash + URI for -- rehydration. Identity verified on rehydrate by recomputing leaf_hash. @@ -310,6 +326,7 @@ def connect(db_path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection: conn.executescript(SCHEMA_SQL) _migrate_audit_mode(conn) _migrate_mesh_peer_chains(conn) + _migrate_document_http_meta(conn) conn.execute("PRAGMA synchronous = NORMAL") conn.execute("PRAGMA cache_size = -65536") conn.execute("PRAGMA temp_store = MEMORY") @@ -369,6 +386,31 @@ def _migrate_audit_mode(conn: sqlite3.Connection) -> None: ) +def _migrate_document_http_meta(conn: sqlite3.Connection) -> None: + """Forward-migrate pre-recrawl-check shards. + + Adds ``document_http_meta`` to DBs that pre-date the recrawl-check + feature. Mirrors ``_migrate_mesh_peer_chains``: idempotent + table-existence probe, then CREATE if missing. + """ + row = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='document_http_meta'" + ).fetchone() + if row is None: + conn.execute( + "CREATE TABLE document_http_meta (" + " document_root TEXT PRIMARY KEY," + " etag TEXT," + " last_modified TEXT," + " last_fetched_at INTEGER NOT NULL," + " last_status INTEGER," + " last_checked_at INTEGER," + " FOREIGN KEY (document_root) REFERENCES documents(document_root)" + " ON DELETE CASCADE" + ")" + ) + + def _migrate_mesh_peer_chains(conn: sqlite3.Connection) -> None: """Forward-migrate pre-mesh-fork-detection shards. diff --git a/tests/crawler/test_bridge.py b/tests/crawler/test_bridge.py new file mode 100644 index 0000000..831cb25 --- /dev/null +++ b/tests/crawler/test_bridge.py @@ -0,0 +1,387 @@ +"""Bridge tests: crawler discovery + ingest + recrawl-check. + +These tests stub the network. The verbatim crawler tests in this same +directory exercise the underlying AsyncWebFetcher; here we exercise +the aborist bridge logic on top of it (BFS bounds, same-domain filter, +ETag capture, conditional HEAD classification). +""" + +from __future__ import annotations + +import json +import time +from pathlib import Path +from unittest.mock import patch + +import httpx +import pytest + +from aborist.sources.crawler.bridge import ( + _normalize, + _same_domain, + crawl_seed, + ingest_crawled, + recrawl_check, +) +from aborist.store import connect + + +# --------------------------------------------------------------------------- +# URL helpers (pure) +# --------------------------------------------------------------------------- + + +def test_same_domain_exact_hostname(): + assert _same_domain("https://x.com/a", "https://x.com/b") is True + + +def test_same_domain_rejects_subdomain(): + assert _same_domain("https://x.com/a", "https://docs.x.com/b") is False + + +def test_same_domain_rejects_other_host(): + assert _same_domain("https://x.com/a", "https://y.com/a") is False + + +def test_normalize_strips_fragment(): + assert _normalize("https://x.com/a#section") == "https://x.com/a" + + +def test_normalize_preserves_query(): + assert _normalize("https://x.com/a?q=1") == "https://x.com/a?q=1" + + +# --------------------------------------------------------------------------- +# crawl_seed: BFS discovery (no real network — mock the fetcher) +# --------------------------------------------------------------------------- + + +class _StubFetcher: + """Stand-in for AsyncWebFetcher that returns pre-canned (html, links) + based on a URL→links map. Enough to drive the bridge's BFS. + """ + + def __init__(self, page_links: dict[str, list[str]]): + self.page_links = page_links + + async def fetch_webpage(self, url: str): + return "", self.page_links.get(url, []) + + +def _run_crawl(page_links, seed, depth, max_pages): + """Helper: drive _crawl_seed_async with a stub fetcher.""" + import asyncio + from aborist.sources.crawler import bridge + + return asyncio.run( + bridge._crawl_seed_async( + seed, max_depth=depth, max_pages=max_pages, fetcher=_StubFetcher(page_links) + ) + ) + + +def test_crawl_seed_breadth_first_within_limit(): + page_links = { + "https://x.com/": ["/a", "/b"], + "https://x.com/a": ["/c"], + "https://x.com/b": [], + "https://x.com/c": [], + } + out = _run_crawl(page_links, "https://x.com/", depth=2, max_pages=10) + # Seed first, then both depth-1 pages, then the depth-2 page. + assert out[0] == "https://x.com/" + assert "https://x.com/a" in out + assert "https://x.com/b" in out + assert "https://x.com/c" in out + + +def test_crawl_seed_respects_max_pages(): + page_links = {f"https://x.com/p{i}": [f"/p{i+1}"] for i in range(50)} + page_links["https://x.com/"] = ["/p0"] + out = _run_crawl(page_links, "https://x.com/", depth=10, max_pages=5) + assert len(out) == 5 + + +def test_crawl_seed_respects_max_depth(): + page_links = { + "https://x.com/": ["/a"], + "https://x.com/a": ["/b"], + "https://x.com/b": ["/c"], + "https://x.com/c": [], + } + out = _run_crawl(page_links, "https://x.com/", depth=1, max_pages=10) + # Seed (depth 0) + /a (depth 1) only. /b is depth 2, skipped. + assert "https://x.com/" in out + assert "https://x.com/a" in out + assert "https://x.com/b" not in out + + +def test_crawl_seed_drops_off_domain_links(): + page_links = { + "https://x.com/": ["https://other.com/foo", "/local"], + "https://x.com/local": [], + } + out = _run_crawl(page_links, "https://x.com/", depth=2, max_pages=10) + assert "https://other.com/foo" not in out + assert "https://x.com/local" in out + + +def test_crawl_seed_dedups(): + page_links = { + "https://x.com/": ["/a", "/a", "/b"], + "https://x.com/a": ["/b"], + "https://x.com/b": ["/a"], + } + out = _run_crawl(page_links, "https://x.com/", depth=3, max_pages=10) + # Each URL appears at most once even with cycles. + assert len(out) == len(set(out)) + + +# --------------------------------------------------------------------------- +# ingest_crawled: stores ETag + Last-Modified per document_root +# --------------------------------------------------------------------------- + + +@pytest.fixture +def http_meta_db(tmp_path): + db = tmp_path / "meta.db" + # Touch the DB so the schema is initialised, then close — we'll + # reopen as the test needs. + conn = connect(db) + try: + pass + finally: + conn.close() + return db + + +def _patched_httpx_get(url_to_response: dict[str, dict]): + """Return a context-manager class that fakes httpx.Client.get.""" + + class _FakeClient: + def __init__(self, *a, **kw): + self._responses = url_to_response + + def __enter__(self): + return self + + def __exit__(self, *a): + pass + + def get(self, url): + spec = self._responses.get(url) + if spec is None: + raise httpx.HTTPError(f"no stub for {url}") + req = httpx.Request("GET", url) + return httpx.Response( + spec.get("status", 200), + headers=spec.get("headers", {"content-type": "text/html"}), + content=spec.get("body", "hi").encode(), + request=req, + ) + + def head(self, url, headers=None): + spec = self._responses.get(url) + if spec is None: + raise httpx.HTTPError(f"no stub for {url}") + head_status = spec.get("head_status", 200) + req = httpx.Request("HEAD", url, headers=headers or {}) + return httpx.Response( + head_status, + headers=spec.get("head_headers", {}), + request=req, + ) + + return _FakeClient + + +def test_ingest_crawled_writes_http_meta(http_meta_db): + url = "https://x.com/page" + responses = { + url: { + "status": 200, + "headers": { + "content-type": "text/html", + "etag": '"abc123"', + "last-modified": "Tue, 28 Apr 2026 00:00:00 GMT", + }, + "body": "

Title

Hello world.", + } + } + + with patch("aborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)): + conn = connect(http_meta_db) + try: + result = ingest_crawled(conn, [url]) + finally: + conn.close() + + assert result["http_meta_written"] == 1 + assert len(result["documents"]) == 1 + document_root = result["documents"][0]["document_root"] + + conn = connect(http_meta_db) + try: + row = conn.execute( + "SELECT etag, last_modified, last_status, last_fetched_at, last_checked_at " + "FROM document_http_meta WHERE document_root = ?", + (document_root,), + ).fetchone() + finally: + conn.close() + assert row["etag"] == '"abc123"' + assert row["last_modified"] == "Tue, 28 Apr 2026 00:00:00 GMT" + assert row["last_status"] == 200 + assert row["last_fetched_at"] > 0 + + +def test_ingest_crawled_skips_non_html(http_meta_db): + url = "https://x.com/file.pdf" + responses = { + url: { + "status": 200, + "headers": {"content-type": "application/pdf"}, + "body": "%PDF-1.4 binary", + } + } + with patch("aborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)): + conn = connect(http_meta_db) + try: + result = ingest_crawled(conn, [url]) + finally: + conn.close() + assert result["http_meta_written"] == 0 + assert result["inserted"] == 0 + + +# --------------------------------------------------------------------------- +# recrawl_check: conditional HEAD classification +# --------------------------------------------------------------------------- + + +def _seed_meta(conn, *, document_root: str, document_uri: str, etag: str | None = None): + """Insert a documents row + http_meta row for recheck tests.""" + from aborist.store import transaction + + with transaction(conn): + conn.execute( + "INSERT INTO documents " + "(document_root, document_uri, source_type, kind, compression_depth, " + " title, chunking_version, canonicalization_version, schema_version, " + " ingest_ts, hit_count) " + "VALUES (?, ?, 'html', 'surface', 0, NULL, 'tok-512-v1', 'norm-v1', 'v9.8.0', ?, 0)", + (document_root, document_uri, int(time.time())), + ) + conn.execute( + "INSERT INTO document_http_meta " + "(document_root, etag, last_modified, last_fetched_at, last_status, last_checked_at) " + "VALUES (?, ?, NULL, ?, 200, NULL)", + (document_root, etag, int(time.time())), + ) + + +def test_recrawl_check_classifies_304_as_fresh(tmp_path): + db = tmp_path / "rc.db" + conn = connect(db) + try: + _seed_meta(conn, document_root="aa" * 32, document_uri="https://x.com/p", etag='"v1"') + finally: + conn.close() + + responses = {"https://x.com/p": {"head_status": 304}} + with patch("aborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)): + conn = connect(db) + try: + result = recrawl_check(conn, limit=10) + finally: + conn.close() + assert result["fresh"] == 1 + assert result["stale"] == 0 + assert result["items"][0]["verdict"] == "fresh" + + +def test_recrawl_check_classifies_200_as_stale(tmp_path): + db = tmp_path / "rc.db" + conn = connect(db) + try: + _seed_meta(conn, document_root="bb" * 32, document_uri="https://x.com/q", etag='"v1"') + finally: + conn.close() + + responses = {"https://x.com/q": {"head_status": 200}} + with patch("aborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)): + conn = connect(db) + try: + result = recrawl_check(conn, limit=10) + finally: + conn.close() + assert result["stale"] == 1 + assert result["items"][0]["verdict"] == "stale" + + +def test_recrawl_check_classifies_404_as_gone(tmp_path): + db = tmp_path / "rc.db" + conn = connect(db) + try: + _seed_meta(conn, document_root="cc" * 32, document_uri="https://x.com/r", etag='"v1"') + finally: + conn.close() + + responses = {"https://x.com/r": {"head_status": 404}} + with patch("aborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)): + conn = connect(db) + try: + result = recrawl_check(conn, limit=10) + finally: + conn.close() + assert result["gone"] == 1 + + +def test_recrawl_check_domain_filter(tmp_path): + db = tmp_path / "rc.db" + conn = connect(db) + try: + _seed_meta(conn, document_root="11" * 32, document_uri="https://keep.com/a", etag='"v"') + _seed_meta(conn, document_root="22" * 32, document_uri="https://drop.com/a", etag='"v"') + finally: + conn.close() + + responses = { + "https://keep.com/a": {"head_status": 304}, + # drop.com would error if the filter is broken; we expect it to be skipped. + } + with patch("aborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)): + conn = connect(db) + try: + result = recrawl_check(conn, domain="keep.com", limit=10) + finally: + conn.close() + assert result["examined"] == 1 + assert result["items"][0]["document_uri"] == "https://keep.com/a" + + +def test_recrawl_check_updates_last_checked_at(tmp_path): + db = tmp_path / "rc.db" + conn = connect(db) + try: + _seed_meta(conn, document_root="33" * 32, document_uri="https://x.com/s", etag='"v1"') + finally: + conn.close() + + responses = {"https://x.com/s": {"head_status": 304}} + with patch("aborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)): + conn = connect(db) + try: + recrawl_check(conn, limit=10) + finally: + conn.close() + + conn = connect(db) + try: + row = conn.execute( + "SELECT last_status, last_checked_at FROM document_http_meta WHERE document_root='33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33'" + ).fetchone() + finally: + conn.close() + assert row["last_status"] == 304 + assert row["last_checked_at"] is not None