From 6392be07e92d20e2771b7e5abf57571209b95cb8 Mon Sep 17 00:00:00 2001
From: Anatolii Yatsuk <tolikyatsuk@gmail.com>
Date: Fri, 7 Aug 2026 12:37:09 +0300
Subject: [PATCH 1/2] feat(http): report per-run HTTP request and cache-hit
 counts as analytics

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. That makes "does this connector's caching work, and
did this version regress it" a question only the connector can answer.

Adds process-wide counters in `sources/streams/http/cache_stats.py`, guarded
by a lock because concurrent sources read streams on a thread pool.
`HttpClient._send` records one request per attempt, plus a hit when the
response carries `from_cache` -- the marker `requests_cache.CacheMixin` sets,
present exactly when caching is in play.

`AirbyteEntrypoint.run` emits the totals from its `finally` block, so a
crashed run still reports: `http-request-count` and `http-cache-hit-count`.
Analytics rather than logs, so a reader needs no `LOG_LEVEL=DEBUG` and gets
one message pair per run instead of per-request spam.

Silent when the run made no requests. That silence is load bearing: absent
means *not measured*, which is also what a Java connector, a connector using
its own `requests.Session`, and a connector on an older CDK all look like.
Consumers must render those as "not reported", never as `0%` -- a measured
zero is a broken cache, an absent one is a cache nobody looked at.

Prerequisite for airbytehq/airbyte-internal-issues#16840, which surfaces the
ratio in the connector regression-test report.
---
 airbyte_cdk/entrypoint.py                     | 26 +++++
 .../sources/streams/http/cache_stats.py       | 72 ++++++++++++++
 .../sources/streams/http/http_client.py       |  7 ++
 unit_tests/conftest.py                        | 16 ++++
 .../sources/streams/http/test_cache_stats.py  | 79 +++++++++++++++
 .../sources/streams/http/test_http_client.py  | 83 ++++++++++++++++
 unit_tests/test_entrypoint.py                 | 96 +++++++++++++++++++
 7 files changed, 379 insertions(+)
 create mode 100644 airbyte_cdk/sources/streams/http/cache_stats.py
 create mode 100644 unit_tests/sources/streams/http/test_cache_stats.py

diff --git a/airbyte_cdk/entrypoint.py b/airbyte_cdk/entrypoint.py
index 57820f0053..048d667ca3 100644
--- a/airbyte_cdk/entrypoint.py
+++ b/airbyte_cdk/entrypoint.py
@@ -35,11 +35,13 @@
 )
 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.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
@@ -217,6 +219,30 @@ def run(self, parsed_args: argparse.Namespace) -> Iterable[str]:
                 self.airbyte_message_to_string(queued_message)
                 for queued_message in self._emit_queued_messages(self.source)
             ]
+            yield from map(
+                AirbyteEntrypoint.airbyte_message_to_string,
+                self._http_cache_stats_messages(),
+            )
+
+    @staticmethod
+    def _http_cache_stats_messages() -> 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.
+
+        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()
+        if stats.requests <= 0:
+            return
+        yield create_analytics_message("http-request-count", stats.requests)
+        yield create_analytics_message("http-cache-hit-count", stats.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 0000000000..2b62517115
--- /dev/null
+++ b/airbyte_cdk/sources/streams/http/cache_stats.py
@@ -0,0 +1,72 @@
+#
+# 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.
+"""
+
+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 c1d0eabd67..4d1aad4a54 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,12 @@ 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 -- the same way a proxy counts wire flows, which is what
+            # makes the two numbers comparable in a regression report.
+            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/unit_tests/conftest.py b/unit_tests/conftest.py
index 3a21552bc8..d5a10e91b1 100644
--- a/unit_tests/conftest.py
+++ b/unit_tests/conftest.py
@@ -7,6 +7,22 @@
 import freezegun
 import pytest
 
+from airbyte_cdk.sources.streams.http.cache_stats import HTTP_CACHE_STATS
+
+
+@pytest.fixture(autouse=True)
+def reset_http_cache_stats():
+    """Isolate the process-wide HTTP counters between tests.
+
+    They are module-level by design -- one `requests_cache` backend is shared by
+    a whole run -- but a test process is many runs, and a leftover count makes
+    `AirbyteEntrypoint.run` emit analytics messages in a test that recorded no
+    requests of its own.
+    """
+    HTTP_CACHE_STATS.reset()
+    yield
+    HTTP_CACHE_STATS.reset()
+
 
 @pytest.fixture()
 def mock_sleep(monkeypatch):
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 0000000000..3e500f4629
--- /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 48d396cb63..421edb0aea 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,
@@ -472,6 +476,85 @@ def test_that_response_was_cached(requests_mock):
     assert not requests_mock.called
 
 
+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(
         ResponseAction.FAIL, FailureType.system_error, "test fail message"
diff --git a/unit_tests/test_entrypoint.py b/unit_tests/test_entrypoint.py
index fcfb449151..458f6db762 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,101 @@ 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) == {}
+
+
 def test_given_message_emitted_during_config_when_read_then_emit_message_before_next_steps(
     entrypoint: AirbyteEntrypoint, mocker, spec_mock, config_mock
 ):

From 9a2f1ad35d231e9d8d84fd7abc4ab76208f93f63 Mon Sep 17 00:00:00 2001
From: Anatolii Yatsuk <tolikyatsuk@gmail.com>
Date: Thu, 20 Aug 2026 15:22:12 +0300
Subject: [PATCH 2/2] fix(http): address review findings on HTTP cache stats
 reporting

Report the counts as a delta against a baseline snapshot taken when the run
starts, rather than as the absolute process-wide total. A process can execute
more than one command -- `entrypoint_wrapper` in connector suites, the manifest
server -- and the absolute snapshot made each run report the sum of every run
before it, so a run that made no requests at all could still emit a non-zero
count. That voided the "absent means not measured" contract the feature is
built on. A delta keeps each run's numbers its own without a reset, which would
race concurrent manifest-server commands.

Removing the autouse reset fixture from `unit_tests/conftest.py` along with it:
it existed only to hide this gap inside the CDK's own suite, and connector
repositories never had it.

Also:
- Skip the emission when the generator is being closed. Yielding from the
  `finally` during `GeneratorExit` raises `RuntimeError: generator ignored
  GeneratorExit`, which the interpreter prints to stderr on top of whatever
  really failed. The hazard was latent before because the pre-existing yield in
  that block was almost always empty; it is armed on every HTTP-making run now.
- Correct the counting comment in `http_client.py`: cache hits are counted as
  requests, so `http-request-count` is responses handled, not wire flows. The
  proxy-comparable figure is `requests - cache_hits`.
- Document the reporting scope in the `cache_stats` module docstring:
  destinations, the manifest server, and the Connector Builder record counts
  that nothing reads.
- Extend `test_that_response_was_cached` to assert the counters against a live
  `CachedLimiterSession`, pinning the coupling to the real `requests_cache`
  marker. Every other test hand-sets `from_cache`, so an upstream rename would
  have reported 0% forever with the suite green.
- Add a `cdk-migrations.md` note: connector tests asserting an exact
  `trace_messages` count need updating.
---
 airbyte_cdk/entrypoint.py                     |  43 +++++--
 .../sources/streams/http/cache_stats.py       |  13 +-
 .../sources/streams/http/http_client.py       |   5 +-
 cdk-migrations.md                             |  18 +++
 unit_tests/conftest.py                        |  16 ---
 .../sources/streams/http/test_http_client.py  |  10 ++
 unit_tests/test_entrypoint.py                 | 116 ++++++++++++++++++
 7 files changed, 193 insertions(+), 28 deletions(-)

diff --git a/airbyte_cdk/entrypoint.py b/airbyte_cdk/entrypoint.py
index 048d667ca3..777126f969 100644
--- a/airbyte_cdk/entrypoint.py
+++ b/airbyte_cdk/entrypoint.py
@@ -35,7 +35,10 @@
 )
 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.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
@@ -170,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,
@@ -214,18 +223,27 @@ 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)
             ]
-            yield from map(
-                AirbyteEntrypoint.airbyte_message_to_string,
-                self._http_cache_stats_messages(),
-            )
+            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() -> Iterable[AirbyteMessage]:
+    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
@@ -233,16 +251,23 @@ def _http_cache_stats_messages() -> Iterable[AirbyteMessage]:
         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()
-        if stats.requests <= 0:
+        requests_made = stats.requests - baseline.requests
+        if requests_made <= 0:
             return
-        yield create_analytics_message("http-request-count", stats.requests)
-        yield create_analytics_message("http-cache-hit-count", stats.cache_hits)
+        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
index 2b62517115..fd79d2bc7d 100644
--- a/airbyte_cdk/sources/streams/http/cache_stats.py
+++ b/airbyte_cdk/sources/streams/http/cache_stats.py
@@ -13,7 +13,18 @@
 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.
+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
diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py
index 4d1aad4a54..6775c6bec0 100644
--- a/airbyte_cdk/sources/streams/http/http_client.py
+++ b/airbyte_cdk/sources/streams/http/http_client.py
@@ -352,8 +352,9 @@ def _send(
 
         if response is not None:
             # Counted per `_send` call, so a retried request counts once per
-            # attempt -- the same way a proxy counts wire flows, which is what
-            # makes the two numbers comparable in a regression report.
+            # 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(
diff --git a/cdk-migrations.md b/cdk-migrations.md
index 3445a04b88..2fffb518d3 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/conftest.py b/unit_tests/conftest.py
index d5a10e91b1..3a21552bc8 100644
--- a/unit_tests/conftest.py
+++ b/unit_tests/conftest.py
@@ -7,22 +7,6 @@
 import freezegun
 import pytest
 
-from airbyte_cdk.sources.streams.http.cache_stats import HTTP_CACHE_STATS
-
-
-@pytest.fixture(autouse=True)
-def reset_http_cache_stats():
-    """Isolate the process-wide HTTP counters between tests.
-
-    They are module-level by design -- one `requests_cache` backend is shared by
-    a whole run -- but a test process is many runs, and a leftover count makes
-    `AirbyteEntrypoint.run` emit analytics messages in a test that recorded no
-    requests of its own.
-    """
-    HTTP_CACHE_STATS.reset()
-    yield
-    HTTP_CACHE_STATS.reset()
-
 
 @pytest.fixture()
 def mock_sleep(monkeypatch):
diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py
index 421edb0aea..eee2fb3571 100644
--- a/unit_tests/sources/streams/http/test_http_client.py
+++ b/unit_tests/sources/streams/http/test_http_client.py
@@ -465,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
@@ -475,6 +477,14 @@ 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.
diff --git a/unit_tests/test_entrypoint.py b/unit_tests/test_entrypoint.py
index 458f6db762..e55e5c58cf 100644
--- a/unit_tests/test_entrypoint.py
+++ b/unit_tests/test_entrypoint.py
@@ -585,6 +585,122 @@ def test_a_run_that_made_no_requests_reports_no_counts(
     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
 ):