diff --git a/airbyte_cdk/entrypoint.py b/airbyte_cdk/entrypoint.py
index 57820f005..777126f96 100644
--- a/airbyte_cdk/entrypoint.py
+++ b/airbyte_cdk/entrypoint.py
@@ -35,11 +35,16 @@
 )
 from airbyte_cdk.sources import Source
 from airbyte_cdk.sources.connector_state_manager import HashableStreamDescriptor
+from airbyte_cdk.sources.streams.http.cache_stats import (
+    HTTP_CACHE_STATS,
+    HttpCacheStatsSnapshot,
+)
 from airbyte_cdk.sources.utils.schema_helpers import check_config_against_spec_or_exit, split_config
 
 # from airbyte_cdk.utils import PrintBuffer, is_cloud_environment, message_utils  # add PrintBuffer back once fixed
 from airbyte_cdk.utils import is_cloud_environment, message_utils
 from airbyte_cdk.utils.airbyte_secrets_utils import get_secrets, update_secrets
+from airbyte_cdk.utils.analytics_message import create_analytics_message
 from airbyte_cdk.utils.constants import ENV_REQUEST_CACHE_PATH
 from airbyte_cdk.utils.memory_monitor import MemoryMonitor
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
@@ -168,6 +173,12 @@ def run(self, parsed_args: argparse.Namespace) -> Iterable[str]:
             self.logger.setLevel(logging.INFO)
 
         source_spec: ConnectorSpecification = self.source.spec(self.logger)
+        # The counters are process-wide, and a process can run more than one command
+        # (`entrypoint_wrapper` in connector tests, the manifest server). Reporting a
+        # delta against this baseline keeps every run's numbers its own; resetting
+        # instead would race whichever command the manifest server is serving next to it.
+        http_cache_stats_baseline = HTTP_CACHE_STATS.snapshot()
+        closing = False
         try:
             with tempfile.TemporaryDirectory(
                 # Cleanup can fail on Windows due to file locks. Ignore if so,
@@ -212,11 +223,51 @@ def run(self, parsed_args: argparse.Namespace) -> Iterable[str]:
                         )
                     else:
                         raise Exception("Unexpected command " + cmd)
+        except GeneratorExit:
+            # The consumer is closing us -- yielding anything now would turn into
+            # `RuntimeError: generator ignored GeneratorExit`, which the interpreter
+            # reports on stderr on top of whatever really went wrong.
+            closing = True
+            raise
         finally:
             yield from [
                 self.airbyte_message_to_string(queued_message)
                 for queued_message in self._emit_queued_messages(self.source)
             ]
+            if not closing:
+                yield from map(
+                    AirbyteEntrypoint.airbyte_message_to_string,
+                    self._http_cache_stats_messages(http_cache_stats_baseline),
+                )
+
+    @staticmethod
+    def _http_cache_stats_messages(
+        baseline: HttpCacheStatsSnapshot,
+    ) -> Iterable[AirbyteMessage]:
+        """Report how many requests the run made and how many its cache served.
+
+        A `requests_cache` hit never reaches the wire, so this is the only place
+        it can be observed from outside the process. Emitted as analytics, which
+        ride the protocol as TRACE messages on stdout: no `LOG_LEVEL=DEBUG` to
+        turn on, and one message pair per run rather than per-request log spam.
+
+        Reported as a delta against `baseline`, the snapshot taken when this run
+        started, so that a process running several commands attributes each
+        request to the run that made it rather than to every run after it.
+
+        Silent when the run made no requests, so `spec` -- and every non-HTTP
+        connector -- does not report a meaningless `0`. That silence is load
+        bearing for readers: absent means *not measured*, which is also what a
+        connector on an older CDK looks like, and is not the same as `0%`.
+        """
+        stats = HTTP_CACHE_STATS.snapshot()
+        requests_made = stats.requests - baseline.requests
+        if requests_made <= 0:
+            return
+        yield create_analytics_message("http-request-count", requests_made)
+        yield create_analytics_message(
+            "http-cache-hit-count", stats.cache_hits - baseline.cache_hits
+        )
 
     def check(
         self, source_spec: ConnectorSpecification, config: TConfig
diff --git a/airbyte_cdk/sources/streams/http/cache_stats.py b/airbyte_cdk/sources/streams/http/cache_stats.py
new file mode 100644
index 000000000..fd79d2bc7
--- /dev/null
+++ b/airbyte_cdk/sources/streams/http/cache_stats.py
@@ -0,0 +1,83 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+"""Process-wide counters for HTTP requests and `requests_cache` hits.
+
+A `requests_cache` hit is served inside `Session.send()` and never reaches the
+wire, so nothing outside the connector process can observe it -- not a proxy, not
+the platform. The connector therefore has to count its own hits and report them,
+which is what these counters exist for; `AirbyteEntrypoint.run` turns the final
+snapshot into analytics messages at the end of every command.
+
+The counters are module-level because the thing being measured is, too: one
+`requests_cache` backend is shared by every stream of a run, and the question the
+numbers answer ("does this connector's caching work, and did this version
+regress it") is about the run rather than about any one stream or client. Being
+process-wide, they are cumulative: a reader wanting one run's numbers takes a
+snapshot when it starts and subtracts, which is what `AirbyteEntrypoint.run` does.
+
+Scope: every `HttpClient` in the process records here, but only
+`AirbyteEntrypoint.run` reports. Destinations (`Destination.run_cmd`), the
+manifest server, and the Connector Builder drive their commands without going
+through it, so they accumulate counts nothing reads. "No counts reported" means
+"a source connector run through the entrypoint, or nothing".
+
+`http-request-count` includes cache hits, so it is responses handled rather than
+wire flows; subtract `http-cache-hit-count` to get the number a proxy would see.
+"""
+
+from __future__ import annotations
+
+import threading
+from dataclasses import dataclass
+
+import requests
+
+
+@dataclass(frozen=True)
+class HttpCacheStatsSnapshot:
+    """The counters at one instant, detached from the lock that guards them."""
+
+    requests: int
+    cache_hits: int
+
+
+class HttpCacheStats:
+    """Requests made and requests served from the connector's own cache.
+
+    Guarded by a lock because concurrent sources read streams on a thread pool,
+    so `record_response` is called from several threads at once and `+= 1` on a
+    plain attribute would drop counts.
+    """
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._requests = 0
+        self._cache_hits = 0
+
+    def record_response(self, response: requests.Response) -> None:
+        """Count one request, and one cache hit when the response was cached.
+
+        `from_cache` is set by `requests_cache.CacheMixin`, so it is present
+        exactly when caching is in play and absent -- counted as a live request
+        -- when it is not.
+        """
+        from_cache = bool(getattr(response, "from_cache", False))
+        with self._lock:
+            self._requests += 1
+            if from_cache:
+                self._cache_hits += 1
+
+    def snapshot(self) -> HttpCacheStatsSnapshot:
+        with self._lock:
+            return HttpCacheStatsSnapshot(requests=self._requests, cache_hits=self._cache_hits)
+
+    def reset(self) -> None:
+        """Zero the counters. For tests, which share one process across cases."""
+        with self._lock:
+            self._requests = 0
+            self._cache_hits = 0
+
+
+HTTP_CACHE_STATS = HttpCacheStats()
diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py
index c1d0eabd6..6775c6bec 100644
--- a/airbyte_cdk/sources/streams/http/http_client.py
+++ b/airbyte_cdk/sources/streams/http/http_client.py
@@ -25,6 +25,7 @@
 from airbyte_cdk.sources.http_config import MAX_CONNECTION_POOL_SIZE
 from airbyte_cdk.sources.message import MessageRepository
 from airbyte_cdk.sources.streams.call_rate import APIBudget, CachedLimiterSession, LimiterSession
+from airbyte_cdk.sources.streams.http.cache_stats import HTTP_CACHE_STATS
 from airbyte_cdk.sources.streams.http.error_handlers import (
     BackoffStrategy,
     DefaultBackoffStrategy,
@@ -349,6 +350,13 @@ def _send(
         except requests.RequestException as e:
             exc = e
 
+        if response is not None:
+            # Counted per `_send` call, so a retried request counts once per
+            # attempt. Cache hits are counted here too, so this is responses
+            # handled rather than wire flows: the figure comparable to what a
+            # proxy sees is `requests - cache_hits`, not `requests`.
+            HTTP_CACHE_STATS.record_response(response)
+
         error_resolution: ErrorResolution = self._error_handler.interpret_response(
             response if response is not None else exc
         )
diff --git a/cdk-migrations.md b/cdk-migrations.md
index 3445a04b8..2fffb518d 100644
--- a/cdk-migrations.md
+++ b/cdk-migrations.md
@@ -1,5 +1,23 @@
 # CDK Migration Guide
 
+## Upgrading to the CDK that reports HTTP cache stats
+
+Every source that makes at least one HTTP request through `HttpClient` now ends each
+`spec`/`check`/`discover`/`read` with two extra `TRACE`/`ANALYTICS` messages:
+`http-request-count` and `http-cache-hit-count`. A run that made no requests emits
+nothing, so absent still means *not measured* rather than `0`.
+
+Migration steps: connector tests that assert an exact protocol message count need
+updating. `EntrypointOutput.trace_messages` grows by two for any run that made a
+request, including runs under `HttpMocker`, which leaves `Session.send` in the call
+path. Assertions of the shape `assert len(output.trace_messages) == N` are the ones
+that break; `> 0` and stream-status filters are unaffected. Nothing else changes --
+no records, state, schemas, or exit codes.
+
+Note on the numbers themselves: `http-request-count` counts responses handled, cache
+hits included, so the figure comparable to a proxy's wire-flow count is
+`http-request-count - http-cache-hit-count`.
+
 ## Upgrading to 7.0.0
 
 [Version 7.0.0](https://github.com/airbytehq/airbyte-python-cdk/releases/tag/v7.0.0) of the CDK migrates the CDK to the Concurrent CDK by removing some of the Declarative CDK concepts that are better expressed in the Concurrent CDK or that are outright incompatible with it. This changes mostly impact the Python implementations although the concept of CustomIncrementalSync has been removed from the declarative language as well.
diff --git a/unit_tests/sources/streams/http/test_cache_stats.py b/unit_tests/sources/streams/http/test_cache_stats.py
new file mode 100644
index 000000000..3e500f462
--- /dev/null
+++ b/unit_tests/sources/streams/http/test_cache_stats.py
@@ -0,0 +1,79 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+import threading
+
+import requests
+
+from airbyte_cdk.sources.streams.http.cache_stats import HttpCacheStats, HttpCacheStatsSnapshot
+
+
+def _response(*, from_cache: bool | None = None) -> requests.Response:
+    """A response with, or deliberately without, the `requests_cache` marker.
+
+    `from_cache` absent is the case that matters most: it is what an uncached
+    session produces, and counting it as a hit would report every connector that
+    does no caching as caching perfectly.
+    """
+    response = requests.Response()
+    response.status_code = 200
+    if from_cache is not None:
+        response.from_cache = from_cache  # type: ignore[attr-defined]
+    return response
+
+
+def test_a_fresh_counter_reports_nothing() -> None:
+    assert HttpCacheStats().snapshot() == HttpCacheStatsSnapshot(requests=0, cache_hits=0)
+
+
+def test_only_a_from_cache_response_counts_as_a_hit() -> None:
+    stats = HttpCacheStats()
+
+    stats.record_response(_response())
+    stats.record_response(_response(from_cache=False))
+    stats.record_response(_response(from_cache=True))
+
+    assert stats.snapshot() == HttpCacheStatsSnapshot(requests=3, cache_hits=1)
+
+
+def test_a_snapshot_does_not_move_under_the_reader() -> None:
+    """The snapshot is a detached value, so a later request cannot backdate it."""
+    stats = HttpCacheStats()
+    stats.record_response(_response(from_cache=True))
+
+    taken = stats.snapshot()
+    stats.record_response(_response(from_cache=True))
+
+    assert taken == HttpCacheStatsSnapshot(requests=1, cache_hits=1)
+
+
+def test_concurrent_recording_loses_no_counts() -> None:
+    """Concurrent sources read streams on a thread pool, so this is the real shape.
+
+    `+= 1` on a plain attribute is not atomic under free-threaded CPython and is
+    only accidentally so under the GIL, which is why the counters take a lock.
+    """
+    stats = HttpCacheStats()
+    threads = [
+        threading.Thread(
+            target=lambda: [stats.record_response(_response(from_cache=True)) for _ in range(200)]
+        )
+        for _ in range(8)
+    ]
+
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert stats.snapshot() == HttpCacheStatsSnapshot(requests=1600, cache_hits=1600)
+
+
+def test_reset_zeroes_the_counters() -> None:
+    stats = HttpCacheStats()
+    stats.record_response(_response(from_cache=True))
+
+    stats.reset()
+
+    assert stats.snapshot() == HttpCacheStatsSnapshot(requests=0, cache_hits=0)
diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py
index 48d396cb6..eee2fb357 100644
--- a/unit_tests/sources/streams/http/test_http_client.py
+++ b/unit_tests/sources/streams/http/test_http_client.py
@@ -13,6 +13,10 @@
 from airbyte_cdk.models import FailureType
 from airbyte_cdk.sources.streams.call_rate import CachedLimiterSession, LimiterSession
 from airbyte_cdk.sources.streams.http import HttpClient
+from airbyte_cdk.sources.streams.http.cache_stats import (
+    HTTP_CACHE_STATS,
+    HttpCacheStatsSnapshot,
+)
 from airbyte_cdk.sources.streams.http.error_handlers import (
     BackoffStrategy,
     ErrorResolution,
@@ -461,6 +465,8 @@ def test_that_response_was_cached(requests_mock):
 
     requests_mock.register_uri("GET", "https://google.com/", json='{"test": "response"}')
 
+    before = HTTP_CACHE_STATS.snapshot()
+
     cached_http_client._send(prepared_request, {})
 
     assert requests_mock.called
@@ -471,6 +477,93 @@ def test_that_response_was_cached(requests_mock):
     assert isinstance(second_response.request, CachedRequest)
     assert not requests_mock.called
 
+    # Pins the counters to the real `requests_cache` marker rather than to a
+    # hand-set attribute: `getattr(response, "from_cache", False)` cannot tell
+    # "not cached" from "upstream renamed it", so without a hit recorded through
+    # a live CachedLimiterSession the metric could read 0% forever, green.
+    after = HTTP_CACHE_STATS.snapshot()
+    assert after.requests - before.requests == 2
+    assert after.cache_hits - before.cache_hits == 1
+
+
+def test_send_counts_a_cache_hit_only_for_the_response_served_from_cache():
+    """The connector's own cache is invisible from outside the process.
+
+    A `requests_cache` hit is served inside `Session.send()` and never reaches
+    the wire, so a proxy cannot see it and neither can the platform. These
+    counters are the only place it is observable, which is what the regression
+    report's connector cache-hit ratio is built on -- so a live response must
+    count as a request and not a hit, and a cached one as both.
+
+    `from_cache` is the marker `requests_cache.CacheMixin` sets on a response it
+    served, which is why the session is stubbed rather than really cached: what
+    is under test is the counting, not `requests_cache` itself.
+    """
+    HTTP_CACHE_STATS.reset()
+
+    live = requests.Response()
+    live.status_code = 200
+    cached = requests.Response()
+    cached.status_code = 200
+    cached.from_cache = True  # type: ignore[attr-defined] # set by requests_cache.CacheMixin
+
+    mocked_session = MagicMock(spec=requests.Session)
+    mocked_session.send.side_effect = [live, cached]
+    http_client = HttpClient(name="test", logger=MagicMock(), session=mocked_session)
+
+    http_client._send(requests.PreparedRequest(), {})
+
+    assert HTTP_CACHE_STATS.snapshot() == HttpCacheStatsSnapshot(requests=1, cache_hits=0)
+
+    http_client._send(requests.PreparedRequest(), {})
+
+    assert HTTP_CACHE_STATS.snapshot() == HttpCacheStatsSnapshot(requests=2, cache_hits=1)
+
+
+def test_send_counts_requests_without_caching_as_never_cached():
+    """A session with no cache has no `from_cache`, and reports no hits.
+
+    This is the shape most connectors have, and the number that must not read as
+    a cache that is failing: requests counted, zero hits.
+    """
+    HTTP_CACHE_STATS.reset()
+
+    response = requests.Response()
+    response.status_code = 200
+    mocked_session = MagicMock(spec=requests.Session)
+    mocked_session.send.return_value = response
+    http_client = HttpClient(name="test", logger=MagicMock(), session=mocked_session)
+
+    http_client._send(requests.PreparedRequest(), {})
+    http_client._send(requests.PreparedRequest(), {})
+
+    assert HTTP_CACHE_STATS.snapshot() == HttpCacheStatsSnapshot(requests=2, cache_hits=0)
+
+
+def test_send_counts_nothing_when_the_request_never_produced_a_response():
+    """No response means no wire flow to count, the way a proxy would see it."""
+    HTTP_CACHE_STATS.reset()
+    mocked_session = MagicMock(spec=requests.Session)
+    mocked_session.send.side_effect = requests.RequestException
+
+    http_client = HttpClient(
+        name="test",
+        logger=MagicMock(),
+        error_handler=HttpStatusErrorHandler(
+            logger=MagicMock(),
+            error_mapping={
+                requests.RequestException: ErrorResolution(
+                    ResponseAction.IGNORE, FailureType.system_error, "ignored"
+                )
+            },
+        ),
+        session=mocked_session,
+    )
+
+    http_client._send(requests.PreparedRequest(), {})
+
+    assert HTTP_CACHE_STATS.snapshot() == HttpCacheStatsSnapshot(requests=0, cache_hits=0)
+
 
 def test_send_handles_response_action_given_session_send_raises_request_exception():
     error_resolution = ErrorResolution(
diff --git a/unit_tests/test_entrypoint.py b/unit_tests/test_entrypoint.py
index fcfb44915..e55e5c58c 100644
--- a/unit_tests/test_entrypoint.py
+++ b/unit_tests/test_entrypoint.py
@@ -45,6 +45,7 @@
 )
 from airbyte_cdk.sources import Source
 from airbyte_cdk.sources.connector_state_manager import HashableStreamDescriptor
+from airbyte_cdk.sources.streams.http.cache_stats import HTTP_CACHE_STATS
 from airbyte_cdk.utils import AirbyteTracedException
 
 
@@ -489,6 +490,217 @@ def test_run_read(entrypoint: AirbyteEntrypoint, mocker, spec_mock, config_mock)
     assert spec_mock.called
 
 
+def _analytics_values(messages: List[str]) -> Mapping[str, str]:
+    """The analytics counters a run reported, keyed by type."""
+    parsed = [orjson.loads(message) for message in messages]
+    return {
+        message["trace"]["analytics"]["type"]: message["trace"]["analytics"]["value"]
+        for message in parsed
+        if message["type"] == "TRACE" and message["trace"]["type"] == "ANALYTICS"
+    }
+
+
+def test_read_reports_the_run_http_request_and_cache_hit_counts(
+    entrypoint: AirbyteEntrypoint, mocker, spec_mock, config_mock
+):
+    """A `requests_cache` hit never reaches the wire, so the run has to say so.
+
+    Analytics rather than logs: they ride the protocol as TRACE messages on
+    stdout, so a harness reads them without `LOG_LEVEL=DEBUG` and without one log
+    line per request.
+    """
+    parsed_args = Namespace(
+        command="read", config="config_path", state="statepath", catalog="catalogpath"
+    )
+    mocker.patch.object(MockSource, "read_state", return_value={})
+    mocker.patch.object(MockSource, "read_catalog", return_value={})
+
+    def read_making_requests(*args, **kwargs):
+        live = requests.Response()
+        live.status_code = 200
+        cached = requests.Response()
+        cached.status_code = 200
+        cached.from_cache = True
+        HTTP_CACHE_STATS.record_response(live)
+        HTTP_CACHE_STATS.record_response(cached)
+        HTTP_CACHE_STATS.record_response(cached)
+        return []
+
+    mocker.patch.object(MockSource, "read", side_effect=read_making_requests)
+
+    messages = list(entrypoint.run(parsed_args))
+
+    assert _analytics_values(messages) == {
+        "http-request-count": "3",
+        "http-cache-hit-count": "2",
+    }
+
+
+def test_read_reports_the_counts_even_when_it_fails(
+    entrypoint: AirbyteEntrypoint, mocker, spec_mock, config_mock
+):
+    """A crashed run is exactly the one whose request count a reviewer wants."""
+    parsed_args = Namespace(
+        command="read", config="config_path", state="statepath", catalog="catalogpath"
+    )
+    mocker.patch.object(MockSource, "read_state", return_value={})
+    mocker.patch.object(MockSource, "read_catalog", return_value={})
+
+    def read_then_fail(*args, **kwargs):
+        response = requests.Response()
+        response.status_code = 200
+        HTTP_CACHE_STATS.record_response(response)
+        raise ValueError("Any error")
+
+    mocker.patch.object(MockSource, "read", side_effect=read_then_fail)
+
+    messages = []
+    with pytest.raises(ValueError):
+        messages.extend(entrypoint.run(parsed_args))
+
+    assert _analytics_values(messages) == {
+        "http-request-count": "1",
+        "http-cache-hit-count": "0",
+    }
+
+
+def test_a_run_that_made_no_requests_reports_no_counts(
+    entrypoint: AirbyteEntrypoint, mocker, spec_mock, config_mock
+):
+    """Silence, not `0` -- absent has to mean *not measured*.
+
+    A connector on a CDK without these counters, or one that never calls out,
+    both report nothing; a `0` here would be indistinguishable from a measured
+    zero and would render as a `0%` cache ratio in a regression report.
+    """
+    parsed_args = Namespace(
+        command="read", config="config_path", state="statepath", catalog="catalogpath"
+    )
+    mocker.patch.object(MockSource, "read_state", return_value={})
+    mocker.patch.object(MockSource, "read_catalog", return_value={})
+    mocker.patch.object(MockSource, "read", return_value=[])
+
+    messages = list(entrypoint.run(parsed_args))
+
+    assert _analytics_values(messages) == {}
+
+
+@pytest.fixture
+def multi_run_entrypoint(mocker) -> AirbyteEntrypoint:
+    """An entrypoint that can be run more than once in a test.
+
+    The shared `entrypoint` fixture hands out a fixed list of queue reads, which a
+    test driving several commands through one process exhausts.
+    """
+    message_repository = MagicMock()
+    message_repository.consume_queue.return_value = []
+    mocker.patch.object(
+        MockSource,
+        "message_repository",
+        new_callable=mocker.PropertyMock,
+        return_value=message_repository,
+    )
+    return AirbyteEntrypoint(MockSource())
+
+
+def _read_recording(*, requests_made: int, cache_hits: int = 0):
+    """A `read` that records `requests_made` responses, `cache_hits` of them cached."""
+
+    def read(*args, **kwargs):
+        for index in range(requests_made):
+            response = requests.Response()
+            response.status_code = 200
+            if index < cache_hits:
+                response.from_cache = True  # set by requests_cache.CacheMixin
+            HTTP_CACHE_STATS.record_response(response)
+        return []
+
+    return read
+
+
+def test_a_second_run_in_the_same_process_reports_only_its_own_requests(
+    multi_run_entrypoint: AirbyteEntrypoint, mocker, spec_mock, config_mock
+):
+    """Connector suites drive the entrypoint in-process, run after run.
+
+    The counters are process-wide, so reporting the absolute snapshot would hand
+    the second run the sum of both, and make every connector's numbers depend on
+    what pytest happened to run before them.
+    """
+    parsed_args = Namespace(
+        command="read", config="config_path", state="statepath", catalog="catalogpath"
+    )
+    mocker.patch.object(MockSource, "read_state", return_value={})
+    mocker.patch.object(MockSource, "read_catalog", return_value={})
+
+    mocker.patch.object(MockSource, "read", side_effect=_read_recording(requests_made=1))
+    first = _analytics_values(list(multi_run_entrypoint.run(parsed_args)))
+
+    mocker.patch.object(
+        MockSource, "read", side_effect=_read_recording(requests_made=2, cache_hits=1)
+    )
+    second = _analytics_values(list(multi_run_entrypoint.run(parsed_args)))
+
+    assert first == {"http-request-count": "1", "http-cache-hit-count": "0"}
+    assert second == {"http-request-count": "2", "http-cache-hit-count": "1"}
+
+
+def test_a_zero_request_run_stays_silent_after_a_run_that_made_requests(
+    multi_run_entrypoint: AirbyteEntrypoint, mocker, spec_mock, config_mock
+):
+    """The "absent means not measured" contract has to survive an earlier run.
+
+    Reporting the absolute snapshot would make this run claim the previous run's
+    requests as its own, which is the one output the design says is meaningless.
+    """
+    parsed_args = Namespace(
+        command="read", config="config_path", state="statepath", catalog="catalogpath"
+    )
+    mocker.patch.object(MockSource, "read_state", return_value={})
+    mocker.patch.object(MockSource, "read_catalog", return_value={})
+
+    mocker.patch.object(MockSource, "read", side_effect=_read_recording(requests_made=1))
+    list(multi_run_entrypoint.run(parsed_args))
+
+    mocker.patch.object(MockSource, "read", side_effect=_read_recording(requests_made=0))
+    messages = list(multi_run_entrypoint.run(parsed_args))
+
+    assert _analytics_values(messages) == {}
+
+
+def test_abandoning_the_run_generator_does_not_raise(
+    multi_run_entrypoint: AirbyteEntrypoint, mocker, spec_mock, config_mock
+):
+    """`launch()` abandons this generator whenever writing to stdout breaks.
+
+    A `yield` reached while the generator is closing raises `RuntimeError:
+    generator ignored GeneratorExit`, which the interpreter prints to stderr and
+    the platform ingests as a log line on top of the real failure.
+    """
+    parsed_args = Namespace(
+        command="read", config="config_path", state="statepath", catalog="catalogpath"
+    )
+    mocker.patch.object(MockSource, "read_state", return_value={})
+    mocker.patch.object(MockSource, "read_catalog", return_value={})
+
+    def read_forever(*args, **kwargs):
+        response = requests.Response()
+        response.status_code = 200
+        HTTP_CACHE_STATS.record_response(response)
+        while True:
+            yield AirbyteMessage(
+                type=Type.RECORD,
+                record=AirbyteRecordMessage(stream="stream", data={}, emitted_at=1),
+            )
+
+    mocker.patch.object(MockSource, "read", side_effect=read_forever)
+
+    messages = multi_run_entrypoint.run(parsed_args)
+    next(messages)
+
+    messages.close()
+
+
 def test_given_message_emitted_during_config_when_read_then_emit_message_before_next_steps(
     entrypoint: AirbyteEntrypoint, mocker, spec_mock, config_mock
 ):