feat: add opt-in exact Markdown source receipts

crawl4ai/async_configs.py

@@ -1968,6 +1968,7 @@ def __init__(
         # Anti-Bot Retry Parameters
         max_retries: int = 0,
         fallback_fetch_function: Optional[Callable[[str], Awaitable[str]]] = None,
+        source_receipt: bool = False,
     ):
         # TODO: Planning to set properties dynamically based on the __init__ signature
         self.url = url
@@ -2008,6 +2009,7 @@ def __init__(
 
         # Caching Parameters
         # Accept "enabled" etc. from CLI -c and YAML/JSON config files
+        self.source_receipt = source_receipt
         self.cache_mode = CacheMode(cache_mode.lower()) if isinstance(cache_mode, str) else cache_mode
         self.session_id = session_id
         self.bypass_cache = bypass_cache
@@ -2392,6 +2394,7 @@ def to_dict(self):
             "timezone_id": self.timezone_id,
             "geolocation": self.geolocation,
             "fetch_ssl_certificate": self.fetch_ssl_certificate,
+            "source_receipt": self.source_receipt,
             "cache_mode": self.cache_mode,
             "session_id": self.session_id,
             "bypass_cache": self.bypass_cache,

crawl4ai/async_webcrawler.py

@@ -30,6 +30,7 @@
     AsyncCrawlResponse,
 )
 from .cache_context import CacheMode, CacheContext
+from .source_receipt import attach_source_receipt
 from .markdown_generation_strategy import (
     DefaultMarkdownGenerator,
     MarkdownGenerationStrategy,
@@ -275,6 +276,7 @@ async def arun(
                 screenshot_data = None
                 pdf_data = None
                 extracted_content = None
+                source_fetched_at = None
                 start_time = time.perf_counter()
 
                 # Try to get cached result if appropriate
@@ -477,6 +479,7 @@ async def arun(
                                     tag="FETCH",
                                 )
 
+                                source_fetched_at = time.time()
                                 crawl_result = await self.aprocess_html(
                                     url=url, html=html,
                                     extracted_content=extracted_content,
@@ -571,6 +574,7 @@ async def arun(
                             _crawl_stats["fallback_fetch_used"] = True
                             try:
                                 _fallback_html = await _fallback_fn(url)
+                                source_fetched_at = time.time()
                                 if _fallback_html:
                                     _sanitized_html = sanitize_input_encode(_fallback_html)
                                     try:
@@ -681,6 +685,9 @@ async def arun(
                         tag="COMPLETE",
                     )
 
+                    if config.source_receipt:
+                        attach_source_receipt(crawl_result, config, fetched_at=source_fetched_at)
+
                     # Update cache if appropriate
                     if cache_context.should_write() and not bool(cached_result):
                         await async_db_manager.acache_url(crawl_result)
@@ -702,6 +709,8 @@ async def arun(
                     # For raw: URLs, don't fall back to the raw HTML string as redirected_url
                     is_raw_url = url.startswith("raw:") or url.startswith("raw://")
                     cached_result.redirected_url = cached_result.redirected_url or (None if is_raw_url else url)
+                    if config.source_receipt:
+                        attach_source_receipt(cached_result, config)
                     return CrawlResultContainer(cached_result)
 
             except Exception as e:

crawl4ai/models.py

@@ -161,6 +161,7 @@ class CrawlResult(BaseModel):
     network_requests: Optional[List[Dict[str, Any]]] = None
     console_messages: Optional[List[Dict[str, Any]]] = None
     tables: List[Dict] = Field(default_factory=list)  # NEW – [{headers,rows,caption,summary}]
+    source_receipt: Optional[Dict[str, Any]] = None
     # Cache validation metadata (Smart Cache)
     head_fingerprint: Optional[str] = None
     cached_at: Optional[float] = None

crawl4ai/source_receipt.py

@@ -0,0 +1,97 @@
+"""Opt-in provenance for exact returned Markdown; never serialize config secrets."""
+
+import hashlib
+import json
+from datetime import datetime, timezone
+from urllib.parse import urlsplit, urlunsplit
+
+from .__version__ import __version__
+
+_METADATA_KEY = "_crawl4ai_source_receipt_v1"
+_PROCESSING_OPTIONS = (
+    "word_count_threshold",
+    "only_text",
+    "remove_forms",
+    "parser_type",
+)
+
+
+def _url(value):
+    """Omit credentials, query parameters, fragments, and caller-provided HTML."""
+    try:
+        parts = urlsplit(value or "")
+        if parts.scheme not in ("http", "https") or not parts.hostname:
+            return None
+        host = parts.hostname
+        if ":" in host:
+            host = "[" + host + "]"
+        if parts.port:
+            host += ":" + str(parts.port)
+        return urlunsplit((parts.scheme, host, parts.path, "", ""))
+    except ValueError:
+        return None
+
+
+def _timestamp(value):
+    return (
+        datetime.fromtimestamp(value, timezone.utc).isoformat()
+        if value is not None
+        else None
+    )
+
+
+def attach_source_receipt(result, config, *, fetched_at=None):
+    """Attach a JSON-safe receipt and persist minimal provenance in cache metadata.
+
+    Cache hits use only previously recorded fetch metadata. Legacy cache entries
+    keep unknown values as null; served_at is never substituted for fetched_at.
+    """
+    if not result.success:
+        return
+    cached = (result.cache_status or "").startswith("hit")
+    metadata = dict(result.metadata or {})
+    previous = metadata.get(_METADATA_KEY)
+    if cached:
+        provenance = previous if isinstance(previous, dict) else {}
+    else:
+        options = {key: getattr(config, key, None) for key in _PROCESSING_OPTIONS}
+        # Only a small, documented allowlist of primitive processing options.
+        encoded = json.dumps(
+            options, sort_keys=True, separators=(",", ":"), allow_nan=False
+        )
+        provenance = {
+            "fetched_at": (
+                None
+                if result.url.startswith(("raw:", "file:"))
+                else _timestamp(fetched_at)
+            ),
+            "crawler_version": __version__,
+            "processing_fingerprint": hashlib.sha256(
+                encoded.encode("utf-8")
+            ).hexdigest(),
+        }
+        metadata[_METADATA_KEY] = provenance
+        result.metadata = metadata
+    markdown = result.markdown
+    outputs = {}
+    for kind in ("raw_markdown", "fit_markdown"):
+        value = getattr(markdown, kind, None) if markdown is not None else None
+        if isinstance(value, str):
+            payload = value.encode("utf-8")
+            outputs[kind] = {
+                "sha256": hashlib.sha256(payload).hexdigest(),
+                "utf8_bytes": len(payload),
+            }
+    result.source_receipt = {
+        "schema_version": 1,
+        "requested_url": _url(result.url),
+        "final_url": _url(result.redirected_url),
+        "fetched_at": provenance.get("fetched_at"),
+        "served_at": datetime.now(timezone.utc).isoformat(),
+        "cache_status": result.cache_status,
+        "crawler_version": provenance.get("crawler_version"),
+        "processing_fingerprint": provenance.get("processing_fingerprint"),
+        "processing_option_names": list(_PROCESSING_OPTIONS),
+        "digest_rule": "sha256-exact-utf8-v1",
+        "outputs": outputs,
+    }

docs/examples/source_receipt.md

@@ -0,0 +1,23 @@
+# Optional source receipts
+
+Pass `CrawlerRunConfig(source_receipt=True)` and inspect `result.source_receipt`.
+The JSON-safe receipt hashes each available raw/fit Markdown output as exact UTF-8
+bytes, without whitespace normalization. It records served time separately from
+fetch completion time and reports cache status. Minimal fetch provenance persists
+in the reserved `_crawl4ai_source_receipt_v1` metadata key without a database migration.
+
+On cache hits, original recorded fetch time and processing fingerprint are reused.
+Legacy cache entries have unknown fetch time/version/options (`null`), rather than
+inventing a fresh fetch time from the cache write or current request. Provided raw
+HTML and local files likewise do not claim a network fetch time. Failed crawls do
+not get receipts. Cached Markdown may expose fewer output kinds than a live result;
+the receipt hashes only the exact outputs actually returned.
+
+Credentials, query strings and fragments are omitted from receipt URLs, and raw
+HTML/file URLs are not included. URL paths can still contain private data; receipts
+are provenance for your application, not automatically safe publication records.
+Only `word_count_threshold`, `only_text`, `remove_forms`, and `parser_type` enter the
+processing fingerprint. It is explicitly not a complete cache key: selectors, custom
+strategies, model/provider parameters and secrets are not serialized or fingerprinted.
+The text digest itself identifies the returned bytes; it does not prove truth,
+source rights, or equivalence of different processing configurations.

tests/test_source_receipt.py

@@ -0,0 +1,131 @@
+import hashlib
+import json
+from unittest.mock import AsyncMock, MagicMock
+import asyncio
+
+from crawl4ai import CrawlerRunConfig
+from crawl4ai.models import CrawlResult, MarkdownGenerationResult
+from crawl4ai.source_receipt import attach_source_receipt
+
+
+def test_arun_live_then_cache_uses_persisted_metadata(tmp_path, monkeypatch):
+    async def exercise():
+        from crawl4ai import AsyncWebCrawler, CacheMode
+        from crawl4ai.models import AsyncCrawlResponse
+        from crawl4ai.async_database import async_db_manager
+
+        import crawl4ai.async_webcrawler as module
+        from crawl4ai.utils import RobotsParser
+
+        monkeypatch.setattr(
+            module,
+            "RobotsParser",
+            lambda: RobotsParser(cache_dir=str(tmp_path / "robots")),
+        )
+        strategy = MagicMock()
+        strategy.crawl = AsyncMock(
+            return_value=AsyncCrawlResponse(
+                html="<html><body><p>Fictional example page.</p></body></html>",
+                response_headers={},
+                status_code=200,
+            )
+        )
+        crawler = AsyncWebCrawler(
+            crawler_strategy=strategy, base_directory=str(tmp_path)
+        )
+        crawler.ready = True
+        crawler.aprocess_html = AsyncMock(return_value=result())
+        cached = None
+
+        async def read(_url):
+            return cached
+
+        async def write(item):
+            nonlocal cached
+            cached = result(
+                metadata=json.loads(json.dumps(item.metadata)), cached_at=2000
+            )
+
+        monkeypatch.setattr(async_db_manager, "aget_cached_url", read)
+        monkeypatch.setattr(async_db_manager, "acache_url", write)
+        config = CrawlerRunConfig(source_receipt=True, cache_mode=CacheMode.ENABLED)
+        first = await crawler.arun("https://example.org/page", config=config)
+        assert first.success
+        assert first.source_receipt["fetched_at"] is not None
+        replay = await crawler.arun("https://example.org/page", config=config)
+        assert replay.source_receipt["cache_status"] == "hit"
+        assert replay.source_receipt["fetched_at"] == first.source_receipt["fetched_at"]
+        strategy.crawl.assert_awaited_once()
+        crawler.aprocess_html.assert_awaited_once()
+
+    asyncio.run(exercise())
+
+
+def result(**kwargs):
+    return CrawlResult(
+        url="https://example.org/page?token=synthetic",
+        html="<p>Example</p>",
+        success=True,
+        markdown=MarkdownGenerationResult(
+            raw_markdown="中文\n\n",
+            markdown_with_citations="",
+            references_markdown="",
+            fit_markdown="fit",
+            fit_html="",
+        ),
+        **kwargs,
+    )
+
+
+def test_exact_returned_text_and_serialization():
+    item = result(cache_status="miss")
+    attach_source_receipt(item, CrawlerRunConfig(source_receipt=True), fetched_at=1000)
+    receipt = item.source_receipt
+    assert (
+        receipt["outputs"]["raw_markdown"]["sha256"]
+        == hashlib.sha256("中文\n\n".encode()).hexdigest()
+    )
+    assert receipt["outputs"]["raw_markdown"]["utf8_bytes"] == 8
+    assert receipt["fetched_at"].startswith("1970-01-01T00:16:40")
+    assert "synthetic" not in json.dumps(receipt)
+    assert item.model_dump()["source_receipt"] == receipt
+
+
+def test_cache_hit_preserves_fetch_and_processing_identity():
+    item = result(cache_status="miss")
+    attach_source_receipt(item, CrawlerRunConfig(), fetched_at=1000)
+    original = item.source_receipt
+    # Cache persists metadata; it need not persist the opt-in result field itself.
+    replay = result(cache_status="hit", metadata=json.loads(json.dumps(item.metadata)))
+    attach_source_receipt(replay, CrawlerRunConfig(word_count_threshold=999))
+    assert replay.source_receipt["fetched_at"] == original["fetched_at"]
+    assert (
+        replay.source_receipt["processing_fingerprint"]
+        == original["processing_fingerprint"]
+    )
+    assert replay.source_receipt["outputs"] == original["outputs"]
+
+
+def test_legacy_cache_keeps_unknown_fetch_time_and_configuration():
+    item = result(cache_status="hit_fallback", cached_at=1000)
+    attach_source_receipt(item, CrawlerRunConfig())
+    assert item.source_receipt["fetched_at"] is None
+    assert item.source_receipt["processing_fingerprint"] is None
+    assert item.source_receipt["served_at"]
+
+
+def test_receipt_is_opt_in_and_config_round_trips():
+    assert CrawlerRunConfig().source_receipt is False
+    assert result().source_receipt is None
+    assert CrawlerRunConfig(source_receipt=True).clone().source_receipt is True
+
+
+def test_no_raw_html_or_credentials_and_failed_results_have_no_receipt():
+    item = CrawlResult(url="raw:<p>private fixture</p>", html="", success=True)
+    attach_source_receipt(item, CrawlerRunConfig(), fetched_at=1000)
+    assert item.source_receipt["requested_url"] is None
+    assert item.source_receipt["fetched_at"] is None
+    assert "private fixture" not in json.dumps(item.source_receipt)
+    failed = CrawlResult(url="https://example.org", html="", success=False)
+    attach_source_receipt(failed, CrawlerRunConfig())
+    assert failed.source_receipt is None