Attribute HTTP client instances and synchronize worker observations

tests/fixtures/worker_startup_reconstruction/README.md

@@ -19,8 +19,9 @@ truncated or invalid, separates existing worker-history requests from
 submissions, and reports HTTP-buffer and milestone-event loss. These are
 availability/retention flags, not end-to-end phase verdicts. A complete JSON
 metadata record does not prove task identity, hydration, handler admission or
-publication. Additional clients remain grouped, and the before/after teardown
-SQL/API comparator and full phase linkage are still pending. The pinned sources
+publication. HTTP clients now have separate bounded instance labels as well as
+the existing role labels. Before/after teardown SQL/API collection and full
+phase linkage are still pending. The pinned sources
 in `proposal.json` remain the historical baseline; this revision is not a new
 historical execution input or qualification receipt.
 
@@ -71,10 +72,18 @@ incompatible harness is a setup finding, not permission to patch the SDK.
   handler, a sent request or a response without an authoritative recorded result
   does not prove durable publication. Match the original task/attempt/run against
   the Server comparator and terminal history.
-- Existing primary HTTP clients have explicit labels; all other clients share
-  the label `additional`. Capture their actual discovery/query-poll coverage,
-  while retaining that limit on per-client attribution. A missing cloned-client
-  trace cannot count as full coverage.
+- Existing primary HTTP clients have explicit role labels; other clients keep
+  the role `additional`. A separate weak-reference registry gives each observed
+  HTTP client a stable instance label and counts its routes, without retaining
+  the client or treating a reused integer object ID as the same instance. Its
+  lifetime count is capped at the observation limit. Requests beyond that cap
+  have no instance label and explicitly truncate client attribution. Known
+  clients retain their labels after overflow; no unknown request is merged into
+  one. Recorder updates and bounded snapshot copies share a lock across the
+  query and workflow threads; no network await is performed while holding it.
+  These are observed HTTP identities, not proof of a Client clone's parent,
+  worker registration or an expected-but-unobserved cloned-client trace. Missing
+  clone coverage still cannot count as complete end-to-end observation.
 
 The driver now captures the same bounded observation before and after shutdown
 on success and failure. Each capture has its own monotonic bracket; capture

tests/test_worker_observation.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
 import asyncio
+import gc
 import itertools
 import json
+import threading
+import weakref
 from typing import Any
 
 import httpx
@@ -54,6 +57,108 @@
     assert "SECRET" not in json.dumps(report)
 
 
+@pytest.mark.asyncio
+async def test_additional_clients_have_stable_distinct_bounded_instance_labels(monkeypatch: Any) -> None:
+    observation = WorkerObservation()
+    observation.install_http(monkeypatch)
+    transport = httpx.MockTransport(lambda request: httpx.Response(200, json={"task": None}))
+    async with httpx.AsyncClient(transport=transport) as first, httpx.AsyncClient(transport=transport) as second:
+        await first.get("https://example.test/api/cluster/info?token=SECRET")
+        await first.post("https://example.test/api/worker/query-tasks/poll")
+        await second.post("https://example.test/api/worker/query-tasks/poll")
+    report = observation.report()
+    assert [entry["client"] for entry in report["recent"]] == ["additional"] * 3
+    assert [entry["client_instance"] for entry in report["recent"]] == ["http-1", "http-1", "http-2"]
+    assert report["client_instances"] == {
+        "http-1": {"counts": {"GET discovery": 1, "POST query_poll": 1}},
+        "http-2": {"counts": {"POST query_poll": 1}},
+    }
+    assert report["completeness"]["client_attribution"] == "complete"
+    assert "SECRET" not in json.dumps(report)
+
+
+@pytest.mark.asyncio
+async def test_client_registry_does_not_retain_clients_or_reuse_their_labels(monkeypatch: Any) -> None:
+    observation = WorkerObservation()
+    observation.install_http(monkeypatch)
+    async with httpx.AsyncClient(transport=httpx.MockTransport(lambda request: httpx.Response(200))) as first:
+        await first.get("https://example.test/api/cluster/info")
+    original = weakref.ref(first)
+    del first
+    gc.collect()
+    assert original() is None
+    async with httpx.AsyncClient(transport=httpx.MockTransport(lambda request: httpx.Response(200))) as second:
+        # Reproduce a reused integer ID without relying on allocator timing.
+        observation._client_instances[id(second)] = (original, "http-1")
+        await second.get("https://example.test/api/cluster/info")
+    report = observation.report()
+    assert [entry["client_instance"] for entry in report["recent"]] == ["http-1", "http-2"]
+    assert len(report["client_instances"]) == 2
+
+
+@pytest.mark.asyncio
+async def test_client_attribution_limit_never_groups_new_clients_as_a_known_instance(monkeypatch: Any) -> None:
+    observation = WorkerObservation(limit=1)
+    observation.install_http(monkeypatch)
+    transport = httpx.MockTransport(lambda request: httpx.Response(200))
+    async with httpx.AsyncClient(transport=transport) as first, httpx.AsyncClient(transport=transport) as second:
+        await first.get("https://example.test/api/cluster/info")
+        await second.get("https://example.test/api/cluster/info")
+        overflow = observation.report()
+        assert overflow["recent"][0]["client_instance"] is None
+        assert overflow["recent"][0]["client_instance_state"] == "truncated"
+        await first.get("https://example.test/api/cluster/info")
+    report = observation.report()
+    assert report["recent"][0]["client_instance"] == "http-1"
+    assert report["recent"][0]["client_instance_state"] == "complete"
+    assert report["completeness"]["client_attribution"] == "truncated"
+    assert len(report["client_instances"]) == 1 and report["unattributed_client_requests"] == 1
+    assert overflow["client_instances"]["http-1"]["counts"]["GET discovery"] == 1
+
+
+def test_query_thread_and_workflow_thread_share_consistent_snapshots(monkeypatch: Any) -> None:
+    observation = WorkerObservation()
+    observation.install_http(monkeypatch)
+    entered = [threading.Event(), threading.Event()]
+    release = threading.Event()
+    errors: list[BaseException] = []
+
+    def run(index: int, family: str) -> None:
+        async def execute():
+            def reply(request: httpx.Request) -> httpx.Response:
+                entered[index].set()
+                assert release.wait(5)
+                return httpx.Response(200, json={"task": None})
+            async with httpx.AsyncClient(transport=httpx.MockTransport(reply)) as client:
+                await client.post(f"https://example.test/api/worker/{family}-tasks/poll")
+                observation.note("request_returned")
+        try:
+            asyncio.run(execute())
+        except BaseException as error:
+            errors.append(error)
+
+    threads = [threading.Thread(target=run, args=(index, family))
+               for index, family in enumerate(("workflow", "query"))]
+    try:
+        for thread in threads:
+            thread.start()
+        assert all(event.wait(5) for event in entered)
+        before = observation.report()
+        assert before["active_count"] == len(before["active"]) == 2
+        assert len({entry["client_instance"] for entry in before["active"]}) == 2
+    finally:
+        release.set()
+        for thread in threads:
+            thread.join(5)
+    assert not errors and not any(thread.is_alive() for thread in threads)
+    after = observation.report()
+    assert after["requests"] == 2 and after["active_count"] == 0
+    assert after["counts"] == {"POST workflow_poll": 1, "POST query_poll": 1}
+    assert after["milestones"]["request_returned"]["count"] == 2
+    assert {entry["request"] for entry in after["recent"]} == {1, 2}
+    assert before["active_count"] == 2 and not before["milestones"]
+
+
 @pytest.mark.asyncio
 async def test_observer_follows_the_actual_client_registration_route(monkeypatch: Any) -> None:
     observation = WorkerObservation()

tests/worker_observation.py

@@ -4,7 +4,9 @@
 
 import asyncio
 import json
+import threading
 import time
+import weakref
 from collections import Counter, deque
 from collections.abc import Callable, Coroutine
 from copy import deepcopy
@@ -80,61 +82,89 @@ def __init__(self, *, clock: Callable[[], float] = time.monotonic, limit: int =
         self.evicted_active = 0
         self.omitted_milestone_events = 0
         self.response_metadata_states: Counter[str] = Counter()
+        self._lock = threading.Lock()
+        self._client_instances: dict[int, tuple[weakref.ReferenceType[httpx.AsyncClient], str]] = {}
+        self.client_instances: dict[str, dict[str, Any]] = {}
+        self.unattributed_client_requests = 0
 
     def note(self, phase: str, task: Any = None) -> None:
-        if phase not in self.milestones and len(self.milestones) >= self.limit:
-            self.omitted_milestone_events += 1
-            return
-        entry = {"at": self.clock() - self.started, **task_identity(task)}
-        milestone = self.milestones.setdefault(phase, {"count": 0, "first": entry})
-        if milestone["count"] >= 2:
-            self.omitted_milestone_events += 1
-        milestone["count"] += 1
-        milestone["last"] = entry
+        with self._lock:
+            if phase not in self.milestones and len(self.milestones) >= self.limit:
+                self.omitted_milestone_events += 1
+                return
+            entry = {"at": self.clock() - self.started, **task_identity(task)}
+            milestone = self.milestones.setdefault(phase, {"count": 0, "first": entry})
+            if milestone["count"] >= 2:
+                self.omitted_milestone_events += 1
+            milestone["count"] += 1
+            milestone["last"] = entry
+
+    def _client_instance(self, client: httpx.AsyncClient, key: str) -> str | None:
+        """Called under the lock; weak references never keep an HTTP client alive."""
+        previous = self._client_instances.get(id(client))
+        if previous is not None and previous[0]() is client:
+            label = previous[1]
+        elif len(self.client_instances) < self.limit:
+            label = f"http-{len(self.client_instances) + 1}"
+            self._client_instances[id(client)] = (weakref.ref(client), label)
+            self.client_instances[label] = {"counts": Counter()}
+        else:
+            self.unattributed_client_requests += 1
+            return None
+        self.client_instances[label]["counts"][key] += 1
+        return label
 
     def install_http(self, monkeypatch: Any) -> None:
         original_send = httpx.AsyncClient.send
 
         async def observed_send(
             client: httpx.AsyncClient, request: httpx.Request, *args: Any, **kwargs: Any,
         ) -> httpx.Response:
-            self.total += 1
-            request_id = self.total
             started = self.clock()
             route = route_name(request)
             method = request.method if request.method in {"GET", "POST", "PUT", "PATCH", "DELETE"} else "OTHER"
-            self.counts[f"{method} {route}"] += 1
-            entry: dict[str, Any] = {
-                "request": request_id, "client": self.clients.get(id(client), "additional"),
-                "method": method, "route": route, "started": started - self.started,
-                "response_metadata_state": "missing",
-            }
-            if len(self.active) == self.limit:
-                self.active.pop(next(iter(self.active)))
-                self.evicted_active += 1
-            self.active[request_id] = entry
-            self.active_count += 1
+            key = f"{method} {route}"
+            with self._lock:
+                self.total += 1
+                request_id = self.total
+                self.counts[key] += 1
+                instance = self._client_instance(client, key)
+                entry: dict[str, Any] = {
+                    "request": request_id, "client": self.clients.get(id(client), "additional"),
+                    "client_instance": instance,
+                    "client_instance_state": "complete" if instance is not None else "truncated",
+                    "method": method, "route": route, "started": started - self.started,
+                    "response_metadata_state": "missing",
+                }
+                if len(self.active) == self.limit:
+                    self.active.pop(next(iter(self.active)))
+                    self.evicted_active += 1
+                self.active[request_id] = entry
+                self.active_count += 1
             try:
                 response = await original_send(client, request, *args, **kwargs)
-                entry["status"] = response.status_code
-                if 200 <= response.status_code < 300:
-                    self.successful[f"{method} {route}"] += 1
+                metadata: dict[str, Any] = {"status": response.status_code}
                 if response.is_stream_consumed:
-                    entry["response_metadata_state"] = self._response_metadata(entry, response)
+                    metadata["response_metadata_state"] = self._response_metadata(metadata, response)
+                with self._lock:
+                    entry.update(metadata)
+                    if 200 <= response.status_code < 300:
+                        self.successful[key] += 1
                 return response
             except BaseException as error:
-                entry["exception"] = type(error).__name__
+                with self._lock:
+                    entry["exception"] = type(error).__name__
                 raise
             finally:
-                entry["elapsed"] = self.clock() - started
-                key = f"{method} {route}"
-                self.max_elapsed[key] = max(self.max_elapsed.get(key, 0), entry["elapsed"])
-                self.active_count -= 1
-                self.active.pop(request_id, None)
-                self.response_metadata_states[entry["response_metadata_state"]] += 1
-                if len(self.recent) == self.limit:
-                    self.omitted_completed += 1
-                self.recent.append(entry)
+                with self._lock:
+                    entry["elapsed"] = self.clock() - started
+                    self.max_elapsed[key] = max(self.max_elapsed.get(key, 0), entry["elapsed"])
+                    self.active_count -= 1
+                    self.active.pop(request_id, None)
+                    self.response_metadata_states[entry["response_metadata_state"]] += 1
+                    if len(self.recent) == self.limit:
+                        self.omitted_completed += 1
+                    self.recent.append(entry)
 
         monkeypatch.setattr(httpx.AsyncClient, "send", observed_send)
 
@@ -176,6 +206,10 @@ def _response_metadata(entry: dict[str, Any], response: httpx.Response) -> str:
         return state
 
     def report(self, *, include_recent: bool = True) -> dict[str, Any]:
+        with self._lock:
+            return self._report(include_recent=include_recent)
+
+    def _report(self, *, include_recent: bool) -> dict[str, Any]:
         return deepcopy({
             "elapsed": self.clock() - self.started, "requests": self.total,
             "counts": dict(self.counts), "milestones": self.milestones,
@@ -190,11 +224,14 @@ def report(self, *, include_recent: bool = True) -> dict[str, Any]:
                 ),
                 "active_http": "truncated" if self.evicted_active else "complete",
                 "milestones": "truncated" if self.omitted_milestone_events else "complete",
+                "client_attribution": "truncated" if self.unattributed_client_requests else "complete",
             },
             "omitted_completed": self.omitted_completed,
             "evicted_active": self.evicted_active,
             "omitted_milestone_events": self.omitted_milestone_events,
             "response_metadata_states": dict(self.response_metadata_states),
+            "client_instances": self.client_instances,
+            "unattributed_client_requests": self.unattributed_client_requests,
         })