diff --git a/benchmarks/real_world/README.md b/benchmarks/real_world/README.md
index 3a6096c..04d96a8 100644
--- a/benchmarks/real_world/README.md
+++ b/benchmarks/real_world/README.md
@@ -82,6 +82,51 @@ HTTP/WebSocket claims addressable by this tool. The output is not a score, and a
 single-review or exploratory set must not be presented as canonical benchmark
 truth.
 
+## Secure AST/runtime comparison protocol
+
+`compare_runtime.py` compares positive observations from isolated runtime
+introspection with secure static results. Runtime output is a **comparator, not
+truth**: `secure_only` and `runtime_only` are disagreements for source review,
+not false positives or false negatives.
+
+A target/baseline comparison consumes four independently generated schema-v1
+artifacts. All four must declare the same app/bootstrap/app-variable/backend
+selection. Runtime records currently require null `app_entry` and
+`bootstrap_entry`, because isolated runtime does not yet execute those options.
+Target and baseline may use different dependency environments, but the secure
+and runtime records within each snapshot must attest the same source, tool,
+lock, immutable runtime image, and SBOM. Runtime provenance additionally binds
+the seccomp policy, sandbox policy, and exact effective invocation. Provenance
+is strictly mode-specific and its canonical digest is retained in the report.
+
+```bash
+python benchmarks/real_world/compare_runtime.py \
+  --secure-target /results/secure-target.json \
+  --runtime-target /results/runtime-target.json \
+  --secure-baseline /results/secure-baseline.json \
+  --runtime-baseline /results/runtime-baseline.json \
+  --output /results/secure-runtime-comparison.json
+```
+
+The report keeps four-artifact operational evidence (success/abstention,
+structured failure phases, inventory strength, timing, and peak RSS) separate
+from the paired-success quality subset. Timing has exact `list` and `impact`
+measurement states; resources have an exact `peak_rss_bytes` state. Each state
+is either `measured` with its unit or `not_measured` with a reason, matching the
+benchmark telemetry convention. Secure inventory strength is exactly one of
+`established`, `conditional`, or `unavailable`; runtime strength is exactly one
+of `runtime_observed`, `runtime_conditional`, or `runtime_unavailable`. A
+successful pair contributes quality only when both timing states and peak RSS
+are measured for both modes. Inventory and impact disagreements are reported
+exactly and with normalized path parameters;
+target/baseline lifecycle deltas are computed separately for secure and runtime
+modes. A failed artifact never contributes partial quality metrics. This tool
+compares one matrix; it does not aggregate or publish a corpus. Comparison
+output is no-clobber and cannot target source artifacts, frozen corpus/review
+files, canonical ground truth, or `benchmarks/results/**`, including through a
+symlinked parent. Its secure publisher currently requires the same Linux/POSIX
+filesystem semantics documented above.
+
 ## Frozen 50-project expansion
 
 Issue #103 adds a disjoint, metadata-only expansion with 50 projects and 100
diff --git a/benchmarks/real_world/compare_runtime.py b/benchmarks/real_world/compare_runtime.py
index 9287318..eafa0f6 100644
--- a/benchmarks/real_world/compare_runtime.py
+++ b/benchmarks/real_world/compare_runtime.py
@@ -14,12 +14,26 @@
 if __package__ in {None, ""}:
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
 
+from benchmarks.real_world._secure_publish import (
+    SecurePathError,
+    ensure_publishable,
+    publish_exclusive_bytes,
+)
 from benchmarks.real_world.benchmark_schema import (
     BenchmarkSchemaError,
     finite_nonnegative,
     strict_json_loads,
 )
 
+HERE = Path(__file__).resolve().parent
+PROJECT_ROOT = HERE.parents[1]
+_FROZEN_FILES = (
+    HERE / "corpus.json",
+    HERE / "adjudicated.jsonl",
+    HERE / "review-a.jsonl",
+    HERE / "review-b.jsonl",
+)
+_FROZEN_ROOTS = (PROJECT_ROOT / "benchmarks" / "results",)
 _FAILURE_PHASES = {
     "dependency",
     "import",
@@ -29,14 +43,61 @@
     "unavailable",
 }
 _PARAMETER = re.compile(r"\{[^{}]+\}")
+_SHA256 = re.compile(r"sha256:[0-9a-f]{64}")
+_IMAGE_DIGEST = re.compile(r"[^\s@]+@sha256:[0-9a-f]{64}")
+_ENTRY_CONFIGURATION_FIELDS = {
+    "app_entry",
+    "bootstrap_entry",
+    "app_variable",
+    "backend",
+}
+_TIMING_FIELDS = {"list", "impact"}
+_RESOURCE_FIELDS = {"peak_rss_bytes"}
+_SECURE_INVENTORY_STATES = {"established", "conditional", "unavailable"}
+_RUNTIME_INVENTORY_STATES = {
+    "runtime_observed",
+    "runtime_conditional",
+    "runtime_unavailable",
+}
+_COMMON_PROVENANCE_FIELDS = {
+    "source_sha256",
+    "tool_sha256",
+    "effective_invocation_sha256",
+    "dependency_lock_sha256",
+    "runtime_image_digest",
+    "runtime_sbom_sha256",
+}
+_RUNTIME_PROVENANCE_FIELDS = {
+    *_COMMON_PROVENANCE_FIELDS,
+    "runtime_seccomp_sha256",
+    "runtime_policy_sha256",
+}
+_ENVIRONMENT_PROVENANCE_FIELDS = {
+    "source_sha256",
+    "tool_sha256",
+    "dependency_lock_sha256",
+    "runtime_image_digest",
+    "runtime_sbom_sha256",
+}
+_RUNTIME_ROLE = "positive_observation_comparator_not_truth"
 
 
 class ComparisonError(ValueError):
     """A paired record is invalid or configured inequitably."""
 
 
+def _canonical_digest(value: object) -> str:
+    encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode(
+        "utf-8"
+    )
+    return f"sha256:{hashlib.sha256(encoded).hexdigest()}"
+
+
 def _load(path: Path) -> tuple[dict[str, Any], str]:
-    raw = path.read_bytes()
+    try:
+        raw = path.read_bytes()
+    except OSError as error:
+        raise ComparisonError(f"could not read {path}: {error}") from error
     try:
         value = strict_json_loads(raw.decode("utf-8"), str(path))
     except (BenchmarkSchemaError, UnicodeError) as exc:
@@ -46,6 +107,88 @@ def _load(path: Path) -> tuple[dict[str, Any], str]:
     return value, f"sha256:{hashlib.sha256(raw).hexdigest()}"
 
 
+def _validate_configuration(configuration: object) -> dict[str, Any]:
+    if not isinstance(configuration, dict):
+        raise ComparisonError("configuration must be an object")
+    expected = {*_ENTRY_CONFIGURATION_FIELDS, "dependency_lock_sha256"}
+    if set(configuration) != expected:
+        raise ComparisonError(
+            "configuration requires exact app/bootstrap/app-variable/backend/lock fields"
+        )
+    for field in ("app_entry", "bootstrap_entry"):
+        value = configuration[field]
+        if value is not None and (not isinstance(value, str) or not value.strip()):
+            raise ComparisonError(f"configuration.{field} must be null or a non-empty string")
+    for field in ("app_variable", "backend"):
+        value = configuration[field]
+        if not isinstance(value, str) or not value.strip():
+            raise ComparisonError(f"configuration.{field} must be a non-empty string")
+    lock_hash = configuration["dependency_lock_sha256"]
+    if not isinstance(lock_hash, str) or not _SHA256.fullmatch(lock_hash):
+        raise ComparisonError("configuration.dependency_lock_sha256 must be a sha256 digest")
+    return configuration
+
+
+def _validate_failure(failure: object) -> None:
+    if not isinstance(failure, dict) or set(failure) != {"phase", "message"}:
+        raise ComparisonError("failed records require exact phase/message metadata")
+    if failure["phase"] not in _FAILURE_PHASES:
+        raise ComparisonError("failed records require a recognized failure phase")
+    if not isinstance(failure["message"], str) or not failure["message"].strip():
+        raise ComparisonError("failure.message must be a non-empty string")
+
+
+def _validate_measurement(value: object, *, field: str, unit: str) -> None:
+    if not isinstance(value, dict):
+        raise ComparisonError(f"{field} measurement state must be an object")
+    if value.get("status") == "measured":
+        if set(value) != {"status", unit}:
+            raise ComparisonError(f"{field} measured state requires only {unit}")
+        try:
+            finite_nonnegative(value[unit], f"{field}.{unit}")
+        except BenchmarkSchemaError as error:
+            raise ComparisonError(f"{field}.{unit} must be finite and non-negative") from error
+    elif value.get("status") == "not_measured":
+        if (
+            set(value) != {"status", "reason"}
+            or not isinstance(value.get("reason"), str)
+            or not value["reason"].strip()
+        ):
+            raise ComparisonError(f"{field} not_measured state requires only a reason")
+    else:
+        raise ComparisonError(f"{field} status must be measured or not_measured")
+
+
+def _validate_telemetry(record: dict[str, Any]) -> None:
+    timing = record["timing"]
+    resources = record["resources"]
+    if not isinstance(timing, dict) or set(timing) != _TIMING_FIELDS:
+        raise ComparisonError("timing requires exact list and impact states")
+    if not isinstance(resources, dict) or set(resources) != _RESOURCE_FIELDS:
+        raise ComparisonError("resources requires exact peak_rss_bytes state")
+    for name in sorted(_TIMING_FIELDS):
+        _validate_measurement(timing[name], field=f"timing.{name}", unit="seconds")
+    _validate_measurement(
+        resources["peak_rss_bytes"],
+        field="resources.peak_rss_bytes",
+        unit="bytes",
+    )
+
+
+def _validate_provenance(value: object, mode: str) -> dict[str, str]:
+    expected = _RUNTIME_PROVENANCE_FIELDS if mode == "runtime" else _COMMON_PROVENANCE_FIELDS
+    if not isinstance(value, dict) or set(value) != expected:
+        raise ComparisonError(f"{mode} provenance requires exact mode-specific fields")
+    for field in expected - {"runtime_image_digest"}:
+        digest = value[field]
+        if not isinstance(digest, str) or not _SHA256.fullmatch(digest):
+            raise ComparisonError(f"provenance.{field} must be a lowercase sha256 digest")
+    image = value["runtime_image_digest"]
+    if not isinstance(image, str) or not _IMAGE_DIGEST.fullmatch(image):
+        raise ComparisonError("provenance.runtime_image_digest must be an immutable image digest")
+    return value
+
+
 def _validate(record: dict[str, Any], expected_mode: str) -> None:  # noqa: PLR0912
     required = {
         "schema_version",
@@ -54,14 +197,11 @@ def _validate(record: dict[str, Any], expected_mode: str) -> None:  # noqa: PLR0
         "status",
         "configuration",
         "timing",
-    }
-    if set(record) - {
-        *required,
-        "failure",
-        "inventory",
-        "impact",
+        "resources",
         "provenance",
-    }:
+    }
+    allowed = {*required, "failure", "inventory", "impact"}
+    if set(record) - allowed:
         raise ComparisonError("record contains unknown top-level fields")
     if required - set(record):
         raise ComparisonError("record is missing required fields")
@@ -71,41 +211,58 @@ def _validate(record: dict[str, Any], expected_mode: str) -> None:  # noqa: PLR0
         raise ComparisonError("snapshot must be target or baseline")
     if record["status"] not in {"success", "failure"}:
         raise ComparisonError("status must be success or failure")
+    configuration = _validate_configuration(record["configuration"])
+    if expected_mode == "runtime" and any(
+        configuration[field] is not None for field in ("app_entry", "bootstrap_entry")
+    ):
+        raise ComparisonError(
+            "runtime records do not support non-null app_entry or bootstrap_entry"
+        )
+    provenance = _validate_provenance(record["provenance"], expected_mode)
+    if provenance["dependency_lock_sha256"] != configuration["dependency_lock_sha256"]:
+        raise ComparisonError("provenance dependency lock does not match configuration")
+    _validate_telemetry(record)
     if record["status"] == "success":
-        if not isinstance(record.get("inventory"), dict) or not isinstance(
-            record.get("impact"), dict
-        ):
+        inventory = record.get("inventory")
+        impact = record.get("impact")
+        if not isinstance(inventory, dict) or not isinstance(impact, dict):
             raise ComparisonError("successful records require inventory and impact")
+        inventory_status = inventory.get("inventory_status")
+        allowed_inventory = (
+            _SECURE_INVENTORY_STATES if expected_mode == "secure" else _RUNTIME_INVENTORY_STATES
+        )
+        if inventory_status not in allowed_inventory:
+            raise ComparisonError(f"successful {expected_mode} inventory status is invalid")
+        _inventory_ids(inventory)
+        _impact_ids(impact)
         if record.get("failure") is not None:
             raise ComparisonError("successful records forbid failure metadata")
     else:
-        failure = record.get("failure")
-        if not isinstance(failure, dict) or failure.get("phase") not in _FAILURE_PHASES:
-            raise ComparisonError("failed records require a recognized failure phase")
+        _validate_failure(record.get("failure"))
         if record.get("inventory") is not None or record.get("impact") is not None:
             raise ComparisonError("failed records forbid partial inventory/impact claims")
-    if not isinstance(record["configuration"], dict):
-        raise ComparisonError("configuration must be an object")
-    timing = record["timing"]
-    if not isinstance(timing, dict):
-        raise ComparisonError("timing values must be finite non-negative numbers")
-    try:
-        for name, value in timing.items():
-            finite_nonnegative(value, f"timing.{name}")
-    except BenchmarkSchemaError as error:
-        raise ComparisonError("timing values must be finite non-negative numbers") from error
 
 
 def _endpoint_id(endpoint: dict[str, Any]) -> str:
     surface = endpoint.get("surface")
     if isinstance(surface, dict):
-        return f"{surface['surface_kind'].upper()} {surface['surface_id']}"
+        kind = surface.get("surface_kind")
+        identity = surface.get("surface_id")
+        if not isinstance(kind, str) or not kind or not isinstance(identity, str) or not identity:
+            raise ComparisonError("surface endpoints require string kind and identity")
+        return f"{kind.upper()} {identity}"
+    if surface is not None:
+        raise ComparisonError("endpoint surface must be an object or null")
     methods = endpoint.get("methods")
     path = endpoint.get("path")
-    if not isinstance(methods, list) or not all(isinstance(item, str) for item in methods):
-        raise ComparisonError("endpoint methods must be strings")
-    if not isinstance(path, str):
-        raise ComparisonError("endpoint path must be a string")
+    if (
+        not isinstance(methods, list)
+        or not methods
+        or not all(isinstance(item, str) and item for item in methods)
+    ):
+        raise ComparisonError("endpoint methods must be non-empty strings")
+    if not isinstance(path, str) or not path:
+        raise ComparisonError("endpoint path must be a non-empty string")
     return f"{','.join(sorted(methods))} {path}"
 
 
@@ -113,18 +270,30 @@ def _inventory_ids(payload: dict[str, Any]) -> set[str]:
     endpoints = payload.get("endpoints")
     if not isinstance(endpoints, list):
         raise ComparisonError("inventory endpoints must be an array")
-    return {_endpoint_id(item) for item in endpoints if isinstance(item, dict)}
+    identities: set[str] = set()
+    for endpoint in endpoints:
+        if not isinstance(endpoint, dict):
+            raise ComparisonError("inventory endpoint entries must be objects")
+        identity = _endpoint_id(endpoint)
+        if identity in identities:
+            raise ComparisonError(f"duplicate inventory endpoint: {identity}")
+        identities.add(identity)
+    return identities
 
 
 def _impact_ids(payload: dict[str, Any]) -> set[str]:
     candidates = payload.get("candidate_endpoints")
     if not isinstance(candidates, list):
         raise ComparisonError("impact candidate_endpoints must be an array")
-    return {
-        _endpoint_id(item["endpoint"])
-        for item in candidates
-        if isinstance(item, dict) and isinstance(item.get("endpoint"), dict)
-    }
+    identities: set[str] = set()
+    for candidate in candidates:
+        if not isinstance(candidate, dict) or not isinstance(candidate.get("endpoint"), dict):
+            raise ComparisonError("impact candidates require endpoint objects")
+        identity = _endpoint_id(candidate["endpoint"])
+        if identity in identities:
+            raise ComparisonError(f"duplicate impact endpoint: {identity}")
+        identities.add(identity)
+    return identities
 
 
 def _normalized(identity: str) -> str:
@@ -134,6 +303,7 @@ def _normalized(identity: str) -> str:
 def _set_metrics(secure: set[str], runtime: set[str]) -> dict[str, Any]:
     intersection = secure & runtime
     union = secure | runtime
+    has_disagreement = secure != runtime
     return {
         "secure_count": len(secure),
         "runtime_count": len(runtime),
@@ -141,29 +311,66 @@ def _set_metrics(secure: set[str], runtime: set[str]) -> dict[str, Any]:
         "secure_only": sorted(secure - runtime),
         "runtime_only": sorted(runtime - secure),
         "jaccard": len(intersection) / len(union) if union else 1.0,
+        "has_disagreement": has_disagreement,
+        "interpretation": "requires_source_adjudication" if has_disagreement else "agreement",
     }
 
 
-def compare(secure_path: Path, runtime_path: Path) -> dict[str, Any]:
-    secure, secure_hash = _load(secure_path)
-    runtime, runtime_hash = _load(runtime_path)
-    _validate(secure, "secure")
-    _validate(runtime, "runtime")
+def _paired_attestations_measured(secure: dict[str, Any], runtime: dict[str, Any]) -> bool:
+    for record in (secure, runtime):
+        if any(record["timing"][name]["status"] != "measured" for name in _TIMING_FIELDS):
+            return False
+        if record["resources"]["peak_rss_bytes"]["status"] != "measured":
+            return False
+    return True
+
+
+def _validate_pair_equivalence(secure: dict[str, Any], runtime: dict[str, Any]) -> None:
     if secure["snapshot"] != runtime["snapshot"]:
         raise ComparisonError("paired records must use the same snapshot")
     if secure["configuration"] != runtime["configuration"]:
-        raise ComparisonError("paired records must use identical entry/backend configuration")
+        raise ComparisonError("paired records must use identical snapshot configuration")
+    for field in _ENVIRONMENT_PROVENANCE_FIELDS:
+        if secure["provenance"][field] != runtime["provenance"][field]:
+            raise ComparisonError(f"paired records have mismatched provenance.{field}")
+
+
+def _compare_loaded(
+    secure: dict[str, Any],
+    runtime: dict[str, Any],
+    *,
+    secure_hash: str,
+    runtime_hash: str,
+) -> dict[str, Any]:
+    _validate_pair_equivalence(secure, runtime)
+    paired_success = secure["status"] == runtime["status"] == "success"
+    quality_eligible = paired_success and _paired_attestations_measured(secure, runtime)
     result: dict[str, Any] = {
         "schema_version": 1,
+        "runtime_role": _RUNTIME_ROLE,
         "snapshot": secure["snapshot"],
-        "paired_success": secure["status"] == runtime["status"] == "success",
+        "paired_success": paired_success,
+        "quality_eligible": quality_eligible,
         "status": {"secure": secure["status"], "runtime": runtime["status"]},
         "failure": {"secure": secure.get("failure"), "runtime": runtime.get("failure")},
         "configuration": secure["configuration"],
         "timing": {"secure": secure["timing"], "runtime": runtime["timing"]},
+        "resources": {"secure": secure["resources"], "runtime": runtime["resources"]},
         "input_hashes": {"secure": secure_hash, "runtime": runtime_hash},
+        "provenance_digests": {
+            "secure": _canonical_digest(secure["provenance"]),
+            "runtime": _canonical_digest(runtime["provenance"]),
+        },
+        "inventory_strength": {
+            "secure": secure.get("inventory", {}).get("inventory_status")
+            if secure["status"] == "success"
+            else None,
+            "runtime": runtime.get("inventory", {}).get("inventory_status")
+            if runtime["status"] == "success"
+            else None,
+        },
     }
-    if not result["paired_success"]:
+    if not quality_eligible:
         result["inventory"] = None
         result["impact_exact"] = None
         result["impact_normalized"] = None
@@ -181,16 +388,258 @@ def compare(secure_path: Path, runtime_path: Path) -> dict[str, Any]:
     return result
 
 
-def main() -> None:
+def compare(secure_path: Path, runtime_path: Path) -> dict[str, Any]:
+    """Compare one snapshot pair while keeping failures in the operational result."""
+    secure, secure_hash = _load(secure_path)
+    runtime, runtime_hash = _load(runtime_path)
+    _validate(secure, "secure")
+    _validate(runtime, "runtime")
+    return _compare_loaded(
+        secure,
+        runtime,
+        secure_hash=secure_hash,
+        runtime_hash=runtime_hash,
+    )
+
+
+def _lifecycle(target: set[str], baseline: set[str]) -> dict[str, Any]:
+    return {
+        "target_count": len(target),
+        "baseline_count": len(baseline),
+        "added": sorted(target - baseline),
+        "removed": sorted(baseline - target),
+        "unchanged_count": len(target & baseline),
+    }
+
+
+def _mode_lifecycle(target: dict[str, Any], baseline: dict[str, Any]) -> dict[str, Any] | None:
+    if target["status"] != "success" or baseline["status"] != "success":
+        return None
+    target_inventory = _inventory_ids(target["inventory"])
+    baseline_inventory = _inventory_ids(baseline["inventory"])
+    target_impact = _impact_ids(target["impact"])
+    baseline_impact = _impact_ids(baseline["impact"])
+    return {
+        "inventory": _lifecycle(target_inventory, baseline_inventory),
+        "impact_exact": _lifecycle(target_impact, baseline_impact),
+        "impact_normalized": _lifecycle(
+            {_normalized(item) for item in target_impact},
+            {_normalized(item) for item in baseline_impact},
+        ),
+    }
+
+
+def _aggregate_peak_rss(
+    records: dict[tuple[str, str], dict[str, Any]], mode: str
+) -> dict[str, Any]:
+    states = [
+        records[(snapshot, mode)]["resources"]["peak_rss_bytes"]
+        for snapshot in ("target", "baseline")
+    ]
+    if all(state["status"] == "measured" for state in states):
+        return {"status": "measured", "bytes": max(state["bytes"] for state in states)}
+    return {"status": "not_measured", "reason": "one_or_more_snapshot_rss_not_measured"}
+
+
+def _operational_summary(records: dict[tuple[str, str], dict[str, Any]]) -> dict[str, Any]:
+    success_count = {"secure": 0, "runtime": 0}
+    failure_phases: dict[str, dict[str, int]] = {"secure": {}, "runtime": {}}
+    inventory_strength: dict[str, dict[str, str | None]] = {"target": {}, "baseline": {}}
+    timing: dict[str, dict[str, dict[str, Any]]] = {"target": {}, "baseline": {}}
+    resources: dict[str, dict[str, dict[str, Any]]] = {"target": {}, "baseline": {}}
+    for (snapshot, mode), record in records.items():
+        timing[snapshot][mode] = record["timing"]
+        resources[snapshot][mode] = record["resources"]
+        if record["status"] == "success":
+            success_count[mode] += 1
+            inventory_strength[snapshot][mode] = record["inventory"]["inventory_status"]
+        else:
+            inventory_strength[snapshot][mode] = None
+            phase = record["failure"]["phase"]
+            failure_phases[mode][phase] = failure_phases[mode].get(phase, 0) + 1
+    return {
+        "artifact_count": len(records),
+        "success_count": success_count,
+        "abstention_count": {mode: 2 - count for mode, count in success_count.items()},
+        "failure_phase_counts": failure_phases,
+        "inventory_strength": inventory_strength,
+        "timing": timing,
+        "resources": resources,
+        "peak_rss_bytes": {
+            mode: _aggregate_peak_rss(records, mode) for mode in ("secure", "runtime")
+        },
+    }
+
+
+def compare_target_baseline(
+    *,
+    secure_target_path: Path,
+    runtime_target_path: Path,
+    secure_baseline_path: Path,
+    runtime_baseline_path: Path,
+) -> dict[str, Any]:
+    """Compare secure/runtime pairs while permitting snapshot-specific environments."""
+    paths = {
+        ("target", "secure"): secure_target_path,
+        ("target", "runtime"): runtime_target_path,
+        ("baseline", "secure"): secure_baseline_path,
+        ("baseline", "runtime"): runtime_baseline_path,
+    }
+    records: dict[tuple[str, str], dict[str, Any]] = {}
+    hashes: dict[tuple[str, str], str] = {}
+    for (snapshot, mode), path in paths.items():
+        record, digest = _load(path)
+        _validate(record, mode)
+        if record["snapshot"] != snapshot:
+            raise ComparisonError(f"{mode} {snapshot} artifact declares {record['snapshot']}")
+        records[(snapshot, mode)] = record
+        hashes[(snapshot, mode)] = digest
+
+    for snapshot in ("target", "baseline"):
+        _validate_pair_equivalence(records[(snapshot, "secure")], records[(snapshot, "runtime")])
+    entry_configuration = {
+        field: records[("target", "secure")]["configuration"][field]
+        for field in sorted(_ENTRY_CONFIGURATION_FIELDS)
+    }
+    for record in records.values():
+        if any(
+            record["configuration"][field] != value for field, value in entry_configuration.items()
+        ):
+            raise ComparisonError(
+                "target/baseline records must use identical entry/backend configuration"
+            )
+
+    pairs = {
+        snapshot: _compare_loaded(
+            records[(snapshot, "secure")],
+            records[(snapshot, "runtime")],
+            secure_hash=hashes[(snapshot, "secure")],
+            runtime_hash=hashes[(snapshot, "runtime")],
+        )
+        for snapshot in ("target", "baseline")
+    }
+    eligible_snapshots = [
+        snapshot for snapshot in ("target", "baseline") if pairs[snapshot]["quality_eligible"]
+    ]
+    quality = {
+        snapshot: {
+            "inventory": pairs[snapshot]["inventory"],
+            "impact_exact": pairs[snapshot]["impact_exact"],
+            "impact_normalized": pairs[snapshot]["impact_normalized"],
+        }
+        for snapshot in eligible_snapshots
+    }
+    return {
+        "schema_version": 1,
+        "protocol": "secure-runtime-target-baseline-v1",
+        "runtime_role": _RUNTIME_ROLE,
+        "configuration": {
+            "entry": entry_configuration,
+            "snapshots": {
+                snapshot: records[(snapshot, "secure")]["configuration"]
+                for snapshot in ("target", "baseline")
+            },
+        },
+        "operational": _operational_summary(records),
+        "snapshot_pairs": pairs,
+        "paired_success_quality": {
+            "eligible_snapshots": eligible_snapshots,
+            "metrics": quality,
+        },
+        "lifecycle": {
+            "secure": _mode_lifecycle(
+                records[("target", "secure")], records[("baseline", "secure")]
+            ),
+            "runtime": _mode_lifecycle(
+                records[("target", "runtime")], records[("baseline", "runtime")]
+            ),
+        },
+        "input_hashes": {
+            snapshot: {mode: hashes[(snapshot, mode)] for mode in ("secure", "runtime")}
+            for snapshot in ("target", "baseline")
+        },
+        "provenance_digests": {
+            snapshot: {
+                mode: _canonical_digest(records[(snapshot, mode)]["provenance"])
+                for mode in ("secure", "runtime")
+            }
+            for snapshot in ("target", "baseline")
+        },
+    }
+
+
+def _write_output(
+    path: Path,
+    result: dict[str, Any],
+    *,
+    input_paths: tuple[Path, ...],
+) -> None:
+    forbidden_files = (*_FROZEN_FILES, *input_paths)
+    try:
+        absolute = path.expanduser().absolute()
+        if absolute in {item.expanduser().absolute() for item in forbidden_files} or any(
+            absolute == root.expanduser().absolute()
+            or absolute.is_relative_to(root.expanduser().absolute())
+            for root in _FROZEN_ROOTS
+        ):
+            raise SecurePathError(f"refusing to target frozen benchmark artifact: {path}")
+        ensure_publishable(
+            path,
+            forbidden_files=forbidden_files,
+            forbidden_roots=_FROZEN_ROOTS,
+        )
+        content = (json.dumps(result, indent=2, sort_keys=True, allow_nan=False) + "\n").encode(
+            "utf-8"
+        )
+        publish_exclusive_bytes(
+            path,
+            content,
+            forbidden_files=forbidden_files,
+            forbidden_roots=_FROZEN_ROOTS,
+        )
+    except (SecurePathError, TypeError, ValueError) as error:
+        raise ComparisonError(f"could not write {path}: {error}") from error
+
+
+def main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--secure", type=Path, required=True)
-    parser.add_argument("--runtime", type=Path, required=True)
+    parser.add_argument("--secure", type=Path)
+    parser.add_argument("--runtime", type=Path)
+    parser.add_argument("--secure-target", type=Path)
+    parser.add_argument("--runtime-target", type=Path)
+    parser.add_argument("--secure-baseline", type=Path)
+    parser.add_argument("--runtime-baseline", type=Path)
     parser.add_argument("--output", type=Path, required=True)
-    args = parser.parse_args()
-    result = compare(args.secure, args.runtime)
-    args.output.parent.mkdir(parents=True, exist_ok=True)
-    args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    args = parser.parse_args(argv)
+    legacy = (args.secure, args.runtime)
+    matrix = (
+        args.secure_target,
+        args.runtime_target,
+        args.secure_baseline,
+        args.runtime_baseline,
+    )
+    input_paths: tuple[Path, ...]
+    try:
+        if all(value is not None for value in legacy) and all(value is None for value in matrix):
+            result = compare(args.secure, args.runtime)
+            input_paths = (args.secure, args.runtime)
+        elif all(value is None for value in legacy) and all(value is not None for value in matrix):
+            result = compare_target_baseline(
+                secure_target_path=args.secure_target,
+                runtime_target_path=args.runtime_target,
+                secure_baseline_path=args.secure_baseline,
+                runtime_baseline_path=args.runtime_baseline,
+            )
+            input_paths = matrix
+        else:
+            parser.error(
+                "pass either --secure/--runtime or all four target/baseline artifact arguments"
+            )
+        _write_output(args.output, result, input_paths=input_paths)
+    except ComparisonError as error:
+        parser.error(str(error))
+    return 0
 
 
 if __name__ == "__main__":
-    main()
+    raise SystemExit(main())
diff --git a/tests/benchmarks/test_compare_runtime.py b/tests/benchmarks/test_compare_runtime.py
index a145620..1e74516 100644
--- a/tests/benchmarks/test_compare_runtime.py
+++ b/tests/benchmarks/test_compare_runtime.py
@@ -1,13 +1,36 @@
 """Frozen secure/runtime comparison protocol tests."""
 
+from __future__ import annotations
+
 import json
-from pathlib import Path
+from typing import TYPE_CHECKING, Any
 
 import pytest
-from benchmarks.real_world.compare_runtime import ComparisonError, compare
+from benchmarks.real_world import compare_runtime
+from benchmarks.real_world.compare_runtime import (
+    ComparisonError,
+    compare,
+    compare_target_baseline,
+    main,
+)
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+H = "sha256:" + "a" * 64
+
 
+def _digest(character: str) -> str:
+    return "sha256:" + character * 64
 
-def _record(mode: str) -> dict[str, object]:
+
+def _record(
+    mode: str,
+    *,
+    snapshot: str = "target",
+    lock: str = H,
+    measured: bool = True,
+) -> dict[str, Any]:
     inventory = {
         "inventory_status": "established" if mode == "secure" else "runtime_observed",
         "endpoints": [
@@ -26,22 +49,48 @@ def _record(mode: str) -> dict[str, object]:
             }
         ]
     }
+    state: dict[str, object]
+    resource: dict[str, object]
+    if measured:
+        state = {"status": "measured", "seconds": 1.0}
+        resource = {"status": "measured", "bytes": 104857600}
+    else:
+        state = {"status": "not_measured", "reason": "collector_unavailable"}
+        resource = {"status": "not_measured", "reason": "collector_unavailable"}
+    configuration: dict[str, object] = {
+        "app_entry": None,
+        "bootstrap_entry": None,
+        "app_variable": "app",
+        "backend": "mypy",
+        "dependency_lock_sha256": lock,
+    }
+    source = _digest("b" if snapshot == "target" else "c")
+    image = f"registry.example/detector@{_digest('d' if snapshot == 'target' else 'e')}"
+    provenance = {
+        "source_sha256": source,
+        "tool_sha256": _digest("f"),
+        "effective_invocation_sha256": _digest("1" if mode == "secure" else "2"),
+        "dependency_lock_sha256": lock,
+        "runtime_image_digest": image,
+        "runtime_sbom_sha256": _digest("3" if snapshot == "target" else "4"),
+    }
+    if mode == "runtime":
+        provenance.update(
+            runtime_seccomp_sha256=_digest("5"),
+            runtime_policy_sha256=_digest("6"),
+        )
     return {
         "schema_version": 1,
         "mode": mode,
-        "snapshot": "target",
+        "snapshot": snapshot,
         "status": "success",
-        "configuration": {
-            "app_entry": "main:app",
-            "bootstrap_entry": None,
-            "app_variable": "app",
-            "backend": "mypy",
-        },
-        "timing": {"list_seconds": 1.0, "impact_seconds": 2.0, "rss_mib": 100},
+        "configuration": configuration,
+        "timing": {"list": dict(state), "impact": dict(state)},
+        "resources": {"peak_rss_bytes": resource},
         "failure": None,
         "inventory": inventory,
         "impact": impact,
-        "provenance": {"artifact": mode},
+        "provenance": provenance,
     }
 
 
@@ -49,23 +98,65 @@ def _write(path: Path, value: object) -> None:
     path.write_text(json.dumps(value), encoding="utf-8")
 
 
+def _matrix_paths(tmp_path: Path) -> dict[tuple[str, str], Path]:
+    paths: dict[tuple[str, str], Path] = {}
+    for snapshot in ("target", "baseline"):
+        lock = _digest("7" if snapshot == "target" else "8")
+        for mode in ("secure", "runtime"):
+            path = tmp_path / f"{mode}-{snapshot}.json"
+            _write(path, _record(mode, snapshot=snapshot, lock=lock))
+            paths[(snapshot, mode)] = path
+    return paths
+
+
+def _compare_matrix(paths: dict[tuple[str, str], Path]) -> dict[str, Any]:
+    return compare_target_baseline(
+        secure_target_path=paths[("target", "secure")],
+        runtime_target_path=paths[("target", "runtime")],
+        secure_baseline_path=paths[("baseline", "secure")],
+        runtime_baseline_path=paths[("baseline", "runtime")],
+    )
+
+
+def _cli_arguments(paths: dict[tuple[str, str], Path], output: Path) -> list[str]:
+    return [
+        "--secure-target",
+        str(paths[("target", "secure")]),
+        "--runtime-target",
+        str(paths[("target", "runtime")]),
+        "--secure-baseline",
+        str(paths[("baseline", "secure")]),
+        "--runtime-baseline",
+        str(paths[("baseline", "runtime")]),
+        "--output",
+        str(output),
+    ]
+
+
 def test_paired_success_preserves_exact_and_normalized_metrics(tmp_path: Path) -> None:
     secure = tmp_path / "secure.json"
     runtime = tmp_path / "runtime.json"
     _write(secure, _record("secure"))
     runtime_record = _record("runtime")
-    runtime_record["inventory"]["endpoints"].append(  # type: ignore[index]
+    runtime_record["inventory"]["endpoints"].append(
         {"methods": ["DELETE"], "path": "/runtime-only", "surface": None}
     )
     _write(runtime, runtime_record)
 
     result = compare(secure, runtime)
 
+    assert result["runtime_role"] == "positive_observation_comparator_not_truth"
     assert result["paired_success"] is True
+    assert result["quality_eligible"] is True
     assert result["inventory"]["runtime_only"] == ["DELETE /runtime-only"]
+    assert result["inventory"]["interpretation"] == "requires_source_adjudication"
     assert result["impact_exact"]["intersection_count"] == 0
     assert result["impact_normalized"]["intersection_count"] == 1
-    assert result["input_hashes"]["secure"].startswith("sha256:")
+    assert result["inventory_strength"] == {
+        "secure": "established",
+        "runtime": "runtime_observed",
+    }
+    assert result["provenance_digests"]["runtime"].startswith("sha256:")
 
 
 def test_failure_phase_abstains_from_quality_metrics(tmp_path: Path) -> None:
@@ -84,19 +175,118 @@ def test_failure_phase_abstains_from_quality_metrics(tmp_path: Path) -> None:
     result = compare(secure, runtime)
 
     assert result["paired_success"] is False
+    assert result["quality_eligible"] is False
     assert result["inventory"] is None
     assert result["failure"]["runtime"]["phase"] == "import"
 
 
-def test_configuration_mismatch_fails_closed(tmp_path: Path) -> None:
+@pytest.mark.parametrize("field", ["app_entry", "bootstrap_entry"])
+def test_runtime_rejects_unsupported_entry_selection(tmp_path: Path, field: str) -> None:
     secure = tmp_path / "secure.json"
     runtime = tmp_path / "runtime.json"
-    _write(secure, _record("secure"))
+    secure_record = _record("secure")
+    runtime_record = _record("runtime")
+    secure_record["configuration"][field] = "main:create_app"
+    runtime_record["configuration"][field] = "main:create_app"
+    _write(secure, secure_record)
+    _write(runtime, runtime_record)
+
+    with pytest.raises(ComparisonError, match="do not support"):
+        compare(secure, runtime)
+
+
+@pytest.mark.parametrize("mode", ["secure", "runtime"])
+def test_absent_provenance_fails_closed(tmp_path: Path, mode: str) -> None:
+    secure = tmp_path / "secure.json"
+    runtime = tmp_path / "runtime.json"
+    records = {name: _record(name) for name in ("secure", "runtime")}
+    records[mode].pop("provenance")
+    _write(secure, records["secure"])
+    _write(runtime, records["runtime"])
+
+    with pytest.raises(ComparisonError, match="missing required"):
+        compare(secure, runtime)
+
+
+def test_spoofed_mode_or_digest_provenance_fails_closed(tmp_path: Path) -> None:
+    secure = tmp_path / "secure.json"
+    runtime = tmp_path / "runtime.json"
+    secure_record = _record("secure")
     runtime_record = _record("runtime")
-    runtime_record["configuration"]["app_entry"] = "other:app"  # type: ignore[index]
+    runtime_record["provenance"] = dict(secure_record["provenance"])
+    _write(secure, secure_record)
     _write(runtime, runtime_record)
+    with pytest.raises(ComparisonError, match="mode-specific"):
+        compare(secure, runtime)
+
+    runtime_record = _record("runtime")
+    runtime_record["provenance"]["runtime_policy_sha256"] = "sha256:" + "A" * 64
+    _write(runtime, runtime_record)
+    with pytest.raises(ComparisonError, match="lowercase sha256"):
+        compare(secure, runtime)
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("timing", {"arbitrary": {"status": "measured", "seconds": 1}}),
+        ("timing", {"list": {"status": "invented"}, "impact": {"status": "invented"}}),
+        ("resources", {"rss_mib": 100}),
+    ],
+)
+def test_arbitrary_telemetry_fails_closed(tmp_path: Path, field: str, value: object) -> None:
+    secure = tmp_path / "secure.json"
+    runtime = tmp_path / "runtime.json"
+    secure_record = _record("secure")
+    secure_record[field] = value
+    _write(secure, secure_record)
+    _write(runtime, _record("runtime"))
+
+    with pytest.raises(ComparisonError):
+        compare(secure, runtime)
+
+
+@pytest.mark.parametrize(
+    ("mode", "status"),
+    [("secure", "invented"), ("runtime", "established"), ("runtime", "invented")],
+)
+def test_arbitrary_inventory_status_fails_closed(tmp_path: Path, mode: str, status: str) -> None:
+    secure = tmp_path / "secure.json"
+    runtime = tmp_path / "runtime.json"
+    records = {name: _record(name) for name in ("secure", "runtime")}
+    records[mode]["inventory"]["inventory_status"] = status
+    _write(secure, records["secure"])
+    _write(runtime, records["runtime"])
+
+    with pytest.raises(ComparisonError, match="inventory status"):
+        compare(secure, runtime)
+
+
+def test_not_measured_attestations_suppress_quality_but_remain_operational(
+    tmp_path: Path,
+) -> None:
+    secure = tmp_path / "secure.json"
+    runtime = tmp_path / "runtime.json"
+    _write(secure, _record("secure", measured=False))
+    _write(runtime, _record("runtime"))
+
+    result = compare(secure, runtime)
 
-    with pytest.raises(ComparisonError, match="identical"):
+    assert result["paired_success"] is True
+    assert result["quality_eligible"] is False
+    assert result["inventory"] is None
+    assert result["timing"]["secure"]["list"]["status"] == "not_measured"
+
+
+def test_not_measured_pair_rejects_malformed_success_artifact(tmp_path: Path) -> None:
+    secure = tmp_path / "secure.json"
+    runtime = tmp_path / "runtime.json"
+    malformed = _record("secure", measured=False)
+    malformed["inventory"].pop("endpoints")
+    _write(secure, malformed)
+    _write(runtime, _record("runtime"))
+
+    with pytest.raises(ComparisonError, match="inventory endpoints must be an array"):
         compare(secure, runtime)
 
 
@@ -123,13 +313,13 @@ def test_all_failure_phases_are_versioned(tmp_path: Path, phase: str) -> None:
 
 
 @pytest.mark.parametrize("invalid", [True, float("nan"), float("inf"), -1.0])
-def test_timing_rejects_bool_non_finite_and_negative_values(
+def test_measurement_rejects_bool_non_finite_and_negative_values(
     tmp_path: Path, invalid: object
 ) -> None:
     secure = tmp_path / "secure.json"
     runtime = tmp_path / "runtime.json"
     secure_record = _record("secure")
-    secure_record["timing"] = {"impact_seconds": invalid}
+    secure_record["timing"]["impact"] = {"status": "measured", "seconds": invalid}
     _write(secure, secure_record)
     _write(runtime, _record("runtime"))
 
@@ -145,3 +335,156 @@ def test_comparison_rejects_duplicate_json_members(tmp_path: Path) -> None:
 
     with pytest.raises(ComparisonError, match="duplicate JSON member"):
         compare(secure, runtime)
+
+
+def test_target_baseline_permits_snapshot_specific_environments(tmp_path: Path) -> None:
+    paths = _matrix_paths(tmp_path)
+
+    result = _compare_matrix(paths)
+
+    assert (
+        result["configuration"]["snapshots"]["target"]["dependency_lock_sha256"]
+        != result["configuration"]["snapshots"]["baseline"]["dependency_lock_sha256"]
+    )
+    assert result["operational"]["peak_rss_bytes"] == {
+        "secure": {"status": "measured", "bytes": 104857600},
+        "runtime": {"status": "measured", "bytes": 104857600},
+    }
+    assert result["paired_success_quality"]["eligible_snapshots"] == ["target", "baseline"]
+    assert set(result["provenance_digests"]) == {"target", "baseline"}
+
+
+def test_matrix_rejects_within_snapshot_environment_mismatch(tmp_path: Path) -> None:
+    paths = _matrix_paths(tmp_path)
+    runtime = _record("runtime", snapshot="target", lock=_digest("9"))
+    _write(paths[("target", "runtime")], runtime)
+
+    with pytest.raises(ComparisonError, match="snapshot configuration"):
+        _compare_matrix(paths)
+
+    runtime = _record("runtime", snapshot="target", lock=_digest("7"))
+    runtime["provenance"]["runtime_sbom_sha256"] = _digest("0")
+    _write(paths[("target", "runtime")], runtime)
+    with pytest.raises(ComparisonError, match="runtime_sbom_sha256"):
+        _compare_matrix(paths)
+
+
+def test_matrix_keeps_failures_operational_and_excludes_them_from_quality(
+    tmp_path: Path,
+) -> None:
+    paths = _matrix_paths(tmp_path)
+    failed = _record("runtime", snapshot="baseline", lock=_digest("8"))
+    failed.update(
+        status="failure",
+        failure={"phase": "dependency", "message": "lock install failed"},
+        inventory=None,
+        impact=None,
+    )
+    _write(paths[("baseline", "runtime")], failed)
+
+    result = _compare_matrix(paths)
+
+    assert result["operational"]["success_count"] == {"secure": 2, "runtime": 1}
+    assert result["operational"]["failure_phase_counts"]["runtime"] == {"dependency": 1}
+    assert result["paired_success_quality"]["eligible_snapshots"] == ["target"]
+    assert result["lifecycle"]["runtime"] is None
+
+
+def test_matrix_rejects_malformed_success_with_failed_counterpart(tmp_path: Path) -> None:
+    paths = _matrix_paths(tmp_path)
+    malformed = _record("runtime", snapshot="target", lock=_digest("7"), measured=False)
+    malformed["impact"] = {}
+    failed = _record("runtime", snapshot="baseline", lock=_digest("8"))
+    failed.update(
+        status="failure",
+        failure={"phase": "dependency", "message": "lock install failed"},
+        inventory=None,
+        impact=None,
+    )
+    _write(paths[("target", "runtime")], malformed)
+    _write(paths[("baseline", "runtime")], failed)
+
+    with pytest.raises(ComparisonError, match="candidate_endpoints must be an array"):
+        _compare_matrix(paths)
+
+
+def test_matrix_rejects_snapshot_or_entry_configuration_mismatch(tmp_path: Path) -> None:
+    paths = _matrix_paths(tmp_path)
+    record = _record("secure", snapshot="target", lock=_digest("8"))
+    _write(paths[("baseline", "secure")], record)
+    with pytest.raises(ComparisonError, match="declares target"):
+        _compare_matrix(paths)
+
+    record = _record("secure", snapshot="baseline", lock=_digest("8"))
+    record["configuration"]["app_variable"] = "application"
+    _write(paths[("baseline", "secure")], record)
+    with pytest.raises(ComparisonError, match="snapshot configuration"):
+        _compare_matrix(paths)
+
+
+def test_duplicate_or_malformed_endpoint_rows_fail_closed(tmp_path: Path) -> None:
+    secure = tmp_path / "secure.json"
+    runtime = tmp_path / "runtime.json"
+    secure_record = _record("secure")
+    secure_record["inventory"]["endpoints"].append(
+        {"methods": ["GET"], "path": "/users/{user_id}", "surface": None}
+    )
+    _write(secure, secure_record)
+    _write(runtime, _record("runtime"))
+    with pytest.raises(ComparisonError, match="duplicate inventory"):
+        compare(secure, runtime)
+
+    secure_record = _record("secure")
+    secure_record["impact"]["candidate_endpoints"] = ["bad"]
+    _write(secure, secure_record)
+    with pytest.raises(ComparisonError, match="endpoint objects"):
+        compare(secure, runtime)
+
+
+def test_cli_matrix_is_no_clobber(tmp_path: Path) -> None:
+    paths = _matrix_paths(tmp_path)
+    output = tmp_path / "comparison.json"
+    arguments = _cli_arguments(paths, output)
+
+    assert main(arguments) == 0
+    result = json.loads(output.read_text(encoding="utf-8"))
+    assert result["runtime_role"] == "positive_observation_comparator_not_truth"
+    with pytest.raises(SystemExit) as raised:
+        main(arguments)
+    assert raised.value.code == 2
+
+
+def test_cli_rejects_existing_and_absent_frozen_files(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    paths = _matrix_paths(tmp_path)
+    existing = tmp_path / "frozen.json"
+    existing.write_text("frozen", encoding="utf-8")
+    absent = tmp_path / "absent-frozen.json"
+    monkeypatch.setattr(compare_runtime, "_FROZEN_FILES", (existing, absent))
+    monkeypatch.setattr(compare_runtime, "_FROZEN_ROOTS", ())
+
+    for output in (existing, absent):
+        with pytest.raises(SystemExit) as raised:
+            main(_cli_arguments(paths, output))
+        assert raised.value.code == 2
+        assert not absent.exists()
+    assert existing.read_text(encoding="utf-8") == "frozen"
+
+
+def test_cli_rejects_frozen_root_and_symlinked_parent(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    paths = _matrix_paths(tmp_path)
+    frozen_root = tmp_path / "results"
+    frozen_root.mkdir()
+    alias = tmp_path / "results-alias"
+    alias.symlink_to(frozen_root, target_is_directory=True)
+    monkeypatch.setattr(compare_runtime, "_FROZEN_FILES", ())
+    monkeypatch.setattr(compare_runtime, "_FROZEN_ROOTS", (frozen_root,))
+
+    for output in (frozen_root / "new.json", alias / "new.json"):
+        with pytest.raises(SystemExit) as raised:
+            main(_cli_arguments(paths, output))
+        assert raised.value.code == 2
+        assert not (frozen_root / "new.json").exists()