diff --git a/docs/source_freshness_evidence.md b/docs/source_freshness_evidence.md
new file mode 100644
index 0000000..3ca2e76
--- /dev/null
+++ b/docs/source_freshness_evidence.md
@@ -0,0 +1,24 @@
+# Source freshness evidence v1
+
+This producer-boundary contract records freshness evidence for one bounded feed
+response. It does not claim that a feed is globally complete, and it does not
+use event `published_at` as a provider freshness signal.
+
+Signal priority is fixed:
+
+1. Atom feed `updated`;
+2. RSS channel `lastBuildDate`;
+3. HTTP `Last-Modified`.
+
+HTTP `Date` is recorded as retrieval/server-time evidence only and never makes
+the source eligible. A present but invalid, future, or stale higher-priority
+signal fails closed; fallback is allowed only when that signal is absent. The
+selected signal must be no more than five minutes in the future and no older
+than eight days relative to the caller-supplied canonical UTC reference.
+
+The evidence is canonical JSON bytes containing only digests for the source URL
+identity and response body, all signal presence/value/kind records, the
+reference, policy version, and decision. It does not persist the source URL or
+request headers. Missing all selectable signals produces
+`source_freshness_unverified`. A period projection with zero selected events is
+not stale by itself; H2C feed quarantine remains a separate producer result.
diff --git a/src/political_event_tracking_research/rss_source_fetch.py b/src/political_event_tracking_research/rss_source_fetch.py
index 0884243..a1b9be0 100644
--- a/src/political_event_tracking_research/rss_source_fetch.py
+++ b/src/political_event_tracking_research/rss_source_fetch.py
@@ -70,6 +70,10 @@ def load_feed_config(path: str | Path) -> list[FeedConfig]:
 
 
 def fetch_url(url: str) -> bytes:
+    return fetch_url_with_metadata(url)[0]
+
+
+def fetch_url_with_metadata(url: str) -> tuple[bytes, dict[str, str]]:
     request = urllib.request.Request(
         url,
         headers={
@@ -79,9 +83,20 @@ def fetch_url(url: str) -> bytes:
     )
     with urllib.request.urlopen(request, timeout=20) as response:
         payload = response.read(MAX_XML_BYTES + 1)
+        response_headers = getattr(response, "headers", None)
+        headers = {}
+        if response_headers is not None:
+            for name in ("Date", "Last-Modified"):
+                get_all = getattr(response_headers, "get_all", None)
+                values = get_all(name) if callable(get_all) else None
+                if values is not None and len(values) > 1:
+                    raise FeedXmlError("feed_header_duplicate")
+                value = response_headers.get(name)
+                if value is not None:
+                    headers[name] = value
     if len(payload) > MAX_XML_BYTES:
         raise FeedXmlError("feed_xml_oversize")
-    return payload
+    return payload, headers
 
 
 def strip_html(value: str) -> str:
diff --git a/src/political_event_tracking_research/source_freshness.py b/src/political_event_tracking_research/source_freshness.py
new file mode 100644
index 0000000..f7edeb2
--- /dev/null
+++ b/src/political_event_tracking_research/source_freshness.py
@@ -0,0 +1,320 @@
+"""Producer-boundary freshness evidence for one bounded feed snapshot."""
+
+from __future__ import annotations
+
+import datetime as dt
+import email.utils
+import hashlib
+import json
+import re
+from collections.abc import Mapping
+from urllib.parse import urlsplit
+
+import defusedxml.ElementTree as ET
+from defusedxml.common import DefusedXmlException
+
+
+EVIDENCE_VERSION = "pert.source_freshness_evidence.v1"
+POLICY_VERSION = "monday_weekly_8d_5m.v1"
+MAX_BODY_BYTES = 1024 * 1024
+MAX_HEADER_VALUE_BYTES = 4096
+MAX_FRESHNESS_AGE = dt.timedelta(days=8)
+FUTURE_TOLERANCE = dt.timedelta(minutes=5)
+_DIGEST_RE = re.compile(r"^[0-9a-f]{64}$")
+_TIMESTAMP_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
+_EVIDENCE_KEYS = frozenset(
+    {
+        "evidence_version",
+        "policy_version",
+        "feed_id",
+        "source_identity_digest",
+        "body_sha256",
+        "reference_time",
+        "selected_signal",
+        "signals",
+        "decision",
+    }
+)
+_SIGNAL_KEYS = frozenset({"kind", "present", "valid", "value"})
+_SIGNAL_KINDS = (
+    "atom_feed_updated",
+    "rss_channel_last_build_date",
+    "http_last_modified",
+    "http_date",
+)
+_SELECTABLE = frozenset(_SIGNAL_KINDS[:3])
+
+
+class FreshnessError(ValueError):
+    """Stable, sanitized producer freshness contract error."""
+
+    def __init__(self, code: str) -> None:
+        self.code = code
+        super().__init__(code)
+
+
+def _fail(code: str) -> None:
+    raise FreshnessError(code)
+
+
+def _string(value: object, code: str, *, allow_empty: bool = False) -> str:
+    if type(value) is not str or (not allow_empty and not value) or any(ord(char) < 0x20 for char in value):
+        _fail(code)
+    return value
+
+
+def _timestamp(value: object, code: str = "freshness_time_invalid") -> dt.datetime:
+    if type(value) is not str or not _TIMESTAMP_RE.fullmatch(value):
+        _fail(code)
+    try:
+        return dt.datetime.fromisoformat(value[:-1] + "+00:00")
+    except ValueError:
+        _fail(code)
+
+
+def _canonical_timestamp(value: dt.datetime) -> str:
+    return value.astimezone(dt.UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+def _parse_date_signal(value: object) -> tuple[bool, str | None]:
+    if type(value) is not str or not value or len(value) > MAX_HEADER_VALUE_BYTES:
+        return False, None
+    try:
+        parsed = email.utils.parsedate_to_datetime(value)
+        if parsed is None:
+            raise ValueError
+        if parsed.tzinfo is None:
+            return False, None
+        return True, _canonical_timestamp(parsed)
+    except (TypeError, ValueError, OverflowError):
+        try:
+            parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
+            if parsed.tzinfo is None:
+                return False, None
+            return True, _canonical_timestamp(parsed)
+        except (TypeError, ValueError, OverflowError):
+            return False, None
+
+
+def _headers(value: Mapping[str, str]) -> dict[str, str]:
+    if not isinstance(value, Mapping):
+        _fail("freshness_headers_invalid")
+    result: dict[str, str] = {}
+    try:
+        items = list(value.items())
+    except (AttributeError, RuntimeError, TypeError, ValueError):
+        _fail("freshness_headers_invalid")
+    for key, item in items:
+        if (
+            type(key) is not str
+            or any(ord(char) < 0x20 for char in key)
+            or type(item) is not str
+            or len(item.encode("utf-8")) > MAX_HEADER_VALUE_BYTES
+        ):
+            _fail("freshness_headers_invalid")
+        normalized = key.lower()
+        if normalized in result:
+            _fail("freshness_headers_invalid")
+        result[normalized] = item
+    return result
+
+
+def _source_identity(source_url: object) -> str:
+    value = _string(source_url, "source_identity_invalid")
+    try:
+        parsed = urlsplit(value)
+        if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password:
+            _fail("source_identity_invalid")
+    except ValueError:
+        _fail("source_identity_invalid")
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()
+
+
+def _xml_signal(body: bytes, kind: str) -> tuple[bool, str | None]:
+    try:
+        root = ET.fromstring(body, forbid_dtd=True, forbid_entities=True, forbid_external=True)
+    except (DefusedXmlException, ET.ParseError, LookupError, UnicodeError, ValueError, RecursionError):
+        _fail("source_freshness_invalid")
+    if root.tag == "rss":
+        channels = [child for child in root if child.tag == "channel"]
+        if len(channels) != 1:
+            _fail("source_freshness_invalid")
+        if kind == "rss_channel_last_build_date":
+            values = [child.text.strip() if child.text is not None else "" for child in channels[0] if child.tag == "lastBuildDate"]
+            if len(values) > 1:
+                _fail("source_freshness_invalid")
+            if not values:
+                return False, None
+            return True, _parse_date_signal(values[0])[1]
+        return False, None
+    if root.tag == "{http://www.w3.org/2005/Atom}feed":
+        if kind == "atom_feed_updated":
+            values = [child.text.strip() if child.text is not None else "" for child in root if child.tag == "{http://www.w3.org/2005/Atom}updated"]
+            if len(values) > 1:
+                _fail("source_freshness_invalid")
+            if not values:
+                return False, None
+            return True, _parse_date_signal(values[0])[1]
+        return False, None
+    _fail("source_freshness_invalid")
+
+
+def _signal(kind: str, present: bool, value: str | None, valid: bool) -> dict[str, object]:
+    return {"kind": kind, "present": present, "valid": valid, "value": value}
+
+
+def _validate_signal(value: object) -> dict[str, object]:
+    if not isinstance(value, Mapping) or set(value) != _SIGNAL_KEYS:
+        _fail("freshness_invalid")
+    kind = _string(value["kind"], "freshness_invalid")
+    if kind not in _SIGNAL_KINDS or type(value["present"]) is not bool or type(value["valid"]) is not bool:
+        _fail("freshness_invalid")
+    signal_value = value["value"]
+    if signal_value is not None:
+        _timestamp(signal_value, "freshness_invalid")
+    if not value["present"] and (signal_value is not None or value["valid"]):
+        _fail("freshness_invalid")
+    if value["valid"] and signal_value is None:
+        _fail("freshness_invalid")
+    return {key: value[key] for key in ("kind", "present", "valid", "value")}
+
+
+def _validate_evidence(value: object) -> dict[str, object]:
+    if not isinstance(value, Mapping) or set(value) != _EVIDENCE_KEYS:
+        _fail("freshness_invalid")
+    if value["evidence_version"] != EVIDENCE_VERSION or value["policy_version"] != POLICY_VERSION:
+        _fail("freshness_invalid")
+    _string(value["feed_id"], "freshness_invalid")
+    for key in ("source_identity_digest", "body_sha256"):
+        if type(value[key]) is not str or not _DIGEST_RE.fullmatch(value[key]):
+            _fail("freshness_invalid")
+    _timestamp(value["reference_time"], "freshness_invalid")
+    signals = value["signals"]
+    if not isinstance(signals, list) or len(signals) != len(_SIGNAL_KINDS):
+        _fail("freshness_invalid")
+    parsed = [_validate_signal(item) for item in signals]
+    if [item["kind"] for item in parsed] != list(_SIGNAL_KINDS):
+        _fail("freshness_invalid")
+    for item in parsed[:3]:
+        if item["present"] and not item["valid"]:
+            _fail("freshness_invalid")
+    first_present = next((item for item in parsed[:3] if item["present"]), None)
+    selected = value["selected_signal"]
+    if selected is not None:
+        selected = _validate_signal(selected)
+        if selected["kind"] not in _SELECTABLE or not selected["valid"]:
+            _fail("freshness_invalid")
+        matching = next(item for item in parsed if item["kind"] == selected["kind"])
+        if selected != matching:
+            _fail("freshness_invalid")
+    if selected != first_present:
+        _fail("freshness_invalid")
+    if selected is not None:
+        reference = _timestamp(value["reference_time"])
+        selected_time = _timestamp(selected["value"])
+        if selected_time > reference + FUTURE_TOLERANCE or reference - selected_time > MAX_FRESHNESS_AGE:
+            _fail("freshness_invalid")
+    if value["decision"] not in {"eligible", "source_freshness_unverified"}:
+        _fail("freshness_invalid")
+    if (value["decision"] == "eligible") != (selected is not None):
+        _fail("freshness_invalid")
+    return {
+        "evidence_version": value["evidence_version"],
+        "policy_version": value["policy_version"],
+        "feed_id": value["feed_id"],
+        "source_identity_digest": value["source_identity_digest"],
+        "body_sha256": value["body_sha256"],
+        "reference_time": value["reference_time"],
+        "selected_signal": selected,
+        "signals": parsed,
+        "decision": value["decision"],
+    }
+
+
+def _canonical(value: object) -> bytes:
+    try:
+        return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")
+    except (TypeError, ValueError, UnicodeError, RecursionError):
+        _fail("freshness_serialization_invalid")
+
+
+def build_freshness_evidence(
+    *,
+    feed_id: str,
+    source_url: str,
+    body: bytes,
+    response_headers: Mapping[str, str],
+    reference_time: str,
+) -> bytes:
+    if type(body) is not bytes or len(body) > MAX_BODY_BYTES:
+        _fail("freshness_body_invalid")
+    reference = _timestamp(reference_time)
+    headers = _headers(response_headers)
+    atom_present, atom_value = _xml_signal(body, "atom_feed_updated")
+    rss_present, rss_value = _xml_signal(body, "rss_channel_last_build_date")
+    last_modified = headers.get("last-modified")
+    http_present = last_modified is not None
+    http_valid, http_value = _parse_date_signal(last_modified) if http_present else (False, None)
+    date_value = headers.get("date")
+    date_present = date_value is not None
+    date_valid, date_parsed = _parse_date_signal(date_value) if date_present else (False, None)
+    signals = [
+        _signal("atom_feed_updated", atom_present, atom_value, atom_present and atom_value is not None),
+        _signal("rss_channel_last_build_date", rss_present, rss_value, rss_present and rss_value is not None),
+        _signal("http_last_modified", http_present, http_value, http_present and http_valid),
+        _signal("http_date", date_present, date_parsed, date_present and date_valid),
+    ]
+    selected: dict[str, object] | None = None
+    for item in signals[:3]:
+        if item["present"]:
+            if not item["valid"]:
+                _fail("source_freshness_invalid")
+            selected = item
+            break
+    if selected is None:
+        decision = "source_freshness_unverified"
+    else:
+        selected_time = _timestamp(selected["value"])
+        if selected_time > reference + FUTURE_TOLERANCE:
+            _fail("source_freshness_future")
+        if reference - selected_time > MAX_FRESHNESS_AGE:
+            _fail("source_freshness_stale")
+        decision = "eligible"
+    payload = {
+        "evidence_version": EVIDENCE_VERSION,
+        "policy_version": POLICY_VERSION,
+        "feed_id": _string(feed_id, "freshness_invalid"),
+        "source_identity_digest": _source_identity(source_url),
+        "body_sha256": hashlib.sha256(body).hexdigest(),
+        "reference_time": _canonical_timestamp(reference),
+        "selected_signal": selected,
+        "signals": signals,
+        "decision": decision,
+    }
+    wire = _canonical(payload)
+    read_freshness_evidence(wire)
+    return wire
+
+
+def read_freshness_evidence(wire: bytes) -> dict[str, object]:
+    if type(wire) is not bytes:
+        _fail("freshness_wire_invalid")
+
+    def pairs(items: list[tuple[str, object]]) -> dict[str, object]:
+        result: dict[str, object] = {}
+        for key, value in items:
+            if key in result:
+                _fail("freshness_duplicate_key")
+            result[key] = value
+        return result
+
+    try:
+        value = json.loads(wire.decode("utf-8"), object_pairs_hook=pairs)
+    except FreshnessError:
+        raise
+    except (UnicodeError, json.JSONDecodeError, RecursionError):
+        _fail("freshness_wire_invalid")
+    parsed = _validate_evidence(value)
+    if _canonical(parsed) != wire:
+        _fail("freshness_noncanonical")
+    return parsed
diff --git a/tests/test_source_freshness.py b/tests/test_source_freshness.py
new file mode 100644
index 0000000..0f30fc6
--- /dev/null
+++ b/tests/test_source_freshness.py
@@ -0,0 +1,176 @@
+from __future__ import annotations
+
+import json
+from collections import OrderedDict
+
+import pytest
+
+from political_event_tracking_research import rss_source_fetch
+from political_event_tracking_research.source_freshness import (
+    FreshnessError,
+    build_freshness_evidence,
+    read_freshness_evidence,
+)
+
+
+REFERENCE = "2026-07-13T12:00:00Z"
+RSS = b"""<?xml version='1.0'?><rss version='2.0'><channel>
+<lastBuildDate>Sun, 12 Jul 2026 11:00:00 GMT</lastBuildDate>
+</channel></rss>"""
+ATOM = b"""<?xml version='1.0'?><feed xmlns='http://www.w3.org/2005/Atom'>
+<updated>2026-07-13T11:00:00Z</updated></feed>"""
+
+
+def build(body: bytes, headers: dict[str, str] | None = None) -> bytes:
+    return build_freshness_evidence(
+        feed_id="feed-a",
+        source_url="https://example.test/feed.xml",
+        body=body,
+        response_headers=headers or {},
+        reference_time=REFERENCE,
+    )
+
+
+def test_atom_signal_has_priority_over_rss_and_http() -> None:
+    wire = build(
+        ATOM,
+        {
+            "Last-Modified": "Sat, 11 Jul 2026 11:00:00 GMT",
+            "Date": "Mon, 13 Jul 2026 12:01:00 GMT",
+        },
+    )
+    value = read_freshness_evidence(wire)
+    assert value["decision"] == "eligible"
+    assert value["selected_signal"]["kind"] == "atom_feed_updated"
+    assert [item["kind"] for item in value["signals"]] == [
+        "atom_feed_updated",
+        "rss_channel_last_build_date",
+        "http_last_modified",
+        "http_date",
+    ]
+
+
+def test_rss_signal_is_used_when_atom_is_missing() -> None:
+    value = read_freshness_evidence(build(RSS, {"Last-Modified": "Sat, 11 Jul 2026 11:00:00 GMT"}))
+    assert value["selected_signal"]["kind"] == "rss_channel_last_build_date"
+
+
+def test_http_last_modified_is_last_fallback() -> None:
+    body = b"<rss version='2.0'><channel/></rss>"
+    value = read_freshness_evidence(build(body, {"Last-Modified": "Sun, 12 Jul 2026 11:00:00 GMT"}))
+    assert value["selected_signal"]["kind"] == "http_last_modified"
+    assert value["signals"][0]["present"] is False
+
+
+def test_http_date_alone_does_not_prove_content_freshness() -> None:
+    wire = build(
+        RSS.replace(b"<lastBuildDate>Sun, 12 Jul 2026 11:00:00 GMT</lastBuildDate>", b""),
+        {"Date": "Mon, 13 Jul 2026 12:00:00 GMT"},
+    )
+    assert read_freshness_evidence(wire)["decision"] == "source_freshness_unverified"
+
+
+def test_invalid_high_priority_signal_does_not_fallback() -> None:
+    body = ATOM.replace(b"2026-07-13T11:00:00Z", b"not-a-time")
+    with pytest.raises(FreshnessError, match="source_freshness_invalid"):
+        build(body, {"Last-Modified": "Sun, 12 Jul 2026 11:00:00 GMT"})
+
+
+def test_present_but_empty_high_priority_signal_does_not_fallback() -> None:
+    body = ATOM.replace(b"2026-07-13T11:00:00Z", b"")
+    with pytest.raises(FreshnessError, match="source_freshness_invalid"):
+        build(body, {"Last-Modified": "Sun, 12 Jul 2026 11:00:00 GMT"})
+
+
+def test_timezone_less_signal_is_invalid() -> None:
+    body = ATOM.replace(b"2026-07-13T11:00:00Z", b"2026-07-13T11:00:00")
+    with pytest.raises(FreshnessError, match="source_freshness_invalid"):
+        build(body, {"Last-Modified": "Sun, 12 Jul 2026 11:00:00 GMT"})
+
+
+@pytest.mark.parametrize(
+    "body,headers,error",
+    [
+        (RSS.replace(b"12 Jul 2026", b"01 Jul 2026"), {}, "source_freshness_stale"),
+        (ATOM.replace(b"2026-07-13T11:00:00Z", b"2026-07-13T12:06:00Z"), {}, "source_freshness_future"),
+    ],
+)
+def test_stale_and_future_signals_fail_closed(body: bytes, headers: dict[str, str], error: str) -> None:
+    with pytest.raises(FreshnessError, match=error):
+        build(body, headers)
+
+
+def test_zero_entry_body_can_be_fresh_and_is_not_inferred_stale() -> None:
+    value = read_freshness_evidence(build(RSS))
+    assert value["decision"] == "eligible"
+
+
+def test_evidence_is_canonical_and_tamper_resistant() -> None:
+    wire = build(RSS)
+    assert json.dumps(json.loads(wire), sort_keys=True, separators=(",", ":")).encode() == wire
+    assert read_freshness_evidence(wire)["body_sha256"]
+    with pytest.raises(FreshnessError, match="freshness_noncanonical"):
+        read_freshness_evidence(json.dumps(json.loads(wire), indent=2).encode())
+    with pytest.raises(FreshnessError, match="freshness_invalid"):
+        read_freshness_evidence(wire.replace(b"eligible", b"tampered", 1))
+
+
+def test_duplicate_and_non_mapping_headers_fail_closed() -> None:
+    with pytest.raises(FreshnessError, match="freshness_headers_invalid"):
+        build(RSS, OrderedDict((("Date", "x"), ("date", "y"))))
+
+
+def test_fetch_boundary_returns_body_and_only_freshness_headers() -> None:
+    class Headers:
+        def get(self, name: str) -> str | None:
+            return {"Date": "Mon, 13 Jul 2026 12:00:00 GMT", "Last-Modified": "Sun, 12 Jul 2026 11:00:00 GMT"}.get(name)
+
+    class Response:
+        headers = Headers()
+
+        def __enter__(self) -> "Response":
+            return self
+
+        def __exit__(self, *_args: object) -> bool:
+            return False
+
+        def read(self, _size: int) -> bytes:
+            return RSS
+
+    original = rss_source_fetch.urllib.request.urlopen
+    rss_source_fetch.urllib.request.urlopen = lambda *_args, **_kwargs: Response()  # type: ignore[assignment]
+    try:
+        body, headers = rss_source_fetch.fetch_url_with_metadata("https://example.test/feed.xml")
+    finally:
+        rss_source_fetch.urllib.request.urlopen = original
+    assert body == RSS
+    assert headers == {"Date": "Mon, 13 Jul 2026 12:00:00 GMT", "Last-Modified": "Sun, 12 Jul 2026 11:00:00 GMT"}
+
+
+def test_fetch_boundary_rejects_duplicate_freshness_header() -> None:
+    class Headers:
+        def get_all(self, name: str) -> list[str]:
+            return ["Sun, 12 Jul 2026 11:00:00 GMT", "Sun, 12 Jul 2026 12:00:00 GMT"] if name == "Last-Modified" else []
+
+        def get(self, _name: str) -> str | None:
+            return None
+
+    class Response:
+        headers = Headers()
+
+        def __enter__(self) -> "Response":
+            return self
+
+        def __exit__(self, *_args: object) -> bool:
+            return False
+
+        def read(self, _size: int) -> bytes:
+            return RSS
+
+    original = rss_source_fetch.urllib.request.urlopen
+    rss_source_fetch.urllib.request.urlopen = lambda *_args, **_kwargs: Response()  # type: ignore[assignment]
+    try:
+        with pytest.raises(ValueError, match="feed_header_duplicate"):
+            rss_source_fetch.fetch_url_with_metadata("https://example.test/feed.xml")
+    finally:
+        rss_source_fetch.urllib.request.urlopen = original