fix(benchmark): retain sampling and teardown failure evidence

moli-benchmark/README.md

@@ -420,6 +420,22 @@ Run the command with `--help` to see fixture, public-site, sampling, and output
 options. The pull-request workflows under `.github/workflows/` are the source
 of truth for CI execution and permissions.
 
+## Sampling and failure evidence
+
+Resource samples retain process identities, read errors, and the raw observations
+used for CPU and memory totals. Missing or incomplete reads remain unavailable;
+they are not reported as zero. Thread CPU deltas and process clock ticks are
+recorded separately, including thread retirement and counter precision.
+
+Target shutdown seals the sampler before terminating the engine and records log
+reader EOF, cleanup errors, and timeout results. Startup and cancellation failures
+retain their original exceptions together with the evidence collected so far.
+Sequential CDP receive failures attach already observed messages through
+`cdp_receive_evidence`, without consuming extra messages or replacing the error.
+
+These records improve measurement validity and failure diagnosis. They do not
+change browser rendering policy or establish a performance improvement.
+
 ## Development
 
 Run the core CLI tests from `moli-benchmark/`:

moli-benchmark/moli_benchmark/raw_cdp.py

@@ -6,8 +6,9 @@ import contextlib
 import json
 import time
 import urllib.request
+from collections.abc import Callable
 from dataclasses import dataclass
-from typing import Any, Callable
+from typing import Any
 
 import websockets
 from websockets.asyncio.client import ClientConnection
@@ -95,19 +96,39 @@ class RawCdpClient:
             raise RawCdpError(f"unexpected CDP payload: {payload!r}")
         return payload
 
-    async def recv_until_id(self, message_id: int, *, timeout: float = 10.0) -> tuple[dict[str, Any], list[dict[str, Any]]]:
+    async def recv_until_id(
+        self,
+        message_id: int,
+        *,
+        timeout: float = 10.0,
+        method: str | None = None,
+    ) -> tuple[dict[str, Any], list[dict[str, Any]]]:
         seen: list[dict[str, Any]] = []
         deadline = asyncio.get_running_loop().time() + timeout
-        while True:
-            remaining = deadline - asyncio.get_running_loop().time()
-            if remaining <= 0:
-                raise RawCdpTimeoutError(f"timed out waiting for CDP response id={message_id}; seen={seen[-20:]}")
-            message = await asyncio.wait_for(self.recv(), timeout=remaining)
-            seen.append(message)
-            if message.get("id") == message_id:
-                if "error" in message:
-                    raise RawCdpError(f"CDP command id={message_id} failed: {message['error']}")
-                return message, seen
+        try:
+            while True:
+                remaining = deadline - asyncio.get_running_loop().time()
+                if remaining <= 0:
+                    raise RawCdpTimeoutError(
+                        f"timed out waiting for CDP response id={message_id}; seen={seen[-20:]}"
+                    )
+                message = await asyncio.wait_for(self.recv(), timeout=remaining)
+                seen.append(message)
+                if message.get("id") == message_id:
+                    if "error" in message:
+                        raise RawCdpError(
+                            f"CDP command id={message_id} failed: {message['error']}"
+                        )
+                    return message, seen
+        except BaseException as error:
+            # Keep the same exception (including cancellation) and receiver.
+            # A failed response/receive must not discard already observed events.
+            error.cdp_receive_evidence = {
+                "message_id": message_id,
+                "method": method,
+                "messages": list(seen),
+            }
+            raise
 
 
 async def connect_raw_cdp(endpoint: str) -> RawCdpClient:

moli-benchmark/moli_benchmark/sampling.py

@@ -1,171 +1,468 @@
 from __future__ import annotations
 
+import copy
+import hashlib
 import os
 import subprocess
 import threading
 import time
+import traceback
 from dataclasses import dataclass, field
+from itertools import pairwise
 from pathlib import Path
 from typing import Any
 
+INSTRUMENTATION_SCHEMA = 2
 
-def _process_tree(root_pid: int) -> list[int]:
-    ppid_by_pid: dict[int, int] = {}
-    for stat_path in Path("/proc").glob("[0-9]*/stat"):
-        try:
-            raw = stat_path.read_text(encoding="utf-8", errors="replace")
-        except OSError:
-            continue
-        close = raw.rfind(")")
-        if close < 0:
-            continue
-        parts = raw[close + 2 :].split()
-        if len(parts) < 2:
-            continue
-        try:
-            pid = int(stat_path.parent.name)
-            ppid = int(parts[1])
-        except ValueError:
-            continue
-        ppid_by_pid[pid] = ppid
 
-    result = [root_pid]
-    seen = {root_pid}
-    changed = True
-    while changed:
-        changed = False
-        for pid, ppid in ppid_by_pid.items():
-            if pid not in seen and ppid in seen:
-                result.append(pid)
-                seen.add(pid)
-                changed = True
+def error_evidence(error: BaseException) -> dict[str, Any]:
+    result = {"type": type(error).__name__, "message": str(error)}
+    if isinstance(error, OSError):
+        result.update(errno=error.errno, filename=error.filename)
     return result
 
 
-def _read_pss_bytes(pid: int) -> int | None:
+def _text_read(path: Path) -> dict[str, Any]:
+    record = {"path": str(path), "started_monotonic_ns": time.monotonic_ns()}
     try:
-        text = Path(f"/proc/{pid}/smaps_rollup").read_text(encoding="utf-8", errors="replace")
-    except OSError:
-        return None
-    for line in text.splitlines():
-        if line.startswith("Pss:"):
-            parts = line.split()
-            if len(parts) >= 2:
-                return int(parts[1]) * 1024
-    return None
+        record.update(
+            status="ok", raw=path.read_text(encoding="utf-8", errors="replace")
+        )
+    except OSError as error:
+        record.update(status="os_error", error=error_evidence(error), raw=None)
+    record["finished_monotonic_ns"] = time.monotonic_ns()
+    return record
+
+
+def _stat_read(pid: int, tid: int | None = None) -> dict[str, Any]:
+    path = Path(f"/proc/{pid}/stat" if tid is None else f"/proc/{pid}/task/{tid}/stat")
+    record = _text_read(path)
+    record["identity"] = None
+    if record["status"] == "ok":
+        try:
+            raw = record["raw"]
+            close = raw.rfind(")")
+            fields = raw[close + 2 :].split()
+            observed_id = int(raw[: raw.index("(")].strip())
+            if close < 0 or observed_id != (pid if tid is None else tid):
+                raise ValueError("proc stat ID/format mismatch")
+            user, system, start = (int(fields[index]) for index in (11, 12, 19))
+            if min(user, system, start) < 0:
+                raise ValueError("negative proc stat counter")
+            record.update(
+                pid=pid,
+                tid=tid,
+                start_ticks=start,
+                identity=f"{pid}:{start}" if tid is None else f"{pid}:{tid}:{start}",
+                state=fields[0],
+                ppid=int(fields[1]),
+                user_ticks=user,
+                system_ticks=system,
+                total_ticks=user + system,
+            )
+        except (ValueError, IndexError) as error:
+            record.update(status="parse_error", error=error_evidence(error))
+    return record
+
+
+def _same_identity(before: dict, after: dict) -> bool:
+    return before.get("identity") is not None and before["identity"] == after.get(
+        "identity"
+    )
+
+
+def _process_tree_evidence(root_pid: int) -> tuple[list[int], dict[int, dict], dict]:
+    stats, failures = {}, {}
+    enumeration_error = None
+    try:
+        paths = list(Path("/proc").glob("[0-9]*/stat"))
+    except OSError as error:
+        paths, enumeration_error = [], error_evidence(error)
+    for path in paths:
+        pid = int(path.parent.name)
+        record = _stat_read(pid)
+        if record["status"] == "ok":
+            stats[pid] = record
+        else:
+            key = f"{record['status']}:{record.get('error', {}).get('errno')}"
+            failures[key] = failures.get(key, 0) + 1
+    pids, seen = [root_pid], {root_pid}
+    for parent in pids:
+        for pid, record in stats.items():
+            if record["ppid"] == parent and pid not in seen:
+                pids.append(pid)
+                seen.add(pid)
+    # Root is kept even when unreadable: never turn missing coverage into 0/0.
+    if root_pid not in stats:
+        stats[root_pid] = _stat_read(root_pid)
+    return (
+        pids,
+        {pid: stats[pid] for pid in pids},
+        {
+            "scope": "best_effort_non_atomic_proc_tree",
+            "enumerated_stat_paths": len(paths),
+            "scan_failures_by_status_errno": failures,
+            "enumeration_error": enumeration_error,
+            "all_descendants_observed": "unknown",
+        },
+    )
+
+
+def _process_tree(root_pid: int) -> list[int]:
+    return _process_tree_evidence(root_pid)[0]
+
+
+def _field_read(source: dict, field_name: str, multiplier: int = 1) -> dict:
+    record = {key: value for key, value in source.items() if key != "raw"}
+    record.update(field=field_name, value=None)
+    if source["status"] != "ok":
+        return record
+    for line in source["raw"].splitlines():
+        fields = line.split()
+        if fields and fields[0] == field_name:
+            record["raw_line"] = line
+            try:
+                value = int(fields[1])
+                if value < 0:
+                    raise ValueError("negative procfs field")
+                record["value"] = value * multiplier
+            except (ValueError, IndexError) as error:
+                record.update(status="parse_error", error=error_evidence(error))
+            return record
+    record["status"] = "missing_field"
+    return record
+
+
+def _fd_read(pid: int) -> dict:
+    path = Path(f"/proc/{pid}/fd")
+    record = {
+        "path": str(path),
+        "value": None,
+        "started_monotonic_ns": time.monotonic_ns(),
+    }
+    try:
+        record.update(status="ok", value=sum(1 for _ in path.iterdir()))
+    except OSError as error:
+        record.update(status="os_error", error=error_evidence(error))
+    record["finished_monotonic_ns"] = time.monotonic_ns()
+    return record
+
+
+def _read_pss_bytes(pid: int) -> int | None:
+    return _field_read(_text_read(Path(f"/proc/{pid}/smaps_rollup")), "Pss:", 1024)[
+        "value"
+    ]
 
 
 def _read_rss_bytes(pid: int) -> int | None:
-    try:
-        text = Path(f"/proc/{pid}/status").read_text(encoding="utf-8", errors="replace")
-    except OSError:
+    return _field_read(_text_read(Path(f"/proc/{pid}/status")), "VmRSS:", 1024)["value"]
+
+
+def _read_thread_count(pid: int) -> int | None:
+    return _field_read(_text_read(Path(f"/proc/{pid}/status")), "Threads:")["value"]
+
+
+def _read_fd_count(pid: int) -> int | None:
+    return _fd_read(pid)["value"]
+
+
+def _read_cpu_identity(pid: int) -> tuple[tuple[int, int], int] | None:
+    value = _stat_read(pid)
+    if value["status"] != "ok":
         return None
-    for line in text.splitlines():
-        if line.startswith("VmRSS:"):
-            parts = line.split()
-            if len(parts) >= 2:
-                return int(parts[1]) * 1024
-    return None
+    return (pid, value["start_ticks"]), value["total_ticks"]
 
 
 def _read_cpu_percent(pids: list[int]) -> float | None:
-    if not pids:
-        return None
+    # Compatibility for other callers: this is ps lifetime %CPU, not phase CPU.
     try:
         completed = subprocess.run(
-            ["ps", "-o", "%cpu=", "-p", ",".join(str(pid) for pid in pids)],
+            ["ps", "-o", "%cpu=", "-p", ",".join(map(str, pids))],
             capture_output=True,
             text=True,
             timeout=1,
             check=False,
         )
-    except (OSError, subprocess.TimeoutExpired):
-        return None
-    total = 0.0
-    found = False
-    for line in completed.stdout.splitlines():
-        try:
-            total += float(line.strip())
-            found = True
-        except ValueError:
-            continue
-    return total if found else None
-
-
-def _read_thread_count(pid: int) -> int | None:
-    try:
-        text = Path(f"/proc/{pid}/status").read_text(encoding="utf-8", errors="replace")
-    except OSError:
-        return None
-    for line in text.splitlines():
-        if line.startswith("Threads:"):
-            parts = line.split()
-            if len(parts) >= 2:
-                return int(parts[1])
-    return None
-
-
-def _read_fd_count(pid: int) -> int | None:
-    try:
-        return sum(1 for _ in Path(f"/proc/{pid}/fd").iterdir())
-    except OSError:
+        return (
+            sum(float(line.strip()) for line in completed.stdout.splitlines())
+            if completed.returncode == 0 and completed.stdout.strip()
+            else None
+        )
+    except (OSError, ValueError, subprocess.TimeoutExpired):
         return None
 
 
-def _read_cpu_identity(pid: int) -> tuple[tuple[int, int], int] | None:
-    try:
-        raw = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8", errors="replace")
-    except OSError:
-        return None
-    close = raw.rfind(")")
-    if close < 0:
-        return None
-    fields = raw[close + 2 :].split()
-    if len(fields) <= 19:
-        return None
-    try:
-        user_ticks = int(fields[11])
-        system_ticks = int(fields[12])
-        start_ticks = int(fields[19])
-    except ValueError:
-        return None
-    return (pid, start_ticks), user_ticks + system_ticks
-
-
-def snapshot_resources(root_pid: int, *, include_lifetime_cpu: bool = True) -> dict[str, Any]:
-    pids = _process_tree(root_pid)
-    pss_values = [_read_pss_bytes(pid) for pid in pids]
-    rss_values = [_read_rss_bytes(pid) for pid in pids]
-    thread_values = [_read_thread_count(pid) for pid in pids]
-    fd_values = [_read_fd_count(pid) for pid in pids]
-    pss_values = [value for value in pss_values if value is not None]
-    rss_values = [value for value in rss_values if value is not None]
-    thread_values = [value for value in thread_values if value is not None]
-    fd_values = [value for value in fd_values if value is not None]
-    return {
+def snapshot_resources(
+    root_pid: int, *, include_lifetime_cpu: bool = True
+) -> dict[str, Any]:
+    started = time.monotonic_ns()
+    pids, before, tree = _process_tree_evidence(root_pid)
+    processes = []
+    for pid in pids:
+        status = _text_read(Path(f"/proc/{pid}/status"))
+        row = {
+            "pid": pid,
+            "identity_before": before[pid],
+            "pss": _field_read(
+                _text_read(Path(f"/proc/{pid}/smaps_rollup")), "Pss:", 1024
+            ),
+            "rss": _field_read(status, "VmRSS:", 1024),
+            "threads": _field_read(status, "Threads:"),
+            "fds": _fd_read(pid),
+        }
+        row["identity_after"] = _stat_read(pid)
+        row["identity_valid"] = _same_identity(before[pid], row["identity_after"])
+        processes.append(row)
+    result = {
+        "instrumentation_schema": INSTRUMENTATION_SCHEMA,
         "pids": pids,
         "process_count": len(pids),
-        "pss_bytes": sum(pss_values) if len(pss_values) == len(pids) else None,
-        "pss_process_count": len(pss_values),
-        "rss_bytes": sum(rss_values) if len(rss_values) == len(pids) else None,
-        "rss_process_count": len(rss_values),
-        "thread_count": sum(thread_values)
-        if len(thread_values) == len(pids)
-        else None,
-        "thread_process_count": len(thread_values),
-        "fd_count": sum(fd_values) if len(fd_values) == len(pids) else None,
-        "fd_process_count": len(fd_values),
+        "tree": tree,
+        "processes": processes,
+        "started_monotonic_ns": started,
+        "finished_monotonic_ns": time.monotonic_ns(),
         "cpu_percent": _read_cpu_percent(pids) if include_lifetime_cpu else None,
+        "cpu_percent_method": "ps_lifetime" if include_lifetime_cpu else None,
+    }
+    for name, total, count in (
+        ("pss", "pss_bytes", "pss_process_count"),
+        ("rss", "rss_bytes", "rss_process_count"),
+        ("threads", "thread_count", "thread_process_count"),
+        ("fds", "fd_count", "fd_process_count"),
+    ):
+        values = [
+            row[name]["value"]
+            for row in processes
+            if row["identity_valid"] and row[name]["status"] == "ok"
+        ]
+        result[count] = len(values)
+        result[total] = sum(values) if pids and len(values) == len(pids) else None
+    result["selected_process_reads_complete"] = (
+        bool(processes)
+        and all(
+            row["identity_valid"]
+            and all(
+                row[name]["status"] == "ok" for name in ("pss", "rss", "threads", "fds")
+            )
+            for row in processes
+        )
+        and tree.get("enumeration_error") is None
+    )
+    result["finished_monotonic_ns"] = time.monotonic_ns()
+    return result
+
+
+def clock_ticks_evidence() -> dict:
+    result = {"source": "os.sysconf(SC_CLK_TCK)", "ticks_per_second": None}
+    try:
+        rate = os.sysconf("SC_CLK_TCK")
+        if not isinstance(rate, int) or rate <= 0:
+            raise ValueError("invalid SC_CLK_TCK")
+        result.update(ticks_per_second=rate, counter_tick_ms=1000 / rate)
+    except (OSError, ValueError) as error:
+        result["error"] = error_evidence(error)
+    return result
+
+
+def cpu_snapshot(pid: int) -> dict:
+    """Root process ticks and its observed live threads; no descendant CPU."""
+    result = {
+        "started_monotonic_ns": time.monotonic_ns(),
+        "root_before": _stat_read(pid),
+        "threads": {},
+        "observations": [],
+        "enumeration_error": None,
+    }
+    try:
+        tasks = list(Path(f"/proc/{pid}/task").iterdir())
+    except OSError as error:
+        tasks = []
+        result["enumeration_error"] = error_evidence(error)
+    result["enumerated_tids"] = sorted(int(task.name) for task in tasks)
+    for tid in result["enumerated_tids"]:
+        before = _stat_read(pid, tid)
+        runtime = _text_read(Path(f"/proc/{pid}/task/{tid}/schedstat"))
+        runtime["value"] = None
+        if runtime["status"] == "ok":
+            try:
+                runtime["value"] = int(runtime["raw"].split()[0])
+                if runtime["value"] < 0:
+                    raise ValueError("negative schedstat runtime")
+            except (ValueError, IndexError) as error:
+                runtime.update(
+                    status="parse_error", error=error_evidence(error), value=None
+                )
+        after = _stat_read(pid, tid)
+        valid = _same_identity(before, after) and runtime["status"] == "ok"
+        result["observations"].append(
+            {
+                "tid": tid,
+                "before": before,
+                "runtime": runtime,
+                "after": after,
+                "valid": valid,
+            }
+        )
+        if valid:
+            result["threads"][before["identity"]] = runtime["value"]
+    result["root_after"] = _stat_read(pid)
+    result["finished_monotonic_ns"] = time.monotonic_ns()
+    return result
+
+
+def cpu_delta(
+    before: dict, after: dict, clock_ticks: dict, expected_root: str | None
+) -> dict:
+    old, new = before["threads"], after["threads"]
+    common, created, retired = (
+        old.keys() & new.keys(),
+        new.keys() - old.keys(),
+        old.keys() - new.keys(),
+    )
+    regressed = sorted(key for key in common if new[key] < old[key])
+    identities = [
+        point[key].get("identity")
+        for point in (before, after)
+        for key in ("root_before", "root_after")
+    ]
+    root_valid = expected_root is not None and all(
+        key == expected_root for key in identities
+    )
+    boundary_complete = root_valid and all(
+        point["enumeration_error"] is None
+        and bool(point["observations"])
+        and all(row["valid"] for row in point["observations"])
+        for point in (before, after)
+    )
+    known_ns = (
+        None
+        if regressed or not root_valid
+        else sum(new[key] - old[key] for key in common)
+        + sum(new[key] for key in created)
+    )
+    first, last = before["root_after"], after["root_after"]
+    process_valid = root_valid and all(
+        last.get(key, -1) >= first.get(key, 0) for key in ("user_ticks", "system_ticks")
+    )
+    ticks = last["total_ticks"] - first["total_ticks"] if process_valid else None
+    rate = clock_ticks["ticks_per_second"]
+    process_cpu = {
+        "method": "proc_root_stat_utime_plus_stime",
+        "scope": "root_process_all_threads_no_descendants",
+        "before": first,
+        "after": last,
+        "clock_ticks": clock_ticks,
+        "identity_valid": root_valid,
+        "counter_valid": process_valid,
+        "delta_ticks": ticks,
+        "cpu_ms": ticks / rate * 1000 if ticks is not None and rate else None,
+        "zero_ticks_mean": "no_exported_tick_change_not_zero_cpu",
+        "remote_process_clock": {"status": "unsupported_not_runtime_validated"},
+    }
+    return {
+        # Historical field stays an explicitly named thread estimate. Never
+        # silently substitute process ticks when thread evidence is incomplete.
+        "cpu_ms": known_ns / 1e6
+        if known_ns is not None and boundary_complete
+        else None,
+        "thread_cpu_observed": {
+            "method": "root_threads_schedstat_boundary_estimate",
+            "before": before,
+            "after": after,
+            "known_cpu_ms": known_ns / 1e6 if known_ns is not None else None,
+            "boundary_read_complete": boundary_complete,
+            "counter_regressions": regressed,
+            "lifetime_coverage": "not_proven_by_boundary_sampling",
+        },
+        "created_threads": sorted(created),
+        "retired_threads": sorted(retired),
+        "process_cpu": process_cpu,
+    }
+
+
+def affinity_snapshot(
+    pid: int, requested_cpu: int | None, expected_root: str | None
+) -> dict:
+    result = {
+        "started_monotonic_ns": time.monotonic_ns(),
+        "root_before": _stat_read(pid),
+        "requested_cpu": requested_cpu,
+        "threads": [],
+        "enumeration_error": None,
+    }
+    try:
+        tids = sorted(int(path.name) for path in Path(f"/proc/{pid}/task").iterdir())
+    except OSError as error:
+        tids = []
+        result["enumeration_error"] = error_evidence(error)
+    for tid in tids:
+        before = _stat_read(pid, tid)
+        row = {"tid": tid, "before": before, "cpus": None, "error": None}
+        try:
+            row["cpus"] = sorted(os.sched_getaffinity(tid))
+        except OSError as error:
+            row["error"] = error_evidence(error)
+        row["after"] = _stat_read(pid, tid)
+        row["identity_valid"] = _same_identity(before, row["after"])
+        result["threads"].append(row)
+    result["root_after"] = _stat_read(pid)
+    result["boundary_complete"] = (
+        bool(tids)
+        and result["enumeration_error"] is None
+        and expected_root is not None
+        and all(
+            result[key].get("identity") == expected_root
+            for key in ("root_before", "root_after")
+        )
+        and all(
+            row["identity_valid"] and row["error"] is None for row in result["threads"]
+        )
+    )
+    result["all_observed_match"] = (
+        result["boundary_complete"]
+        and requested_cpu is not None
+        and all(row["cpus"] == [requested_cpu] for row in result["threads"])
+    )
+    result["lifetime_affinity"] = "not_proven_by_boundary_sampling"
+    result["finished_monotonic_ns"] = time.monotonic_ns()
+    return result
+
+
+def instrumentation_manifest(probe_path: Path, *input_paths: Path) -> dict:
+    paths = [
+        probe_path,
+        Path(__file__),
+        Path(__file__).with_name("target_serve.py"),
+        *input_paths,
+    ]
+    return {
+        "schema": INSTRUMENTATION_SCHEMA,
+        "input_sha256": {
+            str(path.resolve()): hashlib.sha256(path.read_bytes()).hexdigest()
+            for path in paths
+        },
+        "process_cpu_method": "proc_root_stat_utime_plus_stime",
+        "process_cpu_precision": "SC_CLK_TCK_quantized_not_schedstat_nanoseconds",
+        "thread_cpu_scope": "root_observed_boundary_threads",
+        "remote_process_clock": "unsupported_not_runtime_validated",
+        "resource_scope": "best_effort_non_atomic_process_tree",
+        "observer_clock": "time.monotonic_ns_boundaries_not_atomic_snapshot",
     }
 
 
 def _sanitize_cgroup_name(name: str) -> str:
-    return "".join(char if char.isalnum() or char in ("-", "_", ".") else "_" for char in name)
+    return "".join(
+        char if char.isalnum() or char in ("-", "_", ".") else "_" for char in name
+    )
 
 
 def _read_pid_cgroups(pid: int) -> list[dict[str, Any]]:
     try:
-        lines = Path(f"/proc/{pid}/cgroup").read_text(encoding="utf-8", errors="replace").splitlines()
+        lines = (
+            Path(f"/proc/{pid}/cgroup")
+            .read_text(encoding="utf-8", errors="replace")
+            .splitlines()
+        )
     except OSError:
         return []
     entries = []
@@ -177,7 +474,9 @@ def _read_pid_cgroups(pid: int) -> list[dict[str, Any]]:
         entries.append(
             {
                 "hierarchy": hierarchy,
-                "controllers": [controller for controller in controllers.split(",") if controller],
+                "controllers": [
+                    controller for controller in controllers.split(",") if controller
+                ],
                 "path": relative_path,
             }
         )
@@ -197,7 +496,9 @@ def collect_cgroup_artifacts(pid: int, output_dir: Path) -> dict[str, Any]:
     output_dir.mkdir(parents=True, exist_ok=True)
     entries = _read_pid_cgroups(pid)
     try:
-        raw_cgroup = Path(f"/proc/{pid}/cgroup").read_text(encoding="utf-8", errors="replace")
+        raw_cgroup = Path(f"/proc/{pid}/cgroup").read_text(
+            encoding="utf-8", errors="replace"
+        )
     except OSError as error:
         raw_cgroup = f"unavailable: {error}\n"
     proc_cgroup_path = output_dir / "proc-cgroup.txt"
@@ -235,7 +536,10 @@ def collect_cgroup_artifacts(pid: int, output_dir: Path) -> dict[str, Any]:
                 source = root / file_name
                 target = root_dir / file_name
                 try:
-                    target.write_text(source.read_text(encoding="utf-8", errors="replace"), encoding="utf-8")
+                    target.write_text(
+                        source.read_text(encoding="utf-8", errors="replace"),
+                        encoding="utf-8",
+                    )
                     copied.append(str(target))
                 except OSError as error:
                     unavailable.append(f"{source}: {error}")
@@ -262,24 +566,53 @@ class ResourceSampler:
     _thread: threading.Thread | None = None
     _started_monotonic: float | None = None
     _previous_monotonic: float | None = None
-    _previous_cpu: dict[tuple[int, int], int] = field(default_factory=dict)
+    _previous_cpu: dict[str, int] = field(default_factory=dict)
     _observer_error: str | None = None
+    _observer_error_evidence: dict | None = None
+    _clock_ticks: dict = field(default_factory=dict)
+    _previous_cpu_complete: bool = False
+    _samples_lock: threading.Lock = field(default_factory=threading.Lock)
+    _sealed_samples: list[dict[str, Any]] | None = None
+    _stop_requested_ns: int | None = None
+    _join_finished_ns: int | None = None
+    _thread_exited_ns: int | None = None
+    _last_committed_ns: int | None = None
+    _join_ever_timed_out: bool = False
 
     def start(self) -> None:
         if self.interval_seconds <= 0:
             raise ValueError("resource sample interval must be positive")
         if self._thread is not None:
             raise RuntimeError("resource sampler already started")
+        self._clock_ticks = clock_ticks_evidence()
         self._started_monotonic = time.perf_counter()
-        self._thread = threading.Thread(target=self._run, name="moli-resource-sampler", daemon=True)
+        self._thread = threading.Thread(
+            target=self._run, name="moli-resource-sampler", daemon=True
+        )
         self._thread.start()
 
     def stop(self) -> dict[str, Any]:
+        if self._stop_requested_ns is None:
+            self._stop_requested_ns = time.monotonic_ns()
         self._stop.set()
         if self._thread is not None:
             self._thread.join(timeout=2)
+        alive = self._thread is not None and self._thread.is_alive()
+        self._join_finished_ns = time.monotonic_ns()
+        self._join_ever_timed_out |= alive
+        if not alive:
+            with self._samples_lock:
+                self._sealed_samples = copy.deepcopy(self.samples)
         return self.summary()
 
+    def export_samples(self) -> list[dict[str, Any]]:
+        with self._samples_lock:
+            return copy.deepcopy(
+                self._sealed_samples
+                if self._sealed_samples is not None
+                else self.samples
+            )
+
     def _run(self) -> None:
         try:
             next_capture = time.perf_counter()
@@ -287,73 +620,155 @@ class ResourceSampler:
                 capture_started = time.perf_counter()
                 sample = snapshot_resources(self.root_pid, include_lifetime_cpu=False)
                 captured = time.perf_counter()
-                identities = dict(
-                    identity
-                    for pid in sample.get("pids", [])
-                    if (identity := _read_cpu_identity(int(pid))) is not None
+                rows = sample.get("processes", [])
+                identities = {
+                    row["identity_after"]["identity"]: row["identity_after"][
+                        "total_ticks"
+                    ]
+                    for row in rows
+                    if row["identity_valid"]
+                }
+                complete = bool(rows) and len(identities) == sample["process_count"]
+                common = identities.keys() & self._previous_cpu.keys()
+                regressed = sorted(
+                    key for key in common if identities[key] < self._previous_cpu[key]
                 )
-                cpu_percent = None
-                if self._previous_monotonic is not None:
-                    elapsed = captured - self._previous_monotonic
-                    delta_ticks = sum(
-                        max(0, ticks - self._previous_cpu[identity])
-                        for identity, ticks in identities.items()
-                        if identity in self._previous_cpu
+                delta_ticks = (
+                    None
+                    if regressed
+                    else sum(
+                        identities[key] - self._previous_cpu[key] for key in common
                     )
-                    if elapsed > 0:
-                        try:
-                            clock_ticks = float(os.sysconf("SC_CLK_TCK"))
-                        except (OSError, ValueError):
-                            clock_ticks = 100.0
-                        if clock_ticks > 0:
-                            cpu_percent = delta_ticks / clock_ticks / elapsed * 100.0
+                )
+                elapsed = (
+                    captured - self._previous_monotonic
+                    if self._previous_monotonic is not None
+                    else None
+                )
+                stable = (
+                    complete
+                    and self._previous_cpu_complete
+                    and identities.keys() == self._previous_cpu.keys()
+                )
+                rate = self._clock_ticks.get("ticks_per_second")
+                sample["cpu_percent"] = (
+                    delta_ticks / rate / elapsed * 100
+                    if stable
+                    and delta_ticks is not None
+                    and rate
+                    and elapsed
+                    and elapsed > 0
+                    else None
+                )
+                sample["cpu_percent_method"] = "proc_tree_common_identity_tick_delta"
+                sample["cpu_evidence"] = {
+                    "before": self._previous_cpu,
+                    "after": identities,
+                    "clock_ticks": self._clock_ticks,
+                    "elapsed_seconds": elapsed,
+                    "known_common_delta_ticks": delta_ticks,
+                    "counter_regressions": regressed,
+                    "created_processes": sorted(
+                        identities.keys() - self._previous_cpu.keys()
+                    ),
+                    "retired_processes": sorted(
+                        self._previous_cpu.keys() - identities.keys()
+                    ),
+                    "stable_observed_process_set": stable,
+                    "lifetime_coverage": "not_proven_by_boundary_sampling",
+                }
                 self._previous_monotonic = captured
                 self._previous_cpu = identities
-                sample["cpu_percent"] = cpu_percent
+                self._previous_cpu_complete = complete
                 sample["timestamp"] = time.time()
                 sample["elapsed_ms"] = (
-                    (captured - self._started_monotonic) * 1000.0
+                    (captured - self._started_monotonic) * 1000
                     if self._started_monotonic is not None
                     else 0.0
                 )
-                sample["capture_duration_ms"] = (captured - capture_started) * 1000.0
+                sample["capture_duration_ms"] = (captured - capture_started) * 1000
                 sample["kind"] = "periodic"
-                self.samples.append(sample)
+                with self._samples_lock:
+                    sample["sequence"] = len(self.samples)
+                    self.samples.append(sample)
+                    self._last_committed_ns = time.monotonic_ns()
                 next_capture += self.interval_seconds
                 now = time.perf_counter()
                 if next_capture <= now:
                     missed = int((now - next_capture) / self.interval_seconds) + 1
                     next_capture += missed * self.interval_seconds
-                remaining = max(0.0, next_capture - now)
-                self._stop.wait(remaining)
-        except BaseException as error:
+                self._stop.wait(max(0.0, next_capture - now))
+        except BaseException as error:  # noqa: BLE001 -- Preserve observer termination as evidence.
             self._observer_error = f"{type(error).__name__}: {error}"
+            self._observer_error_evidence = {
+                **error_evidence(error),
+                "traceback": traceback.format_exc(),
+            }
+        finally:
+            self._thread_exited_ns = time.monotonic_ns()
 
     def summary(self) -> dict[str, Any]:
-        pss = [sample["pss_bytes"] for sample in self.samples if sample.get("pss_bytes") is not None]
-        rss = [sample["rss_bytes"] for sample in self.samples if sample.get("rss_bytes") is not None]
-        cpu = [sample["cpu_percent"] for sample in self.samples if sample.get("cpu_percent") is not None]
-        process_counts = [sample["process_count"] for sample in self.samples if sample.get("process_count") is not None]
-        thread_counts = [sample["thread_count"] for sample in self.samples if sample.get("thread_count") is not None]
-        fd_counts = [sample["fd_count"] for sample in self.samples if sample.get("fd_count") is not None]
+        samples = self.export_samples()
+        pss = [
+            sample["pss_bytes"]
+            for sample in samples
+            if sample.get("pss_bytes") is not None
+        ]
+        rss = [
+            sample["rss_bytes"]
+            for sample in samples
+            if sample.get("rss_bytes") is not None
+        ]
+        cpu = [
+            sample["cpu_percent"]
+            for sample in samples
+            if sample.get("cpu_percent") is not None
+        ]
+        process_counts = [
+            sample["process_count"]
+            for sample in samples
+            if sample.get("process_count") is not None
+        ]
+        thread_counts = [
+            sample["thread_count"]
+            for sample in samples
+            if sample.get("thread_count") is not None
+        ]
+        fd_counts = [
+            sample["fd_count"]
+            for sample in samples
+            if sample.get("fd_count") is not None
+        ]
         capture_durations = [
             float(sample["capture_duration_ms"])
-            for sample in self.samples
+            for sample in samples
             if sample.get("capture_duration_ms") is not None
         ]
         intervals = [
             float(current["elapsed_ms"]) - float(previous["elapsed_ms"])
-            for previous, current in zip(self.samples, self.samples[1:])
-            if previous.get("elapsed_ms") is not None and current.get("elapsed_ms") is not None
+            for previous, current in pairwise(samples)
+            if previous.get("elapsed_ms") is not None
+            and current.get("elapsed_ms") is not None
         ]
         complete_pss_samples = sum(
             1
-            for sample in self.samples
+            for sample in samples
             if sample.get("pss_bytes") is not None
             and sample.get("pss_process_count") == sample.get("process_count")
         )
         return {
-            "sample_count": len(self.samples),
+            "instrumentation_schema": INSTRUMENTATION_SCHEMA,
+            "sample_count": len(samples),
+            "sealed": self._sealed_samples is not None,
+            "stop_requested_monotonic_ns": self._stop_requested_ns,
+            "join_finished_monotonic_ns": self._join_finished_ns,
+            "thread_exited_monotonic_ns": self._thread_exited_ns,
+            "last_committed_monotonic_ns": self._last_committed_ns,
+            "last_committed_sample_sequence": len(samples) - 1 if samples else None,
+            "join_ever_timed_out": self._join_ever_timed_out,
+            "observer_error_evidence": self._observer_error_evidence,
+            "scope": "sampler_start_through_stop_request_inflight_completion",
+            "totals_are_final": self._sealed_samples is not None,
             "interval_seconds": self.interval_seconds,
             "peak_pss_bytes": max(pss) if pss else None,
             "peak_rss_bytes": max(rss) if rss else None,
@@ -369,15 +784,25 @@ class ResourceSampler:
                 "max": max(capture_durations) if capture_durations else None,
             },
             "late_sample_count": sum(
-                1 for interval in intervals if interval > self.interval_seconds * 1000.0 * 1.5
+                1
+                for interval in intervals
+                if interval > self.interval_seconds * 1000.0 * 1.5
             ),
             "observed_interval_ms": {
                 "average": sum(intervals) / len(intervals) if intervals else None,
                 "max": max(intervals) if intervals else None,
             },
+            "resource_complete_samples": sum(
+                sample.get("selected_process_reads_complete") is True
+                for sample in samples
+            ),
+            "resource_reads_complete": bool(samples)
+            and all(
+                sample.get("selected_process_reads_complete") is True
+                for sample in samples
+            ),
             "pss_complete_samples": complete_pss_samples,
-            "pss_complete": bool(self.samples)
-            and complete_pss_samples == len(self.samples),
+            "pss_complete": bool(samples) and complete_pss_samples == len(samples),
             "observer_error": self._observer_error,
             "thread_alive_after_stop": bool(
                 self._thread is not None and self._thread.is_alive()

moli-benchmark/moli_benchmark/target_serve.py

@@ -1,23 +1,26 @@
 from __future__ import annotations
 
+import asyncio
 import os
 import shutil
 import subprocess
 import tempfile
 import threading
 import time
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
 
 from .config import REPO_ROOT, ReservedPort, clear_proxy_env, reserve_port
-from .sampling import ResourceSampler
+from .sampling import ResourceSampler, _stat_read, error_evidence
 from .serve import probe_url
 from .synthetic_compare import target_enables_all_resource_fetch, target_metadata
 
 
 class TargetServeError(RuntimeError):
-    pass
+    def __init__(self, message: str, *, evidence: dict | None = None):
+        super().__init__(message)
+        self.evidence = evidence
 
 
 class TargetServeProcessExit(TargetServeError):
@@ -35,11 +38,13 @@ class TargetServeHandle:
     endpoint: str
     command: list[str]
     logs: list[str]
-    sampler: ResourceSampler
+    sampler: ResourceSampler | None
     log_threads: list[threading.Thread]
     port_lease: ReservedPort
     temp_dir: Path | None = None
     ready_ms: float | None = None
+    root_identity: dict | None = None
+    log_state: dict[str, dict] = field(default_factory=dict)
 
 
 def _append_log(logs: list[str], line: str) -> None:
@@ -48,39 +53,57 @@ def _append_log(logs: list[str], line: str) -> None:
         del logs[: len(logs) - 400]
 
 
-def _read_available(stream: Any, label: str, logs: list[str]) -> None:
-    if stream is None:
-        return
+def _read_available(stream: Any, label: str, logs: list[str], state: dict) -> None:
+    state.update(
+        eof=False, error=None, lines_read=0, started_monotonic_ns=time.monotonic_ns()
+    )
     try:
+        if stream is None:
+            state["error"] = {"type": "MissingStream", "message": "pipe unavailable"}
+            return
         while True:
             line = stream.readline()
             if not line:
+                state["eof"] = True
                 return
-            _append_log(logs, f"{label}: {line.decode('utf-8', errors='replace').rstrip()}")
-    except OSError:
-        return
+            state["lines_read"] += 1
+            _append_log(
+                logs, f"{label}: {line.decode('utf-8', errors='replace').rstrip()}"
+            )
+    except Exception as error:  # noqa: BLE001 -- Record failure and continue bounded cleanup.
+        state["error"] = error_evidence(error)
+    finally:
+        state["finished_monotonic_ns"] = time.monotonic_ns()
 
 
-def _terminate_process(process: subprocess.Popen[bytes]) -> bool:
-    if process.poll() is not None:
-        return True
-    try:
-        process.terminate()
-    except OSError:
-        return process.poll() is not None
-    try:
-        process.wait(timeout=2)
-        return True
-    except subprocess.TimeoutExpired:
+def _terminate_process(process: subprocess.Popen[bytes]) -> dict:
+    record = {"actions": [], "returncode_before": process.poll()}
+
+    def action(name: str, function: Any) -> bool:
+        item = {"action": name, "started_monotonic_ns": time.monotonic_ns()}
+        record["actions"].append(item)
         try:
-            process.kill()
-        except OSError:
-            return process.poll() is not None
-        try:
-            process.wait(timeout=2)
+            item["result"] = function()
             return True
-        except subprocess.TimeoutExpired:
+        except (OSError, subprocess.TimeoutExpired) as error:
+            item["error"] = error_evidence(error)
             return False
+        finally:
+            item["finished_monotonic_ns"] = time.monotonic_ns()
+
+    if (
+        record["returncode_before"] is None
+        and action("terminate", process.terminate)
+        and not action("wait_after_terminate", lambda: process.wait(timeout=2))
+        and action("kill", process.kill)
+    ):
+        action("wait_after_kill", lambda: process.wait(timeout=2))
+    record.update(returncode=process.poll(), exited=process.poll() is not None)
+    return record
+
+
+def _root_state(handle: TargetServeHandle) -> dict:
+    return {"returncode": handle.process.poll(), "stat": _stat_read(handle.process.pid)}
 
 
 def _serve_command(
@@ -136,7 +159,11 @@ def start_target_serve(
     sample_interval_seconds: float | None = None,
 ) -> TargetServeHandle:
     engine = target_metadata(target)["engine"]
-    temp_dir = Path(tempfile.mkdtemp(prefix=f"moli-benchmark-{target}-")) if engine == "chrome" else None
+    temp_dir = (
+        Path(tempfile.mkdtemp(prefix=f"moli-benchmark-{target}-"))
+        if engine == "chrome"
+        else None
+    )
     logs: list[str] = []
     reserved_port = reserve_port()
     try:
@@ -157,48 +184,109 @@ def start_target_serve(
         if temp_dir is not None:
             shutil.rmtree(temp_dir, ignore_errors=True)
         raise
-    log_threads = [
-        threading.Thread(target=_read_available, args=(process.stdout, "stdout", logs), daemon=True),
-        threading.Thread(target=_read_available, args=(process.stderr, "stderr", logs), daemon=True),
-    ]
-    for thread in log_threads:
-        thread.start()
-    sampler = (
-        ResourceSampler(process.pid)
-        if sample_interval_seconds is None
-        else ResourceSampler(process.pid, interval_seconds=sample_interval_seconds)
-    )
-    sampler.start()
-    handle = TargetServeHandle(
-        target=target,
-        process=process,
-        endpoint=endpoint,
-        command=command,
-        logs=logs,
-        sampler=sampler,
-        log_threads=log_threads,
-        port_lease=reserved_port,
-        temp_dir=temp_dir,
-    )
-    started = time.perf_counter()
-    deadline = started + timeout_seconds
-    version_url = endpoint + "/json/version"
-    while time.perf_counter() < deadline:
-        if process.poll() is not None:
-            stop_target_serve(handle)
-            raise TargetServeProcessExit(
-                f"{target} CDP server exited early with {process.returncode}: "
-                f"{'; '.join(logs[-20:])}"
+    handle = None
+    stage = "create_handle"
+    try:
+        # Publish cleanup ownership before any post-Popen startup operation.
+        handle = TargetServeHandle(
+            target=target,
+            process=process,
+            endpoint=endpoint,
+            command=command,
+            logs=logs,
+            sampler=None,
+            log_threads=[],
+            port_lease=reserved_port,
+            temp_dir=temp_dir,
+        )
+        stage = "root_identity"
+        handle.root_identity = _stat_read(process.pid)
+        handle.log_state = {label: {} for label in ("stdout", "stderr")}
+        stage = "create_log_threads"
+        for stream, label in ((process.stdout, "stdout"), (process.stderr, "stderr")):
+            handle.log_threads.append(
+                threading.Thread(
+                    target=_read_available,
+                    args=(stream, label, logs, handle.log_state[label]),
+                    daemon=True,
+                )
             )
-        if probe_url(version_url):
-            handle.ready_ms = (time.perf_counter() - started) * 1000.0
-            return handle
-        time.sleep(0.025)
-    stop_target_serve(handle)
-    raise TargetServeReadinessTimeout(
-        f"timed out waiting for {target} CDP server at {version_url}: "
-        f"{'; '.join(logs[-20:])}"
-    )
+        for label, thread in zip(("stdout", "stderr"), handle.log_threads):
+            stage = f"start_{label}_reader"
+            thread.start()
+        stage = "create_sampler"
+        handle.sampler = (
+            ResourceSampler(process.pid)
+            if sample_interval_seconds is None
+            else ResourceSampler(process.pid, interval_seconds=sample_interval_seconds)
+        )
+        stage = "start_sampler"
+        handle.sampler.start()
+        stage = "readiness"
+        started = time.perf_counter()
+        deadline = started + timeout_seconds
+        version_url = endpoint + "/json/version"
+        while time.perf_counter() < deadline:
+            if process.poll() is not None:
+                raise TargetServeProcessExit(
+                    f"{target} CDP server exited early with {process.returncode}: "
+                    f"{'; '.join(logs[-20:])}"
+                )
+            if probe_url(version_url):
+                handle.ready_ms = (time.perf_counter() - started) * 1000.0
+                return handle
+            time.sleep(0.025)
+        raise TargetServeReadinessTimeout(
+            f"timed out waiting for {target} CDP server at {version_url}: "
+            f"{'; '.join(logs[-20:])}"
+        )
+    except BaseException as error:
+        # Returning a handle is not the boundary of resource ownership. Preserve
+        # this original startup error even when individual cleanup actions fail.
+        evidence = {"instrumentation_schema": 2, "events": [], "errors": []}
+        if handle is not None:
+            try:
+                evidence = stop_target_serve(handle, include_resource_samples=True)
+            except BaseException as cleanup_error:  # noqa: BLE001 -- Preserve the original startup exception.
+                evidence["errors"].append(
+                    {"action": "stop_target_serve", **error_evidence(cleanup_error)}
+                )
+        if handle is None or "termination" not in evidence:
+            # Handle construction/cleanup itself failed. Use the same process
+            # termination policy and still attempt the independently owned lease.
+            def cleanup_action(name, function):
+                event = {"action": name, "started_monotonic_ns": time.monotonic_ns()}
+                evidence["events"].append(event)
+                try:
+                    result = function()
+                    event["result"] = result
+                    return result
+                except BaseException as cleanup_error:  # noqa: BLE001 -- Keep original startup failure.
+                    detail = {"action": name, **error_evidence(cleanup_error)}
+                    event["error"] = detail
+                    evidence["errors"].append(detail)
+                    return None
+                finally:
+                    event["finished_monotonic_ns"] = time.monotonic_ns()
+
+            evidence["termination"] = cleanup_action(
+                "terminate_root", lambda: _terminate_process(process)
+            )
+            cleanup_action("release_port", reserved_port.close)
+            if temp_dir is not None:
+                cleanup_action("remove_profile", lambda: shutil.rmtree(temp_dir))
+            evidence.update(
+                cleanup_complete=False,
+                stop_protocol_verified=False,
+                sampler_sealed_before_termination=False,
+                descendants_exited="unknown",
+                returncode=cleanup_action("root_returncode", process.poll),
+            )
+        evidence.update(
+            startup_stage=stage, startup_error=error_evidence(error), command=command
+        )
+        error.evidence = evidence
+        raise
 
 
 def stop_target_serve(
@@ -208,24 +296,171 @@ def stop_target_serve(
 ) -> dict[str, Any]:
     if handle is None:
         return {}
-    process_exited = False
-    resources: dict[str, Any] = {}
+    report = {"instrumentation_schema": 2, "events": [], "errors": []}
+
+    def call(name, function):
+        event = {"action": name, "started_monotonic_ns": time.monotonic_ns()}
+        report["events"].append(event)
+        try:
+            result = function()
+            event["result"] = result
+            return result
+        except Exception as error:  # noqa: BLE001 -- Record failure and continue bounded cleanup.
+            evidence = {"action": name, **error_evidence(error)}
+            report["errors"].append(evidence)
+            event["error"] = evidence
+            return None
+        finally:
+            event["finished_monotonic_ns"] = time.monotonic_ns()
+
+    before = call("root_before_sampler_stop", lambda: _root_state(handle))
+    sampler = handle.sampler
+    report["sampler_constructed"] = sampler is not None
+    resources = (
+        call("stop_and_seal_sampler", sampler.stop) if sampler is not None else None
+    )
+    first_stop = resources
+    after = call("root_after_sampler_stop", lambda: _root_state(handle))
+    # This changes only teardown ordering. Every already-started sample is kept;
+    # no periodic tail row is dropped or excused after an engine exit.
+    termination = call("terminate_root", lambda: _terminate_process(handle.process))
+    if sampler is not None and (
+        resources is None or resources.get("thread_alive_after_stop")
+    ):
+        resources = call("sampler_join_after_process_cleanup", sampler.stop)
+    resources = dict(resources or {})
+    if include_resource_samples:
+        resources["samples"] = (
+            call("snapshot_committed_samples", sampler.export_samples)
+            if sampler is not None
+            else None
+        )
+    reader_states = []
+    for label, thread in zip(("stdout", "stderr"), handle.log_threads):
+        call(f"join_{label}", lambda thread=thread: thread.join(timeout=0.2))
+        reader_states.append(
+            {
+                "stream": label,
+                **handle.log_state.get(label, {}),
+                "thread_alive_after_join": thread.is_alive(),
+            }
+        )
+    call("release_port", handle.port_lease.close)
+    if handle.temp_dir is not None:
+        call("remove_profile", lambda: shutil.rmtree(handle.temp_dir))
+    expected = (handle.root_identity or {}).get("identity")
+    alive = expected is not None and all(
+        point
+        and point["returncode"] is None
+        and point["stat"].get("identity") == expected
+        and point["stat"].get("state") not in ("Z", "X")
+        for point in (before, after)
+    )
+    sealed_before = bool(
+        first_stop
+        and first_stop.get("sealed") is True
+        and first_stop.get("thread_alive_after_stop") is False
+        and not first_stop.get("join_ever_timed_out")
+    )
+    readers_complete = len(reader_states) == 2 and all(
+        state.get("eof") is True
+        and state.get("error") is None
+        and not state["thread_alive_after_join"]
+        for state in reader_states
+    )
+    cleanup = bool(
+        termination
+        and termination["exited"]
+        and resources.get("thread_alive_after_stop") is False
+        and readers_complete
+        and not report["errors"]
+    )
+    report.update(
+        root_identity=handle.root_identity,
+        root_observed_alive_at_stop_boundaries=alive,
+        sampler_sealed_before_termination=sealed_before,
+        stop_protocol_verified=alive
+        and sealed_before
+        and resources.get("observer_error") is None
+        and not report["errors"],
+        cleanup_complete=cleanup,
+        descendants_exited="unknown",
+        termination=termination,
+        log_readers=reader_states,
+        log_lines_dropped=max(
+            0,
+            sum(state.get("lines_read", 0) for state in reader_states)
+            - len(handle.logs),
+        ),
+        returncode=handle.process.returncode,
+        resources=resources,
+        log_tail=list(handle.logs[-40:]),
+    )
+    return report
+
+
+async def finish_target_serve_run(client, handle, record, *, key="server") -> None:
+    """Preserve workload and close failures independently; always stop the target."""
+    record["business_ok"] = bool(record.get("ok"))
+    errors = record.setdefault("cleanup_errors", [])
+    record.setdefault(key, {})
     try:
-        process_exited = _terminate_process(handle.process)
-        if process_exited:
-            for thread in handle.log_threads:
-                thread.join(timeout=0.2)
-        else:
-            _append_log(handle.logs, "process did not exit after SIGKILL; skipped pipe drain")
-        resources = handle.sampler.stop()
-        if include_resource_samples:
-            resources["samples"] = list(handle.sampler.samples)
+        if client is not None:
+            try:
+                await client.websocket.close()
+            except asyncio.CancelledError as error:
+                errors.append({"action": "websocket_close", **error_evidence(error)})
+                raise
+            except Exception as error:  # noqa: BLE001 -- Record failure and continue bounded cleanup.
+                errors.append({"action": "websocket_close", **error_evidence(error)})
     finally:
-        handle.port_lease.close()
-        if handle.temp_dir is not None:
-            shutil.rmtree(handle.temp_dir, ignore_errors=True)
-    return {
-        "returncode": handle.process.returncode,
-        "resources": resources,
-        "log_tail": handle.logs[-40:],
-    }
+        # Cancellation at the awaited close must not skip the synchronous
+        # sampler seal/engine stop attempts or overwrite the workload evidence.
+        try:
+            if handle is not None:
+                record[key] = stop_target_serve(handle, include_resource_samples=True)
+            else:
+                record.setdefault(key, {})
+        except Exception as error:  # noqa: BLE001 -- Record failure and continue bounded cleanup.
+            errors.append({"action": "stop_target_serve", **error_evidence(error)})
+        stopped = record[key]
+        resources = stopped.get("resources", {})
+        record["instrumentation_ok"] = bool(
+            stopped.get("stop_protocol_verified")
+            and stopped.get("cleanup_complete")
+            and resources.get("pss_complete")
+            and resources.get("resource_reads_complete")
+            and not errors
+            and record.get("checkpoint_reads_complete", False)
+            and record.get("cpu_evidence_complete", False)
+            and record.get("affinity_evidence_complete", False)
+        )
+        record["ok"] = record["business_ok"] and record["instrumentation_ok"]
+        # Caller saves the complete failed sample before stopping later cohorts.
+        record["safe_to_continue"] = bool(
+            stopped.get("cleanup_complete")
+            and stopped.get("stop_protocol_verified")
+            and not resources.get("join_ever_timed_out")
+            and not resources.get("observer_error")
+            and not errors
+        )
+
+
+def stop_fixture_server(server, thread) -> dict:
+    record = {"actions": [], "errors": []}
+    for name, action in (
+        ("shutdown", server.shutdown),
+        ("server_close", server.server_close),
+        ("join", lambda: thread.join(timeout=2)),
+    ):
+        event = {"action": name, "started_monotonic_ns": time.monotonic_ns()}
+        record["actions"].append(event)
+        try:
+            action()
+        except Exception as error:  # noqa: BLE001 -- Record failure and continue bounded cleanup.
+            record["errors"].append({"action": name, **error_evidence(error)})
+        finally:
+            event["finished_monotonic_ns"] = time.monotonic_ns()
+    record["thread_alive_after_join"] = thread.is_alive()
+    record["complete"] = not record["thread_alive_after_join"] and not record["errors"]
+    return record

moli-benchmark/tests/test_cdp_failure_evidence.py

@@ -0,0 +1,204 @@
+from __future__ import annotations
+
+import asyncio
+import errno
+import json
+import unittest
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+from moli_benchmark import raw_cdp, target_serve
+
+EXCEPTION_EVENT = {
+    "method": "Runtime.exceptionThrown",
+    "params": {"exceptionDetails": {"text": "observed before command failure"}},
+}
+BINDING_EVENT = {"method": "Runtime.bindingCalled", "params": {"payload": "observed"}}
+
+
+class FakeWebSocket:
+    def __init__(self, incoming, *, close_error=None):
+        self.incoming = list(incoming)
+        self.close_error = close_error
+        self.sent = []
+        self.read_count = 0
+        self.close_count = 0
+
+    async def send(self, message):
+        self.sent.append(json.loads(message))
+
+    async def recv(self):
+        self.read_count += 1
+        if not self.incoming:
+            raise AssertionError("unexpected extra receive")
+        item = self.incoming.pop(0)
+        if isinstance(item, BaseException):
+            raise item
+        return json.dumps(item)
+
+    async def close(self):
+        self.close_count += 1
+        if self.close_error is not None:
+            raise self.close_error
+
+
+class SequentialCdpEvidenceTests(unittest.IsolatedAsyncioTestCase):
+    async def test_success_returns_same_ordered_events_without_extra_receive(self):
+        response = {"id": 7, "result": {"value": 42}}
+        websocket = FakeWebSocket([EXCEPTION_EVENT, BINDING_EVENT, response])
+        reply, seen = await raw_cdp.RawCdpClient(websocket).recv_until_id(7)
+        self.assertEqual(reply, response)
+        self.assertEqual(seen, [EXCEPTION_EVENT, BINDING_EVENT, response])
+        self.assertEqual(websocket.read_count, 3)
+
+    async def test_cdp_error_retains_events_and_error_response(self):
+        failed_reply = {"id": 7, "error": {"code": -32000, "message": "rejected"}}
+        websocket = FakeWebSocket([EXCEPTION_EVENT, BINDING_EVENT, failed_reply])
+        with self.assertRaises(raw_cdp.RawCdpError) as caught:
+            await raw_cdp.RawCdpClient(websocket).recv_until_id(
+                7, method="Runtime.evaluate"
+            )
+        self.assertIs(type(caught.exception), raw_cdp.RawCdpError)
+        self.assertEqual(
+            caught.exception.cdp_receive_evidence,
+            {
+                "message_id": 7,
+                "method": "Runtime.evaluate",
+                "messages": [EXCEPTION_EVENT, BINDING_EVENT, failed_reply],
+            },
+        )
+        self.assertEqual(websocket.read_count, 3)
+        self.assertIn("rejected", str(caught.exception))
+
+    async def test_transport_timeout_and_cancel_preserve_original_exception_object(
+        self,
+    ):
+        for failure in (
+            EOFError("closed after event"),
+            TimeoutError("receive timed out after event"),
+            asyncio.CancelledError("receive canceled after event"),
+        ):
+            with self.subTest(kind=type(failure).__name__):
+                websocket = FakeWebSocket([EXCEPTION_EVENT, failure])
+                with self.assertRaises(type(failure)) as caught:
+                    await raw_cdp.RawCdpClient(websocket).recv_until_id(
+                        7, method="Runtime.evaluate"
+                    )
+                self.assertIs(caught.exception, failure)
+                self.assertEqual(
+                    failure.cdp_receive_evidence["messages"], [EXCEPTION_EVENT]
+                )
+                self.assertEqual(websocket.read_count, 2)
+
+    async def test_deadline_expiry_retains_event_without_another_receive(self):
+        # A fake monotonic clock expires after the first received event; no sleep.
+        loop = SimpleNamespace(time=Mock(side_effect=[0, 0, 11]))
+        client = raw_cdp.RawCdpClient(FakeWebSocket([]))
+
+        async def receive():
+            return EXCEPTION_EVENT
+
+        async def direct_wait(awaitable, *, timeout):
+            self.assertEqual(timeout, 10)
+            return await awaitable
+
+        client.recv = receive
+        with (
+            patch.object(raw_cdp.asyncio, "get_running_loop", return_value=loop),
+            patch.object(raw_cdp.asyncio, "wait_for", side_effect=direct_wait),
+            self.assertRaises(raw_cdp.RawCdpTimeoutError) as caught,
+        ):
+            await client.recv_until_id(7, method="Runtime.evaluate")
+        self.assertEqual(
+            caught.exception.cdp_receive_evidence["messages"], [EXCEPTION_EVENT]
+        )
+
+
+class CancelledCleanupEvidenceTests(unittest.IsolatedAsyncioTestCase):
+    async def test_cancelled_close_runs_real_stop_order_and_preserves_primary_error(
+        self,
+    ):
+        cancellation = asyncio.CancelledError("close canceled")
+        websocket = FakeWebSocket([], close_error=cancellation)
+        calls = []
+        process = SimpleNamespace(pid=10, returncode=None)
+        process.poll = lambda: process.returncode
+
+        def seal():
+            self.assertIsNone(process.returncode)
+            calls.append("seal")
+            return {"sealed": True, "thread_alive_after_stop": False}
+
+        def terminate(_process):
+            calls.append("terminate")
+            process.returncode = 143
+            return {"exited": True, "returncode": 143}
+
+        thread = SimpleNamespace(join=lambda timeout: None, is_alive=lambda: False)
+        handle = SimpleNamespace(
+            process=process,
+            root_identity={"identity": "10:100"},
+            sampler=SimpleNamespace(
+                stop=seal, export_samples=lambda: [{"sequence": 1}]
+            ),
+            logs=[],
+            log_threads=[thread, thread],
+            log_state={
+                name: {"eof": True, "error": None} for name in ("stdout", "stderr")
+            },
+            port_lease=SimpleNamespace(close=lambda: calls.append("release")),
+            temp_dir=None,
+        )
+        record = {"ok": False, "error": "original workload failure"}
+        with (
+            patch.object(
+                target_serve,
+                "_stat_read",
+                return_value={"identity": "10:100", "state": "S"},
+            ),
+            patch.object(target_serve, "_terminate_process", side_effect=terminate),
+            self.assertRaises(asyncio.CancelledError) as caught,
+        ):
+            await target_serve.finish_target_serve_run(
+                SimpleNamespace(websocket=websocket), handle, record
+            )
+        self.assertIs(caught.exception, cancellation)
+        self.assertEqual(calls, ["seal", "terminate", "release"])
+        self.assertEqual(record["error"], "original workload failure")
+        self.assertEqual(record["cleanup_errors"][0]["type"], "CancelledError")
+        self.assertEqual(record["server"]["resources"]["samples"], [{"sequence": 1}])
+        self.assertTrue(record["server"]["cleanup_complete"])
+        self.assertFalse(record["ok"])
+        self.assertFalse(record["safe_to_continue"])
+
+    async def test_stop_error_does_not_replace_close_cancellation_or_primary_error(
+        self,
+    ):
+        cancellation = asyncio.CancelledError("close canceled")
+        record = {"ok": False, "error": "original workload failure"}
+        with (
+            patch.object(
+                target_serve,
+                "stop_target_serve",
+                side_effect=OSError(errno.EIO, "stop failed"),
+            ) as stop,
+            self.assertRaises(asyncio.CancelledError) as caught,
+        ):
+            await target_serve.finish_target_serve_run(
+                SimpleNamespace(websocket=FakeWebSocket([], close_error=cancellation)),
+                object(),
+                record,
+            )
+        self.assertIs(caught.exception, cancellation)
+        stop.assert_called_once()
+        self.assertEqual(record["error"], "original workload failure")
+        self.assertEqual(
+            [item["action"] for item in record["cleanup_errors"]],
+            ["websocket_close", "stop_target_serve"],
+        )
+        self.assertEqual(record["cleanup_errors"][1]["errno"], errno.EIO)
+        self.assertFalse(record["safe_to_continue"])
+
+
+if __name__ == "__main__":
+    unittest.main()

moli-benchmark/tests/test_instrumentation_evidence.py

@@ -0,0 +1,460 @@
+from __future__ import annotations
+
+import asyncio
+import errno
+import io
+import threading
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+from moli_benchmark import sampling, target_serve
+
+
+def stat(pid=10, tid=None, start=100, ticks=5):
+    return {
+        "status": "ok",
+        "identity": f"{pid}:{start}" if tid is None else f"{pid}:{tid}:{start}",
+        "pid": pid,
+        "tid": tid,
+        "start_ticks": start,
+        "state": "S",
+        "ppid": 1,
+        "user_ticks": ticks,
+        "system_ticks": 0,
+        "total_ticks": ticks,
+    }
+
+
+def cpu(threads, ticks=5, identity="10:100"):
+    root = {**stat(ticks=ticks), "identity": identity}
+    return {
+        "threads": threads,
+        "observations": [{"valid": True}],
+        "enumeration_error": None,
+        "root_before": root,
+        "root_after": root,
+    }
+
+
+class InstrumentationEvidenceTests(unittest.TestCase):
+    def test_resource_error_keeps_errno_and_missing_is_not_zero(self):
+        with patch.object(
+            sampling.Path,
+            "read_text",
+            side_effect=OSError(errno.EACCES, "denied", "/proc/10/status"),
+        ):
+            read = sampling._text_read(sampling.Path("/proc/10/status"))
+        self.assertEqual(read["error"]["errno"], errno.EACCES)
+        self.assertEqual(sampling._field_read(read, "VmRSS:")["status"], "os_error")
+        self.assertEqual(
+            sampling._field_read({"status": "ok", "raw": "Pss: 0 kB"}, "Pss:", 1024)[
+                "value"
+            ],
+            0,
+        )
+        self.assertEqual(
+            sampling._field_read({"status": "ok", "raw": "Rss: 1 kB"}, "Pss:")[
+                "status"
+            ],
+            "missing_field",
+        )
+        self.assertEqual(
+            sampling._field_read({"status": "ok", "raw": "Pss: -1 kB"}, "Pss:")[
+                "status"
+            ],
+            "parse_error",
+        )
+
+    def test_stat_keeps_raw_and_uses_last_comm_parenthesis(self):
+        fields = ["S", "1"] + ["0"] * 18
+        fields[11], fields[12], fields[19] = "7", "9", "123"
+        raw = "10 (name with ) spaces) " + " ".join(fields)
+        with patch.object(sampling.Path, "read_text", return_value=raw):
+            value = sampling._stat_read(10)
+        self.assertEqual(value["raw"], raw)
+        self.assertEqual(value["identity"], "10:123")
+        self.assertEqual(value["total_ticks"], 16)
+        with patch.object(
+            sampling.Path, "read_text", return_value=raw.replace("10 (", "11 (", 1)
+        ):
+            self.assertEqual(sampling._stat_read(10)["status"], "parse_error")
+
+    def test_changed_root_identity_invalidates_process_total_and_thread_delta(self):
+        result = sampling.cpu_delta(
+            cpu({"10:7:100": 100}),
+            cpu({"10:7:100": 200}, identity="10:200"),
+            {"ticks_per_second": 100},
+            "10:100",
+        )
+        self.assertIsNone(result["cpu_ms"])
+        self.assertIsNone(result["process_cpu"]["delta_ticks"])
+
+    def test_tid_reuse_and_retired_threads_do_not_corrupt_independent_process_total(
+        self,
+    ):
+        before = cpu({"10:7:100": 1_000_000}, ticks=10)
+        after = cpu({"10:7:200": 200_000}, ticks=13)
+        result = sampling.cpu_delta(before, after, {"ticks_per_second": 100}, "10:100")
+        self.assertEqual(result["retired_threads"], ["10:7:100"])
+        self.assertEqual(result["created_threads"], ["10:7:200"])
+        self.assertEqual(result["cpu_ms"], 0.2)
+        self.assertEqual(result["process_cpu"]["delta_ticks"], 3)
+        self.assertEqual(result["process_cpu"]["cpu_ms"], 30)
+        self.assertEqual(result["thread_cpu_observed"]["before"], before)
+        self.assertEqual(result["thread_cpu_observed"]["after"], after)
+
+    def test_incomplete_thread_map_never_silently_substitutes_process_cpu(self):
+        before, after = cpu({"10:7:100": 1}), cpu({}, ticks=6)
+        after["observations"] = [{"valid": False}]
+        result = sampling.cpu_delta(before, after, {"ticks_per_second": 100}, "10:100")
+        self.assertIsNone(result["cpu_ms"])
+        self.assertEqual(result["process_cpu"]["cpu_ms"], 10)
+        self.assertFalse(result["thread_cpu_observed"]["boundary_read_complete"])
+
+    def test_regression_is_not_clamped_and_clock_failure_is_not_100hz(self):
+        value = sampling.cpu_delta(
+            cpu({"10:7:100": 20}),
+            cpu({"10:7:100": 5}, ticks=4),
+            {"ticks_per_second": 100},
+            "10:100",
+        )
+        self.assertIsNone(value["cpu_ms"])
+        self.assertIsNone(value["process_cpu"]["delta_ticks"])
+        self.assertEqual(
+            value["thread_cpu_observed"]["counter_regressions"], ["10:7:100"]
+        )
+        with patch.object(
+            sampling.os, "sysconf", side_effect=OSError(errno.EINVAL, "bad clock")
+        ):
+            clock = sampling.clock_ticks_evidence()
+        self.assertIsNone(clock["ticks_per_second"])
+        self.assertEqual(clock["error"]["errno"], errno.EINVAL)
+        value = sampling.cpu_delta(cpu({}), cpu({}, ticks=6), clock, "10:100")
+        self.assertIsNone(value["process_cpu"]["cpu_ms"])
+        self.assertEqual(value["process_cpu"]["delta_ticks"], 1)
+
+    def test_zero_ticks_is_not_claimed_as_zero_execution_or_ns_precision(self):
+        value = sampling.cpu_delta(
+            cpu({"10:7:100": 100}),
+            cpu({"10:7:100": 200}),
+            {"ticks_per_second": 100, "counter_tick_ms": 10},
+            "10:100",
+        )
+        self.assertEqual(value["process_cpu"]["cpu_ms"], 0)
+        self.assertEqual(value["cpu_ms"], 0.0001)
+        self.assertEqual(
+            value["process_cpu"]["zero_ticks_mean"],
+            "no_exported_tick_change_not_zero_cpu",
+        )
+        self.assertEqual(
+            value["process_cpu"]["remote_process_clock"]["status"],
+            "unsupported_not_runtime_validated",
+        )
+
+    def test_affinity_has_actual_values_identity_and_errno(self):
+        task = SimpleNamespace(name="7")
+        with (
+            patch.object(sampling.Path, "iterdir", return_value=[task]),
+            patch.object(
+                sampling, "_stat_read", side_effect=lambda pid, tid=None: stat(pid, tid)
+            ),
+            patch.object(sampling.os, "sched_getaffinity", return_value={0}),
+        ):
+            result = sampling.affinity_snapshot(10, 0, "10:100")
+        self.assertTrue(result["all_observed_match"])
+        self.assertEqual(result["threads"][0]["before"]["identity"], "10:7:100")
+        with (
+            patch.object(sampling.Path, "iterdir", return_value=[task]),
+            patch.object(
+                sampling, "_stat_read", side_effect=lambda pid, tid=None: stat(pid, tid)
+            ),
+            patch.object(
+                sampling.os,
+                "sched_getaffinity",
+                side_effect=OSError(errno.ESRCH, "gone"),
+            ),
+        ):
+            result = sampling.affinity_snapshot(10, 0, "10:100")
+        self.assertFalse(result["boundary_complete"])
+        self.assertEqual(result["threads"][0]["error"]["errno"], errno.ESRCH)
+
+    def test_cpu_snapshot_retains_disappeared_thread_observation(self):
+        tasks = [SimpleNamespace(name="7"), SimpleNamespace(name="8")]
+
+        def read(path):
+            if str(path).endswith("8/schedstat"):
+                return {
+                    "status": "os_error",
+                    "raw": None,
+                    "error": {"errno": errno.ENOENT},
+                }
+            return {"status": "ok", "raw": "123 0 0"}
+
+        with (
+            patch.object(sampling.Path, "iterdir", return_value=tasks),
+            patch.object(
+                sampling, "_stat_read", side_effect=lambda pid, tid=None: stat(pid, tid)
+            ),
+            patch.object(sampling, "_text_read", side_effect=read),
+        ):
+            result = sampling.cpu_snapshot(10)
+        self.assertEqual(result["enumerated_tids"], [7, 8])
+        self.assertEqual(result["threads"], {"10:7:100": 123})
+        self.assertFalse(result["observations"][1]["valid"])
+        self.assertEqual(
+            result["observations"][1]["runtime"]["error"]["errno"], errno.ENOENT
+        )
+
+    def test_sampler_seals_after_inflight_sample_and_snapshots_are_independent(self):
+        entered, release = threading.Event(), threading.Event()
+
+        def snapshot(*_args, **_kwargs):
+            entered.set()
+            self.assertTrue(release.wait(1))
+            return {
+                "processes": [{"identity_valid": True, "identity_after": stat()}],
+                "process_count": 1,
+                "pss_bytes": 3,
+                "pss_process_count": 1,
+                "rss_bytes": 4,
+                "thread_count": 1,
+                "fd_count": 1,
+            }
+
+        sampler = sampling.ResourceSampler(10, interval_seconds=0.01)
+        with (
+            patch.object(sampling, "snapshot_resources", side_effect=snapshot),
+            patch.object(
+                sampling, "clock_ticks_evidence", return_value={"ticks_per_second": 100}
+            ),
+        ):
+            sampler.start()
+            self.assertTrue(entered.wait(1))
+            sampler._stop.set()
+            release.set()
+            summary = sampler.stop()
+        self.assertTrue(summary["sealed"])
+        self.assertFalse(summary["thread_alive_after_stop"])
+        self.assertEqual(summary["sample_count"], 1)
+        sample = sampler.export_samples()
+        sample[0]["pss_bytes"] = 999
+        self.assertEqual(sampler.export_samples()[0]["pss_bytes"], 3)
+        self.assertLessEqual(
+            summary["last_committed_monotonic_ns"],
+            summary["thread_exited_monotonic_ns"],
+        )
+        self.assertLessEqual(
+            summary["thread_exited_monotonic_ns"], summary["join_finished_monotonic_ns"]
+        )
+
+    def test_join_timeout_remains_a_failure_after_later_success(self):
+        sampler = sampling.ResourceSampler(10)
+        thread = Mock()
+        thread.is_alive.return_value = True
+        sampler._thread = thread
+        first = sampler.stop()
+        self.assertFalse(first["sealed"])
+        self.assertTrue(first["join_ever_timed_out"])
+        thread.is_alive.return_value = False
+        last = sampler.stop()
+        self.assertTrue(last["sealed"])
+        self.assertTrue(last["join_ever_timed_out"])
+
+    def test_log_read_errors_and_eof_are_distinct(self):
+        state, logs = {}, []
+        target_serve._read_available(io.BytesIO(b"a\nb\n"), "stdout", logs, state)
+        self.assertTrue(state["eof"])
+        self.assertEqual(state["lines_read"], 2)
+        target_serve._read_available(
+            SimpleNamespace(
+                readline=Mock(side_effect=OSError(errno.EIO, "read failed"))
+            ),
+            "stderr",
+            [],
+            state := {},
+        )
+        self.assertFalse(state["eof"])
+        self.assertEqual(state["error"]["errno"], errno.EIO)
+
+    def test_stop_seals_before_term_and_logs_require_eof(self):
+        calls = []
+        process = SimpleNamespace(
+            pid=10, returncode=None, poll=lambda: process.returncode
+        )
+
+        def stop():
+            calls.append("stop")
+            self.assertIsNone(process.returncode)
+            return {
+                "sealed": True,
+                "thread_alive_after_stop": False,
+                "join_ever_timed_out": False,
+                "pss_complete": True,
+            }
+
+        def terminate(_):
+            calls.append("terminate")
+            process.returncode = 143
+            return {"exited": True, "returncode": 143}
+
+        thread = SimpleNamespace(join=lambda timeout: None, is_alive=lambda: False)
+        handle = SimpleNamespace(
+            process=process,
+            root_identity=stat(),
+            sampler=SimpleNamespace(stop=stop, export_samples=list),
+            logs=[],
+            log_threads=[thread, thread],
+            log_state={
+                name: {"eof": True, "error": None, "lines_read": 0}
+                for name in ("stdout", "stderr")
+            },
+            port_lease=SimpleNamespace(close=lambda: calls.append("release")),
+            temp_dir=None,
+        )
+        with (
+            patch.object(target_serve, "_stat_read", return_value=stat()),
+            patch.object(target_serve, "_terminate_process", side_effect=terminate),
+        ):
+            result = target_serve.stop_target_serve(
+                handle, include_resource_samples=True
+            )
+        self.assertEqual(calls, ["stop", "terminate", "release"])
+        self.assertTrue(result["stop_protocol_verified"])
+        self.assertTrue(result["cleanup_complete"])
+        self.assertEqual(result["returncode"], 143)
+        self.assertEqual(result["descendants_exited"], "unknown")
+
+    def test_async_close_failure_does_not_skip_stop_or_overwrite_workload_error(self):
+        async def failed_close():
+            raise OSError(errno.EIO, "socket close failed")
+
+        record = {"ok": False, "error": "original workload failure"}
+        with patch.object(
+            target_serve, "stop_target_serve", return_value={"cleanup_complete": True}
+        ) as stop:
+            asyncio.run(
+                target_serve.finish_target_serve_run(
+                    SimpleNamespace(websocket=SimpleNamespace(close=failed_close)),
+                    object(),
+                    record,
+                )
+            )
+        stop.assert_called_once()
+        self.assertEqual(record["error"], "original workload failure")
+        self.assertEqual(record["cleanup_errors"][0]["errno"], errno.EIO)
+        self.assertFalse(record["ok"])
+        self.assertFalse(record["safe_to_continue"])
+
+    def test_unobserved_tree_root_stays_in_selected_set(self):
+        failed = {
+            "status": "os_error",
+            "identity": None,
+            "error": {"errno": errno.ENOENT},
+        }
+        with (
+            patch.object(sampling.Path, "glob", return_value=[]),
+            patch.object(sampling, "_stat_read", return_value=failed),
+        ):
+            pids, records, evidence = sampling._process_tree_evidence(10)
+        self.assertEqual(pids, [10])
+        self.assertEqual(records[10], failed)
+        self.assertEqual(evidence["all_descendants_observed"], "unknown")
+
+    def test_resource_identity_change_prevents_apparently_complete_totals(self):
+        with (
+            patch.object(
+                sampling,
+                "_process_tree_evidence",
+                return_value=([10], {10: stat()}, {}),
+            ),
+            patch.object(sampling, "_stat_read", return_value=stat(start=200)),
+            patch.object(
+                sampling,
+                "_text_read",
+                return_value={
+                    "status": "ok",
+                    "raw": "Pss: 1 kB\nVmRSS: 1 kB\nThreads: 1",
+                },
+            ),
+            patch.object(
+                sampling, "_fd_read", return_value={"status": "ok", "value": 1}
+            ),
+        ):
+            value = sampling.snapshot_resources(10, include_lifetime_cpu=False)
+        self.assertEqual(value["process_count"], 1)
+        self.assertIsNone(value["pss_bytes"])
+        self.assertEqual(value["pss_process_count"], 0)
+        self.assertFalse(value["processes"][0]["identity_valid"])
+
+    def test_readiness_failure_carries_stop_and_eof_evidence(self):
+        process = SimpleNamespace(
+            pid=10, returncode=None, stdout=io.BytesIO(), stderr=io.BytesIO()
+        )
+        process.poll = lambda: process.returncode
+        process.terminate = lambda: setattr(process, "returncode", 143)
+        process.wait = lambda timeout: process.returncode
+
+        class Thread:
+            def __init__(self, *, target, args, daemon):
+                self.target, self.args = target, args
+
+            def start(self):
+                self.target(*self.args)
+
+            def join(self, timeout):
+                pass
+
+            def is_alive(self):
+                return False
+
+        sampler = SimpleNamespace(
+            start=lambda: None,
+            stop=lambda: {
+                "sealed": True,
+                "thread_alive_after_stop": False,
+                "pss_complete": False,
+            },
+            export_samples=list,
+        )
+        lease = SimpleNamespace(port=23456, release_socket=Mock(), close=Mock())
+        with (
+            patch.object(target_serve, "reserve_port", return_value=lease),
+            patch.object(target_serve, "_stat_read", return_value=stat()),
+            patch.object(target_serve.subprocess, "Popen", return_value=process),
+            patch.object(target_serve.threading, "Thread", Thread),
+            patch.object(target_serve, "ResourceSampler", return_value=sampler),
+            self.assertRaises(target_serve.TargetServeReadinessTimeout) as caught,
+        ):
+            target_serve.start_target_serve("moli", Path("/unused"), 0)
+        value = caught.exception.evidence
+        self.assertTrue(value["sampler_sealed_before_termination"])
+        self.assertTrue(value["cleanup_complete"])
+        self.assertEqual(value["returncode"], 143)
+        self.assertEqual(len(value["log_readers"]), 2)
+        self.assertEqual(value["resources"]["samples"], [])
+        lease.close.assert_called_once()
+
+    def test_fixture_cleanup_joins_even_when_close_fails(self):
+        calls = []
+
+        def failed():
+            calls.append("close")
+            raise OSError(errno.EIO, "close")
+
+        value = target_serve.stop_fixture_server(
+            SimpleNamespace(
+                shutdown=lambda: calls.append("shutdown"), server_close=failed
+            ),
+            SimpleNamespace(
+                join=lambda timeout: calls.append("join"), is_alive=lambda: False
+            ),
+        )
+        self.assertEqual(calls, ["shutdown", "close", "join"])
+        self.assertFalse(value["complete"])
+        self.assertEqual(value["errors"][0]["errno"], errno.EIO)
+
+
+if __name__ == "__main__":
+    unittest.main()

moli-benchmark/tests/test_sampling.py

@@ -9,21 +9,41 @@ from moli_benchmark.sampling import ResourceSampler, snapshot_resources
 
 class ResourceSamplingTests(unittest.TestCase):
     def test_snapshot_does_not_report_partial_process_tree_totals(self) -> None:
+        def stat(pid):
+            return {"identity": f"{pid}:100", "status": "ok", "total_ticks": 0}
+
+        def read(path):
+            raw = (
+                "VmRSS: 1 kB\nThreads: 2"
+                if str(path).endswith("status")
+                else "Pss: 1 kB"
+            )
+            if str(path) == "/proc/11/smaps_rollup":
+                return {"status": "os_error", "error": {"errno": 2}, "raw": None}
+            if str(path) == "/proc/11/status":
+                raw = "VmRSS: 1 kB"
+            return {"status": "ok", "raw": raw}
+
         with (
-            patch("moli_benchmark.sampling._process_tree", return_value=[10, 11]),
-            patch("moli_benchmark.sampling._read_pss_bytes", side_effect=[100, None]),
-            patch("moli_benchmark.sampling._read_rss_bytes", side_effect=[200, 300]),
-            patch("moli_benchmark.sampling._read_thread_count", side_effect=[2, None]),
-            patch("moli_benchmark.sampling._read_fd_count", side_effect=[3, 4]),
+            patch(
+                "moli_benchmark.sampling._process_tree_evidence",
+                return_value=([10, 11], {10: stat(10), 11: stat(11)}, {}),
+            ),
+            patch("moli_benchmark.sampling._text_read", side_effect=read),
+            patch("moli_benchmark.sampling._stat_read", side_effect=stat),
+            patch(
+                "moli_benchmark.sampling._fd_read",
+                return_value={"status": "ok", "value": 3},
+            ),
             patch("moli_benchmark.sampling._read_cpu_percent", return_value=12.5),
         ):
             sample = snapshot_resources(10)
 
         self.assertIsNone(sample["pss_bytes"])
         self.assertEqual(sample["pss_process_count"], 1)
-        self.assertEqual(sample["rss_bytes"], 500)
+        self.assertEqual(sample["rss_bytes"], 2048)
         self.assertIsNone(sample["thread_count"])
-        self.assertEqual(sample["fd_count"], 7)
+        self.assertEqual(sample["fd_count"], 6)
 
     def test_summary_records_sampler_health_and_complete_pss_coverage(self) -> None:
         sampler = ResourceSampler(1, interval_seconds=0.5)
@@ -71,7 +91,11 @@ class ResourceSamplingTests(unittest.TestCase):
             "moli_benchmark.sampling.snapshot_resources",
             side_effect=fail_snapshot,
         ):
-            sampler.start()
+            with patch(
+                "moli_benchmark.sampling.clock_ticks_evidence",
+                return_value={"ticks_per_second": 100},
+            ):
+                sampler.start()
             self.assertTrue(attempted.wait(timeout=1))
             summary = sampler.stop()
 

moli-benchmark/tests/test_target_serve.py

@@ -37,6 +37,9 @@ class _FakeSampler:
     def start(self) -> None:
         return None
 
+    def export_samples(self):
+        return list(self.samples)
+
     def stop(self) -> dict[str, int]:
         return {"pid": self.pid}
 
@@ -75,7 +78,9 @@ class TargetServeTests(unittest.TestCase):
             stop_target_serve(handle)
             self.assertNotIn(port, config._RESERVED_PORTS)
 
-    def test_start_target_serve_enables_layout_and_all_resource_fetch_for_full_moli(self) -> None:
+    def test_start_target_serve_enables_layout_and_all_resource_fetch_for_full_moli(
+        self,
+    ) -> None:
         process = _FakeProcess()
 
         with (

moli-benchmark/tests/test_target_start_failure_evidence.py

@@ -0,0 +1,217 @@
+from __future__ import annotations
+
+import asyncio
+import errno
+import io
+import subprocess
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+from moli_benchmark import sampling, target_serve
+
+
+class TargetStartFailureEvidenceTests(unittest.TestCase):
+    def run_start_failure(
+        self, stage, *, fail_release=False, startup_failure=None, term_timeout=False
+    ):
+        failure = startup_failure or RuntimeError(f"initial {stage} failure")
+        calls = []
+        process = SimpleNamespace(
+            pid=10, returncode=None, stdout=io.BytesIO(), stderr=io.BytesIO()
+        )
+        process.poll = lambda: process.returncode
+
+        def terminate():
+            calls.append("terminate")
+            if not term_timeout:
+                process.returncode = -15
+
+        def wait(timeout):
+            calls.append("wait")
+            if process.returncode is None:
+                raise subprocess.TimeoutExpired("fake-engine", timeout)
+            return process.returncode
+
+        def kill():
+            calls.append("kill")
+            process.returncode = -9
+
+        def release():
+            calls.append("release_port")
+            if fail_release:
+                raise OSError(errno.EIO, "port release failed")
+
+        process.terminate, process.wait, process.kill = terminate, wait, kill
+        lease = SimpleNamespace(
+            port=23456,
+            release_socket=lambda: calls.append("release_socket"),
+            close=release,
+        )
+
+        class Thread:
+            def __init__(self, *, target, args=(), daemon, name=None):
+                self.target, self.args = target, args
+                self.label = args[1] if args else "sampler"
+                self.started = False
+
+            def start(self):
+                calls.append(f"start_{self.label}")
+                if stage == f"start_{self.label}":
+                    raise failure
+                self.started = True
+                # Only fake stdout/stderr BytesIO readers may run. Never sample.
+                if self.label == "sampler":
+                    raise AssertionError("test must fail before starting a sampler")
+                self.target(*self.args)
+
+            def join(self, timeout):
+                calls.append(f"join_{self.label}")
+                if not self.started:
+                    raise RuntimeError("cannot join thread before it is started")
+
+            def is_alive(self):
+                return False
+
+        class Sampler(sampling.ResourceSampler):
+            def stop(self):
+                calls.append("stop_sampler")
+                return super().stop()
+
+        def sampler_factory(pid, **kwargs):
+            calls.append("create_sampler")
+            if stage == "create_sampler":
+                raise failure
+            return Sampler(pid, **kwargs)
+
+        with (
+            patch.object(target_serve, "reserve_port", return_value=lease),
+            patch.object(
+                target_serve,
+                "_stat_read",
+                return_value={"identity": "10:100", "state": "S"},
+            ),
+            patch.object(target_serve.subprocess, "Popen", return_value=process),
+            patch.object(target_serve.threading, "Thread", Thread),
+            patch.object(target_serve, "ResourceSampler", side_effect=sampler_factory),
+            patch.object(
+                sampling, "clock_ticks_evidence", return_value={"ticks_per_second": 100}
+            ),
+            patch.object(
+                target_serve,
+                "probe_url",
+                side_effect=AssertionError("readiness must not run"),
+            ),
+            self.assertRaises(type(failure)) as caught,
+        ):
+            target_serve.start_target_serve("moli", Path("/never-executed"), 10)
+        self.assertIs(caught.exception, failure)
+        evidence = failure.evidence
+        self.assertEqual(evidence["startup_error"]["message"], str(failure))
+        self.assertIn("terminate", calls)
+        self.assertIn("release_port", calls)
+        self.assertLess(calls.index("terminate"), calls.index("release_port"))
+        self.assertFalse(evidence["stop_protocol_verified"])
+        self.assertFalse(evidence["sampler_sealed_before_termination"])
+        self.assertFalse(evidence["cleanup_complete"])
+        return failure, calls, evidence
+
+    def test_sampler_start_failure_records_unstarted_join_but_terminates_and_releases(
+        self,
+    ):
+        _failure, calls, evidence = self.run_start_failure("start_sampler")
+        self.assertEqual(evidence["startup_stage"], "start_sampler")
+        self.assertTrue(evidence["sampler_constructed"])
+        self.assertEqual(calls.count("stop_sampler"), 2)
+        self.assertLess(calls.index("stop_sampler"), calls.index("terminate"))
+        self.assertEqual(evidence["resources"]["samples"], [])
+        self.assertEqual(evidence["returncode"], -15)
+        self.assertEqual(
+            [item["action"] for item in evidence["errors"]],
+            ["stop_and_seal_sampler", "sampler_join_after_process_cleanup"],
+        )
+        self.assertTrue(
+            all(item["type"] == "RuntimeError" for item in evidence["errors"])
+        )
+
+    def test_partial_log_start_failure_records_missing_eof_and_does_not_invent_sampler(
+        self,
+    ):
+        _failure, calls, evidence = self.run_start_failure("start_stderr")
+        self.assertEqual(evidence["startup_stage"], "start_stderr_reader")
+        self.assertFalse(evidence["sampler_constructed"])
+        self.assertNotIn("create_sampler", calls)
+        self.assertIsNone(evidence["resources"]["samples"])
+        self.assertTrue(evidence["log_readers"][0]["eof"])
+        self.assertNotIn("eof", evidence["log_readers"][1])
+        self.assertEqual(evidence["errors"][0]["action"], "join_stderr")
+        self.assertIn("cannot join", evidence["errors"][0]["message"])
+
+    def test_sampler_constructor_failure_still_cleans_engine_and_both_readers(self):
+        _failure, calls, evidence = self.run_start_failure("create_sampler")
+        self.assertEqual(evidence["startup_stage"], "create_sampler")
+        self.assertNotIn("stop_sampler", calls)
+        self.assertFalse(evidence["sampler_constructed"])
+        self.assertIsNone(evidence["resources"]["samples"])
+        self.assertTrue(all(reader["eof"] for reader in evidence["log_readers"]))
+
+    def test_cleanup_errors_and_term_timeout_do_not_mask_initial_failure(self):
+        _failure, calls, evidence = self.run_start_failure(
+            "start_sampler", fail_release=True, term_timeout=True
+        )
+        self.assertIn("kill", calls)
+        self.assertEqual(evidence["returncode"], -9)
+        self.assertEqual(evidence["errors"][-1]["action"], "release_port")
+        self.assertEqual(evidence["errors"][-1]["errno"], errno.EIO)
+        self.assertEqual(
+            evidence["termination"]["actions"][1]["error"]["type"], "TimeoutExpired"
+        )
+
+    def test_startup_cancellation_keeps_same_object_with_cleanup_evidence(self):
+        cancellation = asyncio.CancelledError("startup canceled")
+        failure, _calls, evidence = self.run_start_failure(
+            "start_sampler", startup_failure=cancellation
+        )
+        self.assertIs(failure, cancellation)
+        self.assertEqual(evidence["startup_error"]["type"], "CancelledError")
+        self.assertEqual(evidence["returncode"], -15)
+
+    def test_handle_construction_failure_uses_existing_term_policy_and_records_cleanup_failure(
+        self,
+    ):
+        failure = RuntimeError("cannot initialize handle")
+        process = SimpleNamespace(
+            pid=10, poll=Mock(side_effect=OSError(errno.EIO, "poll failed"))
+        )
+        lease = SimpleNamespace(
+            port=23456,
+            release_socket=Mock(),
+            close=Mock(side_effect=OSError(errno.EIO, "close failed")),
+        )
+        with (
+            patch.object(target_serve, "reserve_port", return_value=lease),
+            patch.object(target_serve.subprocess, "Popen", return_value=process),
+            patch.object(target_serve, "TargetServeHandle", side_effect=failure),
+            patch.object(
+                target_serve,
+                "_terminate_process",
+                return_value={"exited": True, "returncode": -15},
+            ) as terminate,
+            self.assertRaises(RuntimeError) as caught,
+        ):
+            target_serve.start_target_serve("moli", Path("/never-executed"), 10)
+        self.assertIs(caught.exception, failure)
+        terminate.assert_called_once_with(process)
+        lease.close.assert_called_once()
+        self.assertEqual(failure.evidence["startup_stage"], "create_handle")
+        self.assertEqual(
+            [item["action"] for item in failure.evidence["errors"]],
+            ["release_port", "root_returncode"],
+        )
+        self.assertIsNone(failure.evidence["returncode"])
+        self.assertFalse(failure.evidence["cleanup_complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()