Expose incomplete worker startup observation evidence

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

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

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),
         })