diff --git a/.github/scripts/collect_harness_e2e_benchmarks.py b/.github/scripts/collect_harness_e2e_benchmarks.py
index f4d065d4d..724211c07 100644
--- a/.github/scripts/collect_harness_e2e_benchmarks.py
+++ b/.github/scripts/collect_harness_e2e_benchmarks.py
@@ -6,7 +6,7 @@
 import argparse
 import json
 import math
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
@@ -42,6 +42,9 @@ class CollectionConfig:
     execution_event: str
     execution_actor: str
     generated_at: str
+    stack_mode: str = "source"
+    stack_versions: dict[str, str] = field(default_factory=dict)
+    stack_digest: str = ""
 
 
 def load_json(path: Path) -> Any:
@@ -77,8 +80,11 @@ def semantic_result_status(
     passed: bool,
     hard_gate_failures: int,
     technical_failures: int,
+    infra_failures: int = 0,
     complete: bool = True,
 ) -> str:
+    if infra_failures:
+        return "infra_failed"
     if not complete:
         return "incomplete"
     if technical_failures:
@@ -132,6 +138,23 @@ def parse_scenarios(raw: str) -> list[str]:
     return parsed
 
 
+def parse_stack_versions(raw: str) -> dict[str, str]:
+    try:
+        versions = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise CollectionError(f"stack_versions JSON is invalid: {exc}") from exc
+    if not isinstance(versions, dict):
+        raise CollectionError("stack_versions must be a JSON object")
+    parsed: dict[str, str] = {}
+    for worker, version in versions.items():
+        if not isinstance(worker, str) or not worker:
+            raise CollectionError(
+                "stack_versions worker names must be non-empty strings"
+            )
+        parsed[worker] = require_string(version, f"stack_versions[{worker}]")
+    return parsed
+
+
 def discover_contexts(root: Path) -> dict[tuple[str, str], Path]:
     contexts: dict[tuple[str, str], Path] = {}
     if not root.exists():
@@ -153,6 +176,64 @@ def discover_contexts(root: Path) -> dict[tuple[str, str], Path]:
     return contexts
 
 
+def load_deployment(context_path: Path | None) -> dict[str, Any] | None:
+    if context_path is None:
+        return None
+    candidates = (
+        context_path.parent / "deployment.json",
+        context_path.parent.parent / "deployment.json",
+        context_path.parent.parent.parent / "deployment.json",
+    )
+    seen: set[Path] = set()
+    for candidate in candidates:
+        if candidate in seen or not candidate.is_file():
+            continue
+        seen.add(candidate)
+        value = load_json(candidate)
+        if not isinstance(value, dict):
+            raise CollectionError(f"{candidate} must contain an object")
+        return value
+    return None
+
+
+def stack_metadata(
+    config: CollectionConfig,
+    deployment: dict[str, Any] | None,
+) -> tuple[dict[str, Any], dict[str, Any]]:
+    versions = dict(config.stack_versions)
+    digest = config.stack_digest
+    release_version = config.release_version
+    if deployment is not None:
+        deployment_versions = deployment.get("stack_versions")
+        if isinstance(deployment_versions, dict):
+            versions = {
+                str(worker): str(version)
+                for worker, version in deployment_versions.items()
+                if isinstance(worker, str) and isinstance(version, str)
+            }
+        deployment_digest = deployment.get("stack_lock_digest")
+        if isinstance(deployment_digest, str):
+            digest = deployment_digest
+        actual_version = deployment.get("actual_release_version")
+        if isinstance(actual_version, str) and actual_version:
+            release_version = actual_version
+
+    return (
+        {
+            "tag": config.release_tag,
+            "worker": config.release_worker,
+            "version": release_version,
+            "url": config.release_url,
+            "registry_tag": config.registry_tag,
+        },
+        {
+            "mode": config.stack_mode,
+            "versions": versions,
+            "lock_digest": digest,
+        },
+    )
+
+
 def compact_extra(value: dict[str, Any]) -> str:
     return json.dumps(value, separators=(",", ":"), sort_keys=True)
 
@@ -227,6 +308,8 @@ def collect(
     efficiency: list[dict[str, Any]] = []
     snapshot_subjects: list[dict[str, Any]] = []
     execution_reports: list[dict[str, Any]] = []
+    stack_observations: list[dict[str, Any]] = []
+    release_observations: list[dict[str, Any]] = []
     execution_id = f"{config.execution_run_id}-{config.execution_attempt}"
     execution = {
         "id": execution_id,
@@ -245,15 +328,23 @@ def collect(
         report_count = 0
         hard_gate_failures = 0
         technical_failures = 0
+        infra_failures = 0
         retries = 0
         engine_revisions: set[str] = set()
         resolved_judge: dict[str, Any] | None = None
 
         for scenario_id in config.scenarios:
             context_path = contexts.get((subject["id"], scenario_id))
+            deployment = load_deployment(context_path)
             report_path = (
                 context_path.parent / "results.json" if context_path is not None else None
             )
+            release_metadata, stack = stack_metadata(config, deployment)
+            if stack["versions"] or stack["lock_digest"]:
+                if stack not in stack_observations:
+                    stack_observations.append(stack)
+            if deployment is not None and release_metadata not in release_observations:
+                release_observations.append(release_metadata)
             base_extra: dict[str, Any] = {
                 "schema_version": SCHEMA_VERSION,
                 "execution": execution,
@@ -265,13 +356,8 @@ def collect(
                     "repository": config.repository,
                 },
                 "workflow_url": config.workflow_url,
-                "release": {
-                    "tag": config.release_tag,
-                    "worker": config.release_worker,
-                    "version": config.release_version,
-                    "url": config.release_url,
-                    "registry_tag": config.registry_tag,
-                },
+                "release": release_metadata,
+                "stack": stack,
                 "subject": subject,
                 "judge": {
                     "model": config.judge_model,
@@ -282,17 +368,25 @@ def collect(
             }
 
             if report_path is None or not report_path.is_file():
-                execution_reports.append(
-                    {
-                        "subject_id": subject["id"],
-                        "scenario_id": scenario_id,
-                        "available": False,
-                        "report": None,
-                    }
+                execution_report = {
+                    "subject_id": subject["id"],
+                    "scenario_id": scenario_id,
+                    "available": False,
+                    "report": None,
+                }
+                if deployment is not None:
+                    execution_report["deployment"] = deployment
+                execution_reports.append(execution_report)
+                status = (
+                    "infra_failed"
+                    if deployment is not None
+                    and deployment.get("status") == "infra_failed"
+                    else "missing_report"
                 )
+                infra_failures += int(status == "infra_failed")
                 scenario_snapshot = {
                     "id": scenario_id,
-                    "status": "missing_report",
+                    "status": status,
                     "passed": False,
                     "threshold": None,
                     "runs": 0,
@@ -300,6 +394,7 @@ def collect(
                     "pass_rate": None,
                     "hard_gate_failures": None,
                     "technical_failures": None,
+                    "infra_failures": int(status == "infra_failed"),
                     "retries": None,
                     "total_cost_usd": None,
                     "wall_time_seconds": None,
@@ -312,9 +407,21 @@ def collect(
                         "missing_reports",
                         "count",
                         1,
-                        {**base_extra, "passed": False, "status": "missing_report"},
+                        {**base_extra, "passed": False, "status": status},
                     )
                 )
+                if status == "infra_failed":
+                    efficiency.append(
+                        metric(
+                            "reliability",
+                            subject["id"],
+                            scenario_id,
+                            "infra_failed",
+                            "count",
+                            1,
+                            {**base_extra, "passed": False, "status": status},
+                        )
+                    )
                 subject_costs.append(None)
                 subject_wall_times.append(None)
                 scenario_snapshots.append(scenario_snapshot)
@@ -326,14 +433,15 @@ def collect(
                 scenario_id=scenario_id,
                 path=report_path,
             )
-            execution_reports.append(
-                {
-                    "subject_id": subject["id"],
-                    "scenario_id": scenario_id,
-                    "available": True,
-                    "report": report,
-                }
-            )
+            execution_report = {
+                "subject_id": subject["id"],
+                "scenario_id": scenario_id,
+                "available": True,
+                "report": report,
+            }
+            if deployment is not None:
+                execution_report["deployment"] = deployment
+            execution_reports.append(execution_report)
             report_count += 1
             report_passed = bool(scenario.get("passed"))
             subject_passed += int(report_passed)
@@ -363,6 +471,9 @@ def collect(
                     f"{report_path}: technical_failures",
                 )
             )
+            deployment_infra_failed = int(
+                deployment is not None and deployment.get("status") == "infra_failed"
+            )
             retry_count = sum(
                 len(run.get("retry_attempts", []))
                 for run in runs
@@ -407,6 +518,7 @@ def collect(
                 passed=report_passed,
                 hard_gate_failures=hard_gates,
                 technical_failures=technical,
+                infra_failures=deployment_infra_failed,
             )
             extra = {
                 **base_extra,
@@ -485,6 +597,15 @@ def collect(
                         0,
                         extra,
                     ),
+                    metric(
+                        "reliability",
+                        subject["id"],
+                        scenario_id,
+                        "infra_failed",
+                        "count",
+                        0,
+                        extra,
+                    ),
                     metric(
                         "efficiency",
                         subject["id"],
@@ -535,6 +656,7 @@ def collect(
 
             hard_gate_failures += hard_gates
             technical_failures += technical
+            infra_failures += deployment_infra_failed
             retries += retry_count
             subject_costs.append(total_cost)
             subject_wall_times.append(wall_time_seconds)
@@ -549,6 +671,7 @@ def collect(
                     "pass_rate": pass_rate,
                     "hard_gate_failures": hard_gates,
                     "technical_failures": technical,
+                    "infra_failures": deployment_infra_failed,
                     "retries": retry_count,
                     "total_cost_usd": total_cost,
                     "wall_time_seconds": wall_time_seconds,
@@ -566,16 +689,23 @@ def collect(
             all_reports_present
             and subject_passed == expected_count
             and technical_failures == 0
+            and infra_failures == 0
         )
         suite_status = semantic_result_status(
             passed=suite_passed,
             hard_gate_failures=hard_gate_failures,
             technical_failures=technical_failures,
+            infra_failures=infra_failures,
             complete=all_reports_present,
         )
         engine_revision = (
             next(iter(engine_revisions)) if len(engine_revisions) == 1 else None
         )
+        suite_release, suite_stack = stack_metadata(config, None)
+        if len(release_observations) == 1:
+            suite_release = release_observations[0]
+        if len(stack_observations) == 1:
+            suite_stack = stack_observations[0]
         suite_extra = {
             "schema_version": SCHEMA_VERSION,
             "execution": execution,
@@ -587,13 +717,8 @@ def collect(
                 "repository": config.repository,
             },
             "workflow_url": config.workflow_url,
-            "release": {
-                "tag": config.release_tag,
-                "worker": config.release_worker,
-                "version": config.release_version,
-                "url": config.release_url,
-                "registry_tag": config.registry_tag,
-            },
+            "release": suite_release,
+            "stack": suite_stack,
             "subject": subject,
             "judge": resolved_judge
             or {"model": config.judge_model, "provider": config.judge_provider},
@@ -604,6 +729,7 @@ def collect(
             "status": suite_status,
             "expected_reports": expected_count,
             "received_reports": report_count,
+            "infra_failures": infra_failures,
         }
         quality.extend(
             [
@@ -665,6 +791,15 @@ def collect(
                     missing_reports,
                     suite_extra,
                 ),
+                metric(
+                    "reliability",
+                    subject["id"],
+                    "suite",
+                    "infra_failed",
+                    "count",
+                    infra_failures,
+                    suite_extra,
+                ),
             ]
         )
         if total_cost is not None:
@@ -704,6 +839,7 @@ def collect(
                 "report_coverage": report_coverage / 100,
                 "hard_gate_failures": hard_gate_failures,
                 "technical_failures": technical_failures,
+                "infra_failures": infra_failures,
                 "retry_attempts": retries,
                 "total_cost_usd": total_cost,
                 "wall_time_seconds": total_wall_time,
@@ -711,6 +847,11 @@ def collect(
             }
         )
 
+    snapshot_release, snapshot_stack = stack_metadata(config, None)
+    if len(release_observations) == 1:
+        snapshot_release = release_observations[0]
+    if len(stack_observations) == 1:
+        snapshot_stack = stack_observations[0]
     snapshot = {
         "schema_version": SCHEMA_VERSION,
         "execution": execution,
@@ -722,13 +863,9 @@ def collect(
             "repository": config.repository,
         },
         "workflow_url": config.workflow_url,
-        "release": {
-            "tag": config.release_tag,
-            "worker": config.release_worker,
-            "version": config.release_version,
-            "url": config.release_url,
-            "registry_tag": config.registry_tag,
-        },
+        "release": snapshot_release,
+        "stack": snapshot_stack,
+        "stack_observations": stack_observations,
         "requested_runs": config.requested_runs,
         "subjects": snapshot_subjects,
     }
@@ -775,6 +912,9 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--release-version", default="")
     parser.add_argument("--release-url", default="")
     parser.add_argument("--registry-tag", default="")
+    parser.add_argument("--stack-mode", default="source")
+    parser.add_argument("--stack-versions", default="{}")
+    parser.add_argument("--stack-digest", default="")
     parser.add_argument("--judge-model", required=True)
     parser.add_argument("--judge-provider", required=True)
     parser.add_argument("--execution-run-id", required=True)
@@ -815,6 +955,9 @@ def main(argv: list[str] | None = None) -> int:
         execution_event=args.execution_event,
         execution_actor=args.execution_actor,
         generated_at=generated_at,
+        stack_mode=require_string(args.stack_mode, "stack mode"),
+        stack_versions=parse_stack_versions(args.stack_versions),
+        stack_digest=args.stack_digest,
     )
     quality, efficiency, snapshot, execution = collect(config)
     write_outputs(args.output_dir, quality, efficiency, snapshot, execution)
diff --git a/.github/scripts/registry_stack_identity.py b/.github/scripts/registry_stack_identity.py
new file mode 100644
index 000000000..2a75a5c3a
--- /dev/null
+++ b/.github/scripts/registry_stack_identity.py
@@ -0,0 +1,73 @@
+#!/usr/bin/env python3
+"""Create a deterministic identity for a Registry-resolved iii.lock."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import re
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+
+WORKER_NAME = re.compile(r"^[a-z0-9][a-z0-9_-]*$")
+EXACT_VERSION = re.compile(
+    r"^[0-9]+\.[0-9]+\.[0-9]+(-(experimental|alpha|beta))?$"
+)
+
+
+def stack_identity(lock_path: Path) -> dict[str, Any]:
+    try:
+        document = yaml.safe_load(lock_path.read_text()) or {}
+    except (OSError, yaml.YAMLError) as error:
+        raise SystemExit(f"invalid_lock: cannot read {lock_path}: {error}") from error
+
+    workers = document.get("workers") if isinstance(document, dict) else None
+    if not isinstance(workers, dict) or not workers:
+        raise SystemExit("invalid_lock: workers must be a non-empty mapping")
+
+    versions: dict[str, str] = {}
+    for worker, record in sorted(workers.items()):
+        if not isinstance(worker, str) or not WORKER_NAME.fullmatch(worker):
+            raise SystemExit(f"invalid_lock: invalid worker name {worker!r}")
+        if not isinstance(record, dict):
+            raise SystemExit(f"invalid_lock: {worker} must be a mapping")
+        version = record.get("version")
+        if not isinstance(version, str) or not EXACT_VERSION.fullmatch(version):
+            raise SystemExit(
+                f"invalid_lock: {worker} must have an exact published version"
+            )
+        versions[worker] = version
+
+    if "harness" not in versions:
+        raise SystemExit("invalid_lock: harness is required in the resolved stack")
+
+    return {
+        "schema_version": 1,
+        "lock_digest": hashlib.sha256(lock_path.read_bytes()).hexdigest(),
+        "stack_versions": versions,
+    }
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--lock", type=Path, required=True)
+    parser.add_argument("--output", type=Path)
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    identity = stack_identity(args.lock)
+    rendered = json.dumps(identity, sort_keys=True)
+    if args.output:
+        args.output.parent.mkdir(parents=True, exist_ok=True)
+        args.output.write_text(rendered + "\n")
+    print(rendered)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/.github/scripts/tests/test_collect_harness_e2e_benchmarks.py b/.github/scripts/tests/test_collect_harness_e2e_benchmarks.py
index cfe07a54a..ae948f77c 100644
--- a/.github/scripts/tests/test_collect_harness_e2e_benchmarks.py
+++ b/.github/scripts/tests/test_collect_harness_e2e_benchmarks.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+from dataclasses import replace
 from pathlib import Path
 
 import pytest
@@ -57,6 +58,15 @@ def test_semantic_result_status_uses_blocking_precedence() -> None:
         )
         == "passed"
     )
+    assert (
+        semantic_result_status(
+            passed=True,
+            hard_gate_failures=0,
+            technical_failures=0,
+            infra_failures=1,
+        )
+        == "infra_failed"
+    )
 
 
 @pytest.mark.parametrize(
@@ -338,6 +348,51 @@ def test_missing_report_is_visible_and_totals_are_not_fabricated(
     }
 
 
+def test_registry_identity_is_recorded_and_registry_failure_is_infra_failed(
+    tmp_path: Path,
+) -> None:
+    write_report(tmp_path, subject_id="glm", scenario_id="direct_answer")
+    directory = tmp_path / "harness-e2e-glm-security_review-results"
+    directory.mkdir()
+    (directory / "benchmark-context.json").write_text(
+        json.dumps({"subject_id": "glm", "scenario_id": "security_review"})
+    )
+    (directory / "deployment.json").write_text(
+        json.dumps(
+            {
+                "status": "infra_failed",
+                "actual_release_version": "1.8.0",
+                "stack_versions": {"harness": "1.8.0", "state": "0.22.0"},
+                "stack_lock_digest": "a" * 64,
+            }
+        )
+    )
+
+    _, efficiency, snapshot, execution = collect(
+        replace(
+            config(tmp_path, ["direct_answer", "security_review"]),
+            stack_mode="registry",
+        )
+    )
+    efficiency_by_name = by_name(efficiency)
+    security_extra = json.loads(
+        efficiency_by_name[
+            "reliability::glm::security_review::infra_failed"
+        ]["extra"]
+    )
+
+    assert security_extra["status"] == "infra_failed"
+    assert security_extra["release"]["version"] == "1.8.0"
+    assert security_extra["stack"]["versions"] == {
+        "harness": "1.8.0",
+        "state": "0.22.0",
+    }
+    assert security_extra["stack"]["lock_digest"] == "a" * 64
+    assert snapshot["stack"] == security_extra["stack"]
+    assert snapshot["subjects"][0]["infra_failures"] == 1
+    assert execution["reports"][1]["deployment"]["status"] == "infra_failed"
+
+
 def test_unknown_cost_is_omitted_instead_of_recorded_as_zero(
     tmp_path: Path,
 ) -> None:
diff --git a/.github/scripts/tests/test_registry_stack_identity.py b/.github/scripts/tests/test_registry_stack_identity.py
new file mode 100644
index 000000000..f9bd417d9
--- /dev/null
+++ b/.github/scripts/tests/test_registry_stack_identity.py
@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+import yaml
+
+
+SCRIPT = Path(__file__).parents[1] / "registry_stack_identity.py"
+
+
+def run_identity(
+    tmp_path: Path, workers: dict[str, dict[str, str]]
+) -> subprocess.CompletedProcess[str]:
+    lock = tmp_path / "iii.lock"
+    lock.write_text(yaml.safe_dump({"workers": workers}, sort_keys=True))
+    return subprocess.run(
+        [sys.executable, str(SCRIPT), "--lock", str(lock)],
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+
+
+def test_returns_exact_versions_and_lock_digest(tmp_path: Path) -> None:
+    workers = {
+        "harness": {"version": "1.8.0"},
+        "state": {"version": "0.22.0"},
+    }
+    result = run_identity(tmp_path, workers)
+
+    assert result.returncode == 0, result.stderr
+    identity = json.loads(result.stdout)
+    lock = tmp_path / "iii.lock"
+    assert identity["stack_versions"] == {"harness": "1.8.0", "state": "0.22.0"}
+    assert identity["lock_digest"] == hashlib.sha256(lock.read_bytes()).hexdigest()
+
+
+@pytest.mark.parametrize(
+    ("workers", "message"),
+    [
+        ({"state": {"version": "0.22.0"}}, "harness is required"),
+        ({"harness": {"version": "latest"}}, "exact published version"),
+    ],
+)
+def test_rejects_non_identity_locks(
+    tmp_path: Path,
+    workers: dict[str, dict[str, str]],
+    message: str,
+) -> None:
+    result = run_identity(tmp_path, workers)
+
+    assert result.returncode != 0
+    assert message in result.stderr
diff --git a/.github/workflows/_harness-e2e.yml b/.github/workflows/_harness-e2e.yml
index 85f116e8c..a9c6f6b96 100644
--- a/.github/workflows/_harness-e2e.yml
+++ b/.github/workflows/_harness-e2e.yml
@@ -88,6 +88,16 @@ on:
         required: false
         type: string
         default: '{}'
+      stack_digest:
+        description: SHA-256 digest of the resolved registry lock
+        required: false
+        type: string
+        default: ''
+      resolve_stack:
+        description: Resolve the live registry stack from latest and freeze its lock
+        required: false
+        type: boolean
+        default: false
       validation_profile:
         description: Scenario profile resolved for this execution
         required: false
@@ -195,6 +205,7 @@ jobs:
           RELEASE_WORKER: ${{ inputs.release_worker }}
           RELEASE_VERSION: ${{ inputs.release_version }}
           STACK_VERSIONS: ${{ inputs.stack_versions }}
+          RESOLVE_STACK: ${{ inputs.resolve_stack }}
           COVERAGE: ${{ inputs.coverage }}
         run: |
           set -euo pipefail
@@ -215,16 +226,22 @@ jobs:
               [[ "$CLI_CHANNEL" == "latest" || "$CLI_CHANNEL" == "next" ]]
               [[ "$REGISTRY_TAG" =~ ^[A-Za-z0-9._-]+$ ]]
               [[ "$RELEASE_WORKER" =~ ^[A-Za-z0-9_-]+$ ]]
-              [[ -n "$RELEASE_VERSION" ]]
-              jq -e --arg worker "$RELEASE_WORKER" --arg version "$RELEASE_VERSION" '
-                type == "object" and
-                all(to_entries[];
-                  (.key | test("^[a-z0-9][a-z0-9_-]*$")) and
-                  (.value | type == "string" and test("^[0-9]+\\.[0-9]+\\.[0-9]+(-(experimental|alpha|beta))?$"))
-                ) and
-                .[$worker] == $version
-              ' <<<"$STACK_VERSIONS" >/dev/null
               [[ "$COVERAGE" == "false" ]]
+              if [[ "$RESOLVE_STACK" == "true" ]]; then
+                [[ "$RELEASE_WORKER" == "harness" ]]
+                [[ "$RELEASE_VERSION" == "latest" ]]
+                jq -e 'type == "object" and length == 0' <<<"$STACK_VERSIONS" >/dev/null
+              else
+                [[ -n "$RELEASE_VERSION" ]]
+                jq -e --arg worker "$RELEASE_WORKER" --arg version "$RELEASE_VERSION" '
+                  type == "object" and
+                  all(to_entries[];
+                    (.key | test("^[a-z0-9][a-z0-9_-]*$")) and
+                    (.value | type == "string" and test("^[0-9]+\\.[0-9]+\\.[0-9]+(-(experimental|alpha|beta))?$"))
+                  ) and
+                  .[$worker] == $version
+                ' <<<"$STACK_VERSIONS" >/dev/null
+              fi
               ;;
             *)
               echo "::error::stack_mode must be source or registry"
@@ -510,6 +527,8 @@ jobs:
           HARNESS_E2E_RELEASE_RUN_ID: ${{ inputs.release_run_id }}
           HARNESS_E2E_SMOKE_RUN_ID: ${{ inputs.smoke_run_id }}
           HARNESS_E2E_STACK_VERSIONS: ${{ inputs.stack_versions }}
+          HARNESS_E2E_STACK_DIGEST: ${{ inputs.stack_digest }}
+          HARNESS_E2E_RESOLVE_STACK: ${{ inputs.resolve_stack }}
           HARNESS_E2E_BIN: ${{ github.workspace }}/target/runner/harness-e2e
           HARNESS_E2E_ARTIFACTS_DIR: ${{ github.workspace }}/target/harness-e2e
           HARNESS_E2E_SCENARIO: ${{ matrix.scenario }}
@@ -574,7 +593,11 @@ jobs:
         uses: actions/upload-artifact@v6
         with:
           name: harness-e2e-${{ matrix.subject.id }}-${{ matrix.scenario }}-results
-          path: target/harness-e2e/results/
+          path: |
+            target/harness-e2e/results/
+            target/harness-e2e/deployment.json
+            target/harness-e2e/cli-version.txt
+            target/harness-e2e/stack/
           retention-days: 14
           if-no-files-found: error
 
@@ -676,6 +699,9 @@ jobs:
           RELEASE_VERSION: ${{ inputs.release_version }}
           RELEASE_URL: ${{ inputs.release_url }}
           REGISTRY_TAG: ${{ inputs.registry_tag }}
+          STACK_MODE: ${{ inputs.stack_mode }}
+          STACK_VERSIONS: ${{ inputs.stack_versions }}
+          STACK_DIGEST: ${{ inputs.stack_digest }}
           JUDGE_MODEL: ${{ inputs.judge_model }}
           JUDGE_PROVIDER: ${{ inputs.judge_provider }}
           EXECUTION_RUN_ID: ${{ github.run_id }}
@@ -700,6 +726,9 @@ jobs:
             --release-version "$RELEASE_VERSION" \
             --release-url "$RELEASE_URL" \
             --registry-tag "$REGISTRY_TAG" \
+            --stack-mode "$STACK_MODE" \
+            --stack-versions "$STACK_VERSIONS" \
+            --stack-digest "$STACK_DIGEST" \
             --judge-model "$JUDGE_MODEL" \
             --judge-provider "$JUDGE_PROVIDER" \
             --execution-run-id "$EXECUTION_RUN_ID" \
diff --git a/.github/workflows/harness-e2e-daily.yml b/.github/workflows/harness-e2e-daily.yml
index 5baf0024d..08a3ed37f 100644
--- a/.github/workflows/harness-e2e-daily.yml
+++ b/.github/workflows/harness-e2e-daily.yml
@@ -1,5 +1,5 @@
 name: Harness E2E Daily
-run-name: Test · harness_source · ${{ inputs.operation_id || github.event_name }}
+run-name: Test · harness_registry · ${{ inputs.operation_id || github.event_name }}
 
 on:
   schedule:
@@ -79,13 +79,15 @@ jobs:
       runs: 3
       max_parallel: 2
       source_ref: ${{ needs.context.outputs.source_sha }}
+      stack_mode: registry
+      resolve_stack: true
       benchmark_lane: daily
       release_tag: daily/${{ needs.context.outputs.benchmark_day }}
-      release_worker: main
-      release_version: ${{ needs.context.outputs.benchmark_day }}
+      release_worker: harness
+      release_version: latest
       release_url: ${{ needs.context.outputs.source_url }}
-      registry_tag: daily
-      coverage: true
+      registry_tag: latest
+      coverage: false
       subjects: ${{ inputs.subjects || vars.HARNESS_E2E_SUBJECTS || '[{"id":"anthropic-sonnet","model":"claude-sonnet-4-6","provider":"anthropic"}]' }}
       judge_model: ${{ inputs.judge_model || vars.HARNESS_E2E_JUDGE_MODEL || 'claude-sonnet-4-6' }}
       judge_provider: ${{ inputs.judge_provider || vars.HARNESS_E2E_JUDGE_PROVIDER || 'anthropic' }}
@@ -94,17 +96,3 @@ jobs:
       openai_api_key: ${{ secrets.OPENAI_API_KEY }}
       zai_api_key: ${{ secrets.ZAI_API_KEY }}
       deepseek_api_key: ${{ secrets.DEEPSEEK_API_KEY }}
-
-  integration:
-    needs: context
-    if: github.ref == 'refs/heads/main'
-    permissions:
-      actions: read
-      contents: read
-    uses: ./.github/workflows/_harness-integration.yml
-    with:
-      coverage: true
-      # rust-cache saving is driven by the coverage input; the pinned engine
-      # cache key stays owned by cache-warm.
-      save-cache: false
-      source_ref: ${{ needs.context.outputs.source_sha }}
diff --git a/.github/workflows/harness-e2e-main.yml b/.github/workflows/harness-e2e-main.yml
index a19d40be3..7c3e41400 100644
--- a/.github/workflows/harness-e2e-main.yml
+++ b/.github/workflows/harness-e2e-main.yml
@@ -44,6 +44,7 @@ jobs:
       runs: 1
       max_parallel: 2
       benchmark_lane: main
+      coverage: true
       subjects: ${{ vars.HARNESS_E2E_SUBJECTS || '[{"id":"anthropic-sonnet","model":"claude-sonnet-4-6","provider":"anthropic"}]' }}
       judge_model: ${{ vars.HARNESS_E2E_JUDGE_MODEL || 'claude-sonnet-4-6' }}
       judge_provider: ${{ vars.HARNESS_E2E_JUDGE_PROVIDER || 'anthropic' }}
diff --git a/harness/tests/e2e/README.md b/harness/tests/e2e/README.md
index 6a60fd5a4..e5dd1c447 100644
--- a/harness/tests/e2e/README.md
+++ b/harness/tests/e2e/README.md
@@ -231,6 +231,13 @@ fresh stack, and repetitions run sequentially inside that job with unique table,
 session, and state namespaces. At most two matrix jobs make live-model calls
 concurrently.
 
+The daily lane uses the registry mode to measure the currently published live
+stack. It resolves `latest` once per matrix job, records the exact versions and
+SHA-256 digest of the resulting `iii.lock`, and treats Registry resolution
+failures as `infra_failed`; it does not fall back to a source build. The main
+lane keeps the source build and LLVM coverage so operational daily metrics stay
+separate from checkout regression coverage.
+
 The deployed lane is a separate workflow run dispatched by the release smoke
 workflow. The release publishes first, the smoke validates the published
 installation and exact released version, and only a successful smoke dispatches
diff --git a/harness/tests/e2e/run-deployed-ci.sh b/harness/tests/e2e/run-deployed-ci.sh
index a20860bbd..881fb63d8 100755
--- a/harness/tests/e2e/run-deployed-ci.sh
+++ b/harness/tests/e2e/run-deployed-ci.sh
@@ -20,10 +20,15 @@ install_url=${III_INSTALL_URL:-https://install.iii.dev/iii/main/install.sh}
 cli_channel=${III_CLI_CHANNEL:-latest}
 worker_tag=${III_WORKER_TAG:-latest}
 stack_versions=${HARNESS_E2E_STACK_VERSIONS:-'{}'}
+resolve_stack=${HARNESS_E2E_RESOLVE_STACK:-false}
+expected_stack_digest=${HARNESS_E2E_STACK_DIGEST:-}
 runs=${HARNESS_E2E_RUNS:-1}
 engine_port=49134
 wait_seconds=180
 add_timeout_seconds=600
+release_worker=$HARNESS_E2E_RELEASE_WORKER
+release_version=$HARNESS_E2E_RELEASE_VERSION
+lock_digest=
 
 if [[ -n "${III_CHANNEL:-}" ]]; then
   echo "III_CHANNEL was split into III_CLI_CHANNEL and III_WORKER_TAG" >&2
@@ -41,21 +46,47 @@ esac
   echo "III_WORKER_TAG must be a valid Registry tag" >&2
   exit 2
 }
-stack_versions=$(jq -c \
-  --arg worker "$HARNESS_E2E_RELEASE_WORKER" \
-  --arg version "$HARNESS_E2E_RELEASE_VERSION" '
-    if length == 0 then {($worker): $version} else . end
-  ' <<<"$stack_versions")
-jq -e --arg worker "$HARNESS_E2E_RELEASE_WORKER" --arg version "$HARNESS_E2E_RELEASE_VERSION" '
-  type == "object" and length > 0 and
-  all(to_entries[];
-    (.key | test("^[a-z0-9][a-z0-9_-]*$")) and
-    (.value | type == "string" and test("^[0-9]+\\.[0-9]+\\.[0-9]+(-(experimental|alpha|beta))?$"))
-  ) and .[$worker] == $version
-' <<<"$stack_versions" >/dev/null || {
-  echo "HARNESS_E2E_STACK_VERSIONS must contain the release worker and strict exact versions" >&2
+case "$resolve_stack" in
+  true | false) ;;
+  *)
+    echo "HARNESS_E2E_RESOLVE_STACK must be true or false" >&2
+    exit 2
+    ;;
+esac
+if [[ "$resolve_stack" == true ]]; then
+  [[ "$release_worker" == harness ]] || {
+    echo "dynamic Registry stack resolution requires release_worker=harness" >&2
+    exit 2
+  }
+  [[ "$release_version" == latest ]] || {
+    echo "dynamic Registry stack resolution requires release_version=latest" >&2
+    exit 2
+  }
+  jq -e 'type == "object" and length == 0' <<<"$stack_versions" >/dev/null || {
+    echo "dynamic Registry stack resolution requires empty stack_versions" >&2
+    exit 2
+  }
+else
+  stack_versions=$(jq -c \
+    --arg worker "$release_worker" \
+    --arg version "$release_version" '
+      if length == 0 then {($worker): $version} else . end
+    ' <<<"$stack_versions")
+  jq -e --arg worker "$release_worker" --arg version "$release_version" '
+    type == "object" and length > 0 and
+    all(to_entries[];
+      (.key | test("^[a-z0-9][a-z0-9_-]*$")) and
+      (.value | type == "string" and test("^[0-9]+\\.[0-9]+\\.[0-9]+(-(experimental|alpha|beta))?$"))
+    ) and .[$worker] == $version
+  ' <<<"$stack_versions" >/dev/null || {
+    echo "HARNESS_E2E_STACK_VERSIONS must contain the release worker and strict exact versions" >&2
+    exit 2
+  }
+fi
+if [[ -n "$expected_stack_digest" && ! "$expected_stack_digest" =~ ^[0-9a-f]{64}$ ]]; then
+  echo "HARNESS_E2E_STACK_DIGEST must be a SHA-256 digest" >&2
   exit 2
-}
+fi
 [[ -x "$e2e_bin" ]] || {
   echo "Harness E2E binary is not executable: $e2e_bin" >&2
   exit 2
@@ -97,20 +128,25 @@ die() {
 }
 
 write_deployment_result() {
-  local status=$1
+  local outcome=$1
+  local result_status=$outcome
+  if [[ "$outcome" != passed && "$failure_phase" == registry ]]; then
+    result_status=infra_failed
+  fi
   jq -n \
-    --arg status "$status" \
+    --arg status "$result_status" \
     --arg reason "$failure_reason" \
     --arg phase "$failure_phase" \
     --arg cli_version "$cli_version" \
     --arg cli_channel "$cli_channel" \
     --arg worker_tag "$worker_tag" \
-    --arg release_worker "$HARNESS_E2E_RELEASE_WORKER" \
-    --arg release_version "$HARNESS_E2E_RELEASE_VERSION" \
+    --arg release_worker "$release_worker" \
+    --arg release_version "$release_version" \
     --arg actual_release_version "$actual_release_version" \
     --arg release_tag "${HARNESS_E2E_RELEASE_TAG:-}" \
     --arg release_run_id "${HARNESS_E2E_RELEASE_RUN_ID:-}" \
     --arg smoke_run_id "${HARNESS_E2E_SMOKE_RUN_ID:-}" \
+    --arg lock_digest "$lock_digest" \
     --argjson stack_versions "$stack_versions" \
     --argjson elapsed_ms "$(((SECONDS - started_at_seconds) * 1000))" \
     '{
@@ -127,8 +163,10 @@ write_deployment_result() {
       release_run_id: $release_run_id,
       smoke_run_id: $smoke_run_id,
       stack_versions: $stack_versions,
+      stack_lock_digest: $lock_digest,
       elapsed_ms: $elapsed_ms
     }' >"$artifact_dir/deployment.json"
+  cp "$artifact_dir/deployment.json" "$artifact_dir/results/deployment.json"
 }
 
 snapshot_stack() {
@@ -311,9 +349,9 @@ printf 'workers: []\n' >"$project_dir/config.yaml"
 engine_pid=$!
 wait_for_engine
 
-# The candidate channel belongs only to the workers pinned in stack_versions.
 # Auxiliary E2E workers are not released as part of this operation and may not
-# expose the candidate tag at all, so keep them on their stable channel.
+# expose a candidate tag, so keep them on their stable channel unless the
+# entire live stack is being frozen below.
 support_worker_tag=latest
 workers=("database@$support_worker_tag" "fp@$support_worker_tag" "web@$support_worker_tag")
 declare -A providers=()
@@ -341,24 +379,44 @@ add_with_retry() {
   return 1
 }
 
-log "Installing stable E2E support stack: ${workers[*]}"
-add_with_retry worker-add "${workers[@]}"
-
-# Install the released worker first, then apply its exact candidate dependency
-# overrides. Resolving Harness necessarily selects the stable versions allowed
-# by its semver ranges; installing Harness last would overwrite the exact
-# dependency pins that Release Control supplied.
-while IFS=$'\t' read -r candidate_worker candidate_version; do
-  log "Installing exact stack candidate: ${candidate_worker}@${candidate_version}"
-  add_with_retry "candidate-${candidate_worker}" \
-    "${candidate_worker}@${candidate_version}" --force
-done < <(jq -r --arg release_worker "$HARNESS_E2E_RELEASE_WORKER" '
-  to_entries
-  | sort_by([if .key == $release_worker then 0 else 1 end, .key])[]
-  | [.key, .value]
-  | @tsv
-' <<<"$stack_versions")
+if [[ "$resolve_stack" == true ]]; then
+  failure_phase=registry
+  log "Resolving the live Registry stack from latest: harness@latest ${workers[*]}"
+  identity=$(add_with_retry live-stack "harness@latest" "${workers[@]}" >/dev/null && \
+    python3 "$repo_root/.github/scripts/registry_stack_identity.py" \
+      --lock "$project_dir/iii.lock")
+  stack_versions=$(jq -c '.stack_versions' <<<"$identity")
+  release_version=$(jq -er '.stack_versions.harness' <<<"$identity")
+  lock_digest=$(jq -er '.lock_digest' <<<"$identity")
+  if [[ -n "$expected_stack_digest" && "$lock_digest" != "$expected_stack_digest" ]]; then
+    die "resolved iii.lock digest $lock_digest does not match expected $expected_stack_digest"
+  fi
+else
+  failure_phase=registry
+  log "Installing stable E2E support stack: ${workers[*]}"
+  add_with_retry worker-add "${workers[@]}"
+
+  # Install the released worker first, then apply its exact candidate
+  # dependency overrides. Resolving Harness necessarily selects the stable
+  # versions allowed by its semver ranges; installing Harness last would
+  # overwrite the exact dependency pins that Release Control supplied.
+  while IFS=$'\t' read -r candidate_worker candidate_version; do
+    log "Installing exact stack candidate: ${candidate_worker}@${candidate_version}"
+    add_with_retry "candidate-${candidate_worker}" \
+      "${candidate_worker}@${candidate_version}" --force
+  done < <(jq -r --arg release_worker "$release_worker" '
+    to_entries
+    | sort_by([if .key == $release_worker then 0 else 1 end])[]
+    | [.key, .value]
+    | @tsv
+  ' <<<"$stack_versions")
+  lock_digest=$(sha256sum "$project_dir/iii.lock" | awk '{print $1}')
+  if [[ -n "$expected_stack_digest" && "$lock_digest" != "$expected_stack_digest" ]]; then
+    die "resolved iii.lock digest $lock_digest does not match expected $expected_stack_digest"
+  fi
+fi
 
+failure_phase=e2e
 wait_for_functions \
   harness::send harness::status worker::add database::query state::get \
   queue::define session::messages context::assemble router::models::get \
@@ -375,14 +433,15 @@ verify_args=(
   --manifest "$harness_root/iii.worker.yaml"
   --required harness
   --required database
-  --worker "$HARNESS_E2E_RELEASE_WORKER"
-  --version "$HARNESS_E2E_RELEASE_VERSION"
+  --worker "$release_worker"
+  --version "$release_version"
   --expected-versions-json "$stack_versions"
   --output "$stack_dir/lock-verification.json"
 )
 for worker in "${workers[@]}"; do
   verify_args+=(--required "${worker%@*}")
 done
+failure_phase=registry
 verification=$(python3 "$repo_root/.github/scripts/verify_registry_lock.py" "${verify_args[@]}")
 actual_release_version=$(jq -r '.actual_version' <<<"$verification")