From 2968624eb624ad4efb447771f68ab90822b27672 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 18:23:31 +0000 Subject: [PATCH] Expose missing and truncated worker observation evidence --- .../worker_startup_reconstruction/README.md | 10 +++ tests/test_worker_observation.py | 66 +++++++++++++++++++ tests/worker_observation.py | 50 ++++++++++++-- 3 files changed, 119 insertions(+), 7 deletions(-) diff --git a/tests/fixtures/worker_startup_reconstruction/README.md b/tests/fixtures/worker_startup_reconstruction/README.md index a5fefe29..bce669be 100644 --- a/tests/fixtures/worker_startup_reconstruction/README.md +++ b/tests/fixtures/worker_startup_reconstruction/README.md @@ -14,6 +14,16 @@ describes synthetic evidence and the limited conclusion each combination can support. The vectors are review inputs, not captured observations or an implemented classifier. Neither file is an execution manifest. +The first observer-source slice now labels metadata as complete, missing, +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 +in `proposal.json` remain the historical baseline; this revision is not a new +historical execution input or qualification receipt. + ## Reuse the existing driver Use `tests/integration/test_worker_task_families.py` and diff --git a/tests/test_worker_observation.py b/tests/test_worker_observation.py index 1fc85cbc..22291571 100644 --- a/tests/test_worker_observation.py +++ b/tests/test_worker_observation.py @@ -129,6 +129,7 @@ async def __aiter__(self): response = await client.send(client.build_request("GET", "https://example.test/stream"), stream=True) assert reads == [] assert "task_state" not in observation.report()["recent"][0] + assert observation.report()["recent"][0]["response_metadata_state"] == "missing" assert await response.aread() == b'{"task":null}' await response.aclose() @@ -145,6 +146,10 @@ async def test_observation_bounds_completed_records_and_milestones(monkeypatch: assert report["requests"] == 5 assert len(report["recent"]) == 2 assert len(report["milestones"]) == 2 + assert report["omitted_completed"] == 3 + assert report["omitted_milestone_events"] == 3 + assert report["completeness"]["recent_http"] == "truncated" + assert report["completeness"]["milestones"] == "truncated" assert "SECRET" not in json.dumps(report) @@ -171,11 +176,67 @@ async def held(request: httpx.Request) -> httpx.Response: assert report["active_count"] == 5 assert len(report["active"]) == 2 assert report["omitted_active"] == 3 + assert report["evicted_active"] == 3 + assert report["completeness"]["active_http"] == "truncated" finally: for runner in runners: runner.cancel() await asyncio.gather(*runners, return_exceptions=True) assert observation.report()["active_count"] == 0 + assert observation.report()["completeness"]["active_http"] == "truncated" + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("body", "content_type", "expected"), [ + (b'{"task":null,"recorded":false,"input":"SECRET"}', "application/json", "complete"), + (b"SECRET", "text/plain", "missing"), + (b'{"input":"' + b"x" * 32_768 + b'"}', "application/json", "truncated"), + (b'{"input":"SECRET"', "application/json", "invalid"), + (b'["SECRET"]', "application/json", "invalid"), + (b'{"task":"SECRET"}', "application/json", "invalid"), + (b'{"recorded":"SECRET"}', "application/json", "invalid"), +]) +async def test_http_metadata_omissions_have_explicit_states( + monkeypatch: Any, body: bytes, content_type: str, expected: str, +) -> None: + observation = WorkerObservation() + observation.install_http(monkeypatch) + async with httpx.AsyncClient(transport=httpx.MockTransport( + lambda request: httpx.Response(200, content=body, headers={"Content-Type": content_type}), + )) as client: + response = await client.post("https://example.test/api/worker/workflow-tasks/original/history") + assert response.content == body + report = observation.report() + assert report["counts"] == {"POST workflow_history": 1} + assert report["recent"][0]["response_metadata_state"] == expected + assert report["response_metadata_states"] == {expected: 1} + assert "SECRET" not in json.dumps(report) + + +def test_middle_milestone_loss_and_summary_omission_are_explicit() -> None: + observation = WorkerObservation() + for task_id in ("first", "middle", "last"): + observation.note("workflow_task_enter", {"task_id": task_id}) + report = observation.report() + assert report["milestones"]["workflow_task_enter"]["first"]["task_id"] == "first" + assert report["milestones"]["workflow_task_enter"]["last"]["task_id"] == "last" + assert report["omitted_milestone_events"] == 1 + assert report["completeness"]["milestones"] == "truncated" + + +@pytest.mark.asyncio +async def test_recent_records_suppressed_in_report_are_missing_not_complete(monkeypatch: Any) -> None: + observation = WorkerObservation() + observation.install_http(monkeypatch) + async with httpx.AsyncClient(transport=httpx.MockTransport( + lambda request: httpx.Response(200, json={}), + )) as client: + await client.get("https://example.test/result") + report = observation.report(include_recent=False) + assert report["recent"] == [] + assert report["omitted_completed"] == 0 + assert report["completeness"]["recent_http"] == "missing" + assert report["response_metadata_states"] == {"complete": 1} @pytest.mark.asyncio @@ -304,6 +365,11 @@ async def fail(context): ("POST", "/api/worker/registrations/worker/deregister", "deregistration"), ("DELETE", "/api/worker/registrations/worker", "deregistration"), ("GET", "/api/worker/registrations/worker", "registration"), + ("POST", "/api/worker/workflow-tasks/original/history", "workflow_history"), + ("GET", "/api/worker/workflow-tasks/original/history", "workflow_history"), + ("POST", "/api/worker/workflow-tasks/original/events", "workflow_submission"), + ("POST", "/api/worker/workflow-tasks/original/other/history", "workflow_submission"), + ("POST", "/api/worker/workflow-tasks//history", "workflow_submission"), ]) def test_registration_and_deregistration_are_counted_separately(method: str, path: str, expected: str) -> None: from tests.worker_observation import route_name diff --git a/tests/worker_observation.py b/tests/worker_observation.py index 8d38a4e0..18992d28 100644 --- a/tests/worker_observation.py +++ b/tests/worker_observation.py @@ -22,6 +22,11 @@ def route_name(request: httpx.Request) -> str: path = request.url.path.rstrip("/") + if "/worker/workflow-tasks/" in path: + task_path = path.rsplit("/worker/workflow-tasks/", 1)[1] + task_id, _, operation = task_path.partition("/") + if task_id and operation == "history": + return "workflow_history" for family in ("workflow", "activity", "query"): if path.endswith(f"/worker/{family}-tasks/poll"): return f"{family}_poll" @@ -70,12 +75,19 @@ def __init__(self, *, clock: Callable[[], float] = time.monotonic, limit: int = self.max_elapsed: dict[str, float] = {} self.milestones: dict[str, dict[str, Any]] = {} self.clients: dict[int, str] = {} + self.omitted_completed = 0 + self.evicted_active = 0 + self.omitted_milestone_events = 0 + self.response_metadata_states: Counter[str] = Counter() 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 @@ -94,9 +106,11 @@ async def observed_send( 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 try: @@ -105,7 +119,7 @@ async def observed_send( if 200 <= response.status_code < 300: self.successful[f"{method} {route}"] += 1 if response.is_stream_consumed: - self._response_metadata(entry, response) + entry["response_metadata_state"] = self._response_metadata(entry, response) return response except BaseException as error: entry["exception"] = type(error).__name__ @@ -116,29 +130,39 @@ async def observed_send( 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) @staticmethod - def _response_metadata(entry: dict[str, Any], response: httpx.Response) -> None: + def _response_metadata(entry: dict[str, Any], response: httpx.Response) -> str: if not response.headers.get("content-type", "").startswith("application/json"): - return + return "missing" try: if len(response.content) > 32_768: - return + return "truncated" body = response.json() - except (ValueError, httpx.ResponseNotRead): - return + except httpx.ResponseNotRead: + return "missing" + except ValueError: + return "invalid" if not isinstance(body, dict): - return + return "invalid" + state = "complete" if "task" in body: entry["task_state"] = ( "present" if isinstance(body["task"], dict) else "null" if body["task"] is None else "invalid" ) entry.update(task_identity(body["task"])) + if entry["task_state"] == "invalid": + state = "invalid" if type(body.get("recorded")) is bool: entry["recorded"] = body["recorded"] + elif "recorded" in body: + state = "invalid" error = body.get("error") if isinstance(error, dict): code = error.get("code") @@ -148,6 +172,7 @@ def _response_metadata(entry: dict[str, Any], response: httpx.Response) -> None: value = body.get(field) if isinstance(value, str) and len(value) <= 80 and value.replace("_", "").isalnum(): entry[field] = value + return state def report(self, *, include_recent: bool = True) -> dict[str, Any]: return deepcopy({ @@ -158,6 +183,17 @@ def report(self, *, include_recent: bool = True) -> dict[str, Any]: "active_count": self.active_count, "active": list(self.active.values()), "omitted_active": self.active_count - len(self.active), "recent": list(self.recent) if include_recent else [], + "completeness": { + "recent_http": "missing" if not include_recent and self.recent else ( + "truncated" if self.omitted_completed else "complete" + ), + "active_http": "truncated" if self.evicted_active else "complete", + "milestones": "truncated" if self.omitted_milestone_events 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), })