feat(storage): index exact content references

2026-07-29 15:21:03 +03:00
parent 611853d3af
commit 32c5d0dda5
13 changed files with 1081 additions and 32 deletions
@@ -36,8 +36,17 @@ export interface ArtifactHealth {
    limitations: string[];
  };
  content_references: {
-    state: "not-indexed";
-    storage_model: "materialized-copies";
+    state: "not-indexed" | "indexed";
+    storage_model:
+      | "materialized-copies"
+      | "materialized-copies-with-reference-index";
+    result_id: string | null;
+    indexed_at_utc: string | null;
+    logical_reference_count: number;
+    canonical_content_count: number;
+    duplicate_reference_count: number;
+    exact_duplicate_bytes: number;
+    physical_reclamation_applied: false;
    storage_migration_authorized: false;
    next_gate: string;
  };
@@ -66,6 +75,15 @@ export async function fetchArtifactHealth(
    || !Array.isArray(document.inventory.scopes)
    || !isRecord(document.exact_audit)
    || !isRecord(document.content_references)
+    || !["not-indexed", "indexed"].includes(
+      String(document.content_references.state),
+    )
+    || ![
+      "materialized-copies",
+      "materialized-copies-with-reference-index",
+    ].includes(String(document.content_references.storage_model))
+    || document.content_references.physical_reclamation_applied !== false
+    || document.content_references.storage_migration_authorized !== false
  ) {
    throw new Error("Ответ здоровья данных не соответствует контракту.");
  }
@@ -135,6 +135,7 @@ export function ArtifactHealthWorkspace() {
    1,
  );
  const exactCurrent = health.exact_audit.state === "exact-current";
+  const referencesIndexed = health.content_references.state === "indexed";
  return (
    <div className="artifact-health-workspace">
      <section className="artifact-health-lead">
@@ -242,21 +243,55 @@ export function ArtifactHealthWorkspace() {
          <header>
            <div>
              <span className="section-eyebrow">CONTENT REFERENCES</span>
-              <h3>Материализованные копии</h3>
+              <h3>
+                {referencesIndexed
+                  ? "Индекс точного контента"
+                  : "Материализованные копии"}
+              </h3>
            </div>
-            <StatusBadge tone="warning">Не индексировано</StatusBadge>
+            <StatusBadge tone={referencesIndexed ? "success" : "warning"}>
+              {referencesIndexed ? "Индекс активен" : "Не индексировано"}
+            </StatusBadge>
          </header>
-          <p>
-            Content-addressed ссылки ещё не включены. Миграция хранения не
-            авторизована: сначала выбираются классы дублей с максимальным
-            измеренным эффектом.
-          </p>
-          <dl>
-            <div><dt>Режим</dt><dd>Наблюдение</dd></div>
-            <div><dt>Удаление</dt><dd>Запрещено</dd></div>
-            <div><dt>Автомиграция</dt><dd>Запрещена</dd></div>
-            <div><dt>Следующий gate</dt><dd>Content refs / chunking</dd></div>
-          </dl>
+          {referencesIndexed ? (
+            <>
+              <p>
+                Логические пути сведены к проверяемым SHA-256-ссылкам. Исходные
+                файлы не изменены, физическое освобождение места не выполнялось.
+              </p>
+              <dl>
+                <div>
+                  <dt>Логических путей</dt>
+                  <dd>{formatInteger(health.content_references.logical_reference_count)}</dd>
+                </div>
+                <div>
+                  <dt>Канонических объектов</dt>
+                  <dd>{formatInteger(health.content_references.canonical_content_count)}</dd>
+                </div>
+                <div>
+                  <dt>Ссылок на дубли</dt>
+                  <dd>{formatInteger(health.content_references.duplicate_reference_count)}</dd>
+                </div>
+                <div>
+                  <dt>Измеренный резерв</dt>
+                  <dd>{formatBytes(health.content_references.exact_duplicate_bytes)}</dd>
+                </div>
+              </dl>
+            </>
+          ) : (
+            <>
+              <p>
+                Content-addressed ссылки ещё не включены. Исходные файлы
+                остаются только наблюдаемыми, автоматическое удаление запрещено.
+              </p>
+              <dl>
+                <div><dt>Режим</dt><dd>Наблюдение</dd></div>
+                <div><dt>Удаление</dt><dd>Запрещено</dd></div>
+                <div><dt>Автомиграция</dt><dd>Запрещена</dd></div>
+                <div><dt>Индекс</dt><dd>Отсутствует</dd></div>
+              </dl>
+            </>
+          )}
        </GlassSurface>
      </section>
    </div>
@@ -12,7 +12,7 @@ Each gate produces evidence and an explicit GO, PAUSE or BLOCKED result.
 | Source-scoped quality | PAUSE — E40 visible engineering evaluation is 84.2466% presence, 84.2466% geometry association and 94.5205% freshness, with 11 high-severity failures, complete accounting and zero false-free claims. This is not a blind accuracy gate. |
 | Structural regressions | GO (bounded) — E42 passes predictor identity/order/chunk invariance and PointSlab order/SE(3) checks. Raw-producer and cross-route invariance remain unproved. |
 | Native pipeline telemetry | GO (durable Worker 006) — lifecycle events and JSONL/MQTT sink boundaries exist; the persistent worker writes to its existing D:-backed publish mount, host Telegraf tails with saved offset and publishes through the existing authenticated QoS 1 path. Real request `durable-native-acceptance-20260729-115418` produced native run `started → failed` after reaching the unavailable live-source boundary, and both documents were observed through broker → normalizer → Timescale query API with the same run/request identity. No broker credential or additional container entered the inference runtime. |
-| Evidence storage | MEASURED — E44 finds 525,471,092 logical bytes, 312,753,179 unique-content bytes and 1.680146× amplification across 14 E30–E40 roots. Exact-content references/deduplication precede any format migration. |
+| Evidence storage | GO (reference index only) — E50 converts the accepted E44 audit into 2,962 verified logical references over 1,029 canonical content objects, including 1,933 exact duplicate references and 212,717,913 measured duplicate bytes. Existing artifacts remain materialized and unchanged; physical reclamation and storage migration remain unauthorized. |
 | Future transfer | PREREGISTERED — E43 freezes same-K1/mount/calibration/firmware, required streams, connected-component split and independent label reveal. Capture and labels do not yet exist. |
 | E31 binding sensitivity | MEASURED — E45 closes accounting for 87/87 accepted correspondences and finds no material monotonic residual association with represented image radius, rig speed or pose age. It does not supply calibration-target truth or outer-fisheye coverage. |
 | Detector Truth Island | PREPARED — E46 freezes 32 references with no prelabels, predictions, scores or candidate identity in the review package. Two independent reviews and adjudication are still required. |
@@ -9,7 +9,8 @@ qualified; E28 complete worker replay accepted; E29 camera-first semantic and
 parallel geometry-only replay implemented; E30–E35 source-scoped qualification
 accepted; RAVNOVES00 reference-source product maturation active; E36 transfer
 preregistered and deferred by ADR 0030/0032; E41 methodology boundary, E42
-metamorphic checks and E44 amplification audit complete
+metamorphic checks, E44 amplification audit and E50 exact-content reference
+index complete
 Scope: passively received real-time K1 point/pose evidence, immutable replay and
 future live shadow processing
 Explicitly out of scope: K1 firmware modification, a new onboard exporter, new
@@ -751,8 +752,10 @@ prediction contain no reference/split/scoring material, reproduce E40
 predictions exactly and are evaluated only afterward by
 `e41-visible-evaluation-57aa6c8569e8339630406e3c88cd539df13917fb08c1af57f0baae5cdd1069d2`.
 E42 adds bounded structural invariants; E44 measures exact-content
-amplification; E43 freezes the later independent transfer protocol. ADR 0032
-contains the corrected decision boundary and current priority order.
+amplification; E50 indexes its 2,962 logical paths against 1,029 canonical
+SHA-256 content objects without rewriting or deleting historical evidence; E43
+freezes the later independent transfer protocol. ADR 0032 contains the
+corrected decision boundary and current priority order.

 E36 is the first generalization gate. A separate product decision follows:
 either keep the result as operator/shadow evidence, or start L5 occupied-space
@@ -142,6 +142,13 @@ across 14 admitted E30–E40 roots: `1.680146×` amplification and
 content-addressed referencing and exact deduplication, not an unmeasured
 format/database migration.

+E50 immutable reference index
+`e50-content-reference-index-c10c5ed7345a4f23d46ed0f132d54c378433aa4351e14375805cfe3e28878c81`
+reproduces that E44 identity and maps `2,962` logical paths to `1,029`
+canonical SHA-256 content objects. The remaining `1,933` paths are verified
+exact references. No existing artifact was rewritten, linked, moved or
+deleted; physical reclamation and storage migration remain false.
+
 E45 immutable diagnostic
 `e45-binding-sensitivity-b10fa117f110d5ac7942f73c62f1300c7d9ce833a187542b992e58fb24d1be72`
 accounts for all 87 accepted E31 correspondences and stratifies the existing
@@ -210,8 +217,9 @@ The current non-UI priority order is:
   regression checks;
 3. wire native pipeline telemetry into durable worker stages without inferring
   absent measurements;
-4. design exact-content references/deduplication from E44 before any storage
-   migration;
+4. preserve the accepted E50 exact-content reference index and require future
+   producers to adopt verified resolution before any separately authorized
+   physical reclamation;
 5. preserve E43 unchanged until the owner supplies the new capture.
 6. complete two independent E46 reviews and adjudication without exposing
   model prelabels or E47 predictions;
@@ -47,7 +47,9 @@ authority. It changes the claim that those artifacts are allowed to support.
   are admitted; no network transport starts implicitly.
 7. Current storage work starts with content-addressed references and exact-content
   deduplication. E44 measures `1.680146×` amplification across the admitted E30–E40
-   package/result roots. It does not authorize a storage-format or database migration.
+   package/result roots. E50 indexes every logical path against one verified canonical
+   SHA-256 locator without rewriting historical evidence. Neither result authorizes
+   physical reclamation, a storage-format change or a database migration.
 8. The later same-K1 transfer capture is preregistered before collection. E43 freezes
   minimum streams, route/control-bridge duration, connected scene/track/time
   partitioning, independent human review and truth reveal only after frozen
@@ -71,6 +73,8 @@ authority. It changes the claim that those artifacts are allowed to support.
  `e43-future-capture-protocol-28f091b9648daffce988d44c183e21f56d77988061630de934f8003fb13701d8`;
 - data-amplification audit:
  `e44-data-amplification-89791978894ec785009e5b76c9325de4d9e910237af57476c9331eb47a8ccca4`.
+- exact-content reference index:
+  `e50-content-reference-index-c10c5ed7345a4f23d46ed0f132d54c378433aa4351e14375805cfe3e28878c81`.

 ## Consequences

@@ -81,6 +85,9 @@ authority. It changes the claim that those artifacts are allowed to support.
  high-severity failures, complete accounting and zero false-free claims.
 - Further threshold or model tuning against the visible 146 items cannot close the
  blind gate. It may only create another explicitly visible engineering comparison.
+- Exact-content references remove duplicate namespace semantics for current evidence,
+  but the existing physical copies remain in place until a separate producer migration
+  is reviewed and authorized.
 - Work that does not require a new recording can proceed now: methodology enforcement,
  predictor/evaluator separation, invariants, telemetry, exact-content deduplication
  design and preregistration.
@@ -0,0 +1,78 @@
+# LAB E50 · exact-content reference index
+
+Date: 2026-07-29
+
+Status: accepted non-destructive reference index only
+
+Immutable result:
+`e50-content-reference-index-c10c5ed7345a4f23d46ed0f132d54c378433aa4351e14375805cfe3e28878c81`
+
+## Task
+
+E44 measured exact duplication across the 14 admitted E30–E40 result and
+package roots. E50 converts that measurement into a deterministic logical
+reference index without changing the existing evidence files.
+
+E50 does not:
+
+- delete, rewrite, move, hard-link or copy an existing artifact;
+- reclaim physical storage;
+- authorize a storage migration or new database;
+- weaken source digest, path or root-identity verification;
+- grant command, navigation or safety authority.
+
+## Immutable input
+
+The sole measurement input is E44 result
+`e44-data-amplification-89791978894ec785009e5b76c9325de4d9e910237af57476c9331eb47a8ccca4`.
+E50 requires the same 14 labelled roots, reproduces every E44 catalog digest
+and reproduces the exact E44 analysis SHA-256 before it writes an index.
+
+## Method
+
+Every logical file is addressed by its exact SHA-256 and byte length. One
+canonical locator is selected per digest by the frozen policy
+`prefer-non-package-then-lexical/v1`; every other logical path to the same
+content becomes an `exact-reference`.
+
+The persisted locators contain a root label, root basename and relative path,
+not an absolute host path. Resolution requires an explicit trusted-root map and
+fails closed when the root name, member path, regular-file status, byte length
+or SHA-256 no longer matches. Symlinks and path traversal are rejected.
+
+## Result
+
+| Measurement | Value |
+| --- | ---: |
+| Admitted roots | 14 |
+| Logical references | 2,962 |
+| Canonical content objects | 1,029 |
+| Exact duplicate references | 1,933 |
+| Duplicate content groups | 967 |
+| Logical bytes | 525,471,092 |
+| Addressable unique bytes | 312,753,179 |
+| Exact duplicate bytes | 212,717,913 |
+| Amplification | 1.680146× |
+
+The immutable index contains one canonical reference for every content object.
+All 2,962 logical paths remain addressable.
+
+## Product projection
+
+The existing `Здоровье данных` surface reads the latest valid E50 result and
+reports the active reference index, logical-path count, canonical-content
+count, duplicate-reference count and measured duplicate bytes. It does not
+present the measured reserve as reclaimed disk space.
+
+## Decision
+
+The exact-content reference layer is accepted. Namespace-level deduplication is
+implemented and independently resolvable, while all historical evidence
+remains byte-for-byte in place.
+
+Physical reclamation remains false. A future producer may adopt verified
+reference resolution only through a separately reviewed migration that
+preserves immutable evidence, rollback and consumer compatibility. Existing
+artifacts are not retroactively converted.
+
+Navigation, safety and command authority remain false.
@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+"""Build a non-destructive exact-content reference index from accepted E44 roots."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from k1link.compute.e50_content_reference_index import (
+    build_e50_content_reference_index,
+)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--e44-result", type=Path, required=True)
+    parser.add_argument(
+        "--artifact-root",
+        action="append",
+        required=True,
+        metavar="LABEL=PATH",
+    )
+    parser.add_argument("--output-root", type=Path, required=True)
+    args = parser.parse_args()
+    roots: dict[str, Path] = {}
+    for value in args.artifact_root:
+        label, separator, raw_path = value.partition("=")
+        if not separator or label in roots:
+            parser.error("--artifact-root must be a unique LABEL=PATH")
+        roots[label] = Path(raw_path)
+    result = build_e50_content_reference_index(
+        e44_result_root=args.e44_result,
+        artifact_roots=roots,
+        output_root=args.output_root,
+    )
+    print(
+        json.dumps(
+            {
+                "result_id": result.result_id,
+                "result_root": str(result.result_root),
+                "metrics": result.report["metrics"],
+                "decision": result.report["decision"],
+            },
+            ensure_ascii=False,
+            sort_keys=True,
+        )
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,571 @@
+"""Non-destructive exact-content references over an accepted E44 catalog."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import shutil
+import uuid
+from collections import defaultdict
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any, Final
+
+from k1link.compute.e44_data_amplification_audit import (
+    analyze_data_amplification,
+    read_e44_data_amplification_audit,
+)
+
+E50_RESULT_SCHEMA: Final = "missioncore.e50-content-reference-index/v1"
+E50_REPORT_SCHEMA: Final = "missioncore.e50-content-reference-report/v1"
+CONTENT_REFERENCE_SCHEMA: Final = "missioncore.content-reference/v1"
+E50_INDEX_NAME: Final = "content-references.jsonl"
+E50_REPORT_NAME: Final = "content-reference-report.json"
+E50_MANIFEST_NAME: Final = "manifest.json"
+CANONICAL_POLICY_ID: Final = "prefer-non-package-then-lexical/v1"
+
+_LABEL = re.compile(r"^[a-z0-9][a-z0-9._-]{1,63}$")
+_SHA256 = re.compile(r"^[a-f0-9]{64}$")
+_AUTHORITY: Final = {
+    "commands_enabled": False,
+    "navigation_or_safety_accepted": False,
+}
+
+
+class E50ContentReferenceIndexError(RuntimeError):
+    """An E50 source catalog, reference, or immutable result is invalid."""
+
+
+@dataclass(frozen=True, slots=True)
+class E50ContentReferenceIndex:
+    result_id: str
+    result_root: Path
+    manifest: dict[str, Any]
+    report: dict[str, Any]
+    references: tuple[dict[str, Any], ...]
+
+
+def build_e50_content_reference_index(
+    *,
+    e44_result_root: Path,
+    artifact_roots: dict[str, Path],
+    output_root: Path,
+) -> E50ContentReferenceIndex:
+    """Index exact content without copying, linking, rewriting, or deleting sources."""
+
+    e44 = read_e44_data_amplification_audit(e44_result_root)
+    expected_rows = e44.manifest["identity"]["artifact_roots"]
+    if not isinstance(expected_rows, list):
+        raise E50ContentReferenceIndexError("E50 E44 root identity is invalid")
+    expected_roots = {
+        str(row["label"]): {
+            "root_name": str(row["root_name"]),
+            "catalog_sha256": str(row["catalog_sha256"]),
+        }
+        for row in expected_rows
+        if isinstance(row, dict)
+    }
+    if (
+        len(expected_roots) != len(expected_rows)
+        or set(artifact_roots) != set(expected_roots)
+    ):
+        raise E50ContentReferenceIndexError("E50 artifact root coverage changed")
+
+    resolved_output = output_root.expanduser().resolve(strict=False)
+    resolved_roots: dict[str, Path] = {}
+    for label, source in artifact_roots.items():
+        if _LABEL.fullmatch(label) is None:
+            raise E50ContentReferenceIndexError("E50 artifact label is invalid")
+        source_path = source.expanduser().absolute()
+        if source_path.is_symlink():
+            raise E50ContentReferenceIndexError("E50 artifact root is a symlink")
+        root = source_path.resolve(strict=True)
+        expected = expected_roots[label]
+        if (
+            not root.is_dir()
+            or root.name != expected["root_name"]
+            or resolved_output == root
+            or resolved_output.is_relative_to(root)
+        ):
+            raise E50ContentReferenceIndexError("E50 artifact root is invalid")
+        resolved_roots[label] = root
+
+    files = _catalog_files(resolved_roots)
+    for label in sorted(resolved_roots):
+        rows = [row for row in files if row["root"] == label]
+        observed = hashlib.sha256(_canonical_json(rows)).hexdigest()
+        if observed != expected_roots[label]["catalog_sha256"]:
+            raise E50ContentReferenceIndexError(
+                f"E50 source catalog changed for {label}"
+            )
+    analysis = analyze_data_amplification(files)
+    analysis_sha256 = hashlib.sha256(_canonical_json(analysis)).hexdigest()
+    if analysis_sha256 != e44.manifest["identity"]["analysis_sha256"]:
+        raise E50ContentReferenceIndexError("E50 no longer reproduces the E44 audit")
+
+    references = _build_references(files, resolved_roots)
+    index_payload = b"".join(_canonical_json(row) + b"\n" for row in references)
+    index_sha256 = hashlib.sha256(index_payload).hexdigest()
+    root_identity = [
+        {
+            "label": label,
+            "root_name": expected_roots[label]["root_name"],
+            "catalog_sha256": expected_roots[label]["catalog_sha256"],
+        }
+        for label in sorted(expected_roots)
+    ]
+    identity = {
+        "schema_version": E50_RESULT_SCHEMA,
+        "source_e44_result_id": e44.result_id,
+        "source_e44_identity_sha256": e44.manifest["identity_sha256"],
+        "artifact_roots": root_identity,
+        "canonical_policy_id": CANONICAL_POLICY_ID,
+        "reference_schema": CONTENT_REFERENCE_SCHEMA,
+        "reference_count": len(references),
+        "reference_index_sha256": index_sha256,
+        "producer_sha256": _sha256(Path(__file__).resolve(strict=True)),
+        "authority": _AUTHORITY,
+    }
+    identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest()
+    result_id = f"e50-content-reference-index-{identity_sha256}"
+    destination = resolved_output / result_id
+    if destination.exists():
+        return read_e50_content_reference_index(destination)
+
+    unique_content = {str(row["content_id"]) for row in references}
+    canonical_references = sum(
+        row["relation"] == "canonical" for row in references
+    )
+    report = {
+        "schema_version": E50_REPORT_SCHEMA,
+        "result_id": result_id,
+        "identity_sha256": identity_sha256,
+        "status": "completed-non-destructive-content-reference-index",
+        "source_e44_result_id": e44.result_id,
+        "metrics": {
+            "root_count": len(resolved_roots),
+            "logical_reference_count": len(references),
+            "canonical_content_count": len(unique_content),
+            "canonical_reference_count": canonical_references,
+            "duplicate_reference_count": len(references) - len(unique_content),
+            "duplicate_content_groups": analysis["duplicate_content_groups"],
+            "logical_bytes": analysis["logical_bytes"],
+            "addressable_unique_bytes": analysis["unique_content_bytes"],
+            "exact_duplicate_bytes": analysis["duplicate_bytes"],
+            "amplification_ratio": analysis["amplification_ratio"],
+        },
+        "decision": {
+            "content_references_indexed": True,
+            "existing_artifacts_rewritten": False,
+            "existing_artifacts_deleted": False,
+            "physical_reclamation_applied": False,
+            "storage_migration_authorized": False,
+            "resolver_requires_explicit_trusted_roots": True,
+        },
+        "limitations": [
+            "the index removes duplicate namespace semantics, not existing physical files",
+            "physical reclamation requires a separately authorized producer migration",
+            "references are valid only while exact root name, path, size and SHA-256 match",
+        ],
+        "authority": _AUTHORITY,
+    }
+
+    destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+    staging = destination.parent / f".{result_id}.{uuid.uuid4().hex}.tmp"
+    staging.mkdir(mode=0o700, exist_ok=False)
+    try:
+        _write_bytes(staging / E50_INDEX_NAME, index_payload)
+        _write_json(staging / E50_REPORT_NAME, report)
+        manifest = {
+            "schema_version": E50_RESULT_SCHEMA,
+            "result_id": result_id,
+            "identity_sha256": identity_sha256,
+            "identity": identity,
+            "created_at_utc": _utc_now(),
+            "acceptance_state": "accepted-reference-index-only",
+            "artifacts": [
+                _artifact(staging / E50_INDEX_NAME, "content-reference-index"),
+                _artifact(staging / E50_REPORT_NAME, "content-reference-report"),
+            ],
+            "authority": _AUTHORITY,
+        }
+        _write_json(staging / E50_MANIFEST_NAME, manifest)
+        os.replace(staging, destination)
+    except BaseException:
+        shutil.rmtree(staging, ignore_errors=True)
+        raise
+    return read_e50_content_reference_index(destination)
+
+
+def read_e50_content_reference_index(root: Path) -> E50ContentReferenceIndex:
+    """Read and fully validate one immutable E50 result."""
+
+    source = root.expanduser().absolute()
+    if source.is_symlink():
+        raise E50ContentReferenceIndexError("E50 result root is a symlink")
+    resolved = source.resolve(strict=True)
+    manifest = _read_json(resolved / E50_MANIFEST_NAME)
+    identity = _object(manifest.get("identity"), "E50 identity")
+    identity_sha256 = manifest.get("identity_sha256")
+    if (
+        manifest.get("schema_version") != E50_RESULT_SCHEMA
+        or not isinstance(identity_sha256, str)
+        or _SHA256.fullmatch(identity_sha256) is None
+        or hashlib.sha256(_canonical_json(identity)).hexdigest() != identity_sha256
+        or manifest.get("result_id")
+        != f"e50-content-reference-index-{identity_sha256}"
+        or resolved.name != manifest.get("result_id")
+        or manifest.get("acceptance_state") != "accepted-reference-index-only"
+        or manifest.get("authority") != _AUTHORITY
+        or identity.get("canonical_policy_id") != CANONICAL_POLICY_ID
+        or identity.get("reference_schema") != CONTENT_REFERENCE_SCHEMA
+        or identity.get("authority") != _AUTHORITY
+    ):
+        raise E50ContentReferenceIndexError("E50 result identity is invalid")
+    artifacts = manifest.get("artifacts")
+    expected_artifacts = {
+        E50_INDEX_NAME: "content-reference-index",
+        E50_REPORT_NAME: "content-reference-report",
+    }
+    if not isinstance(artifacts, list) or len(artifacts) != 2:
+        raise E50ContentReferenceIndexError("E50 artifact set is invalid")
+    observed_artifacts: set[str] = set()
+    for row in artifacts:
+        if not isinstance(row, dict):
+            raise E50ContentReferenceIndexError("E50 artifact is invalid")
+        relative = row.get("path")
+        path = resolved / str(relative)
+        if (
+            not isinstance(relative, str)
+            or relative not in expected_artifacts
+            or relative in observed_artifacts
+            or row.get("role") != expected_artifacts[relative]
+            or not path.is_file()
+            or path.is_symlink()
+            or row.get("byte_length") != path.stat().st_size
+            or row.get("sha256") != _sha256(path)
+        ):
+            raise E50ContentReferenceIndexError("E50 artifact content changed")
+        observed_artifacts.add(relative)
+    if observed_artifacts != set(expected_artifacts):
+        raise E50ContentReferenceIndexError("E50 artifact coverage changed")
+
+    index_payload = (resolved / E50_INDEX_NAME).read_bytes()
+    if hashlib.sha256(index_payload).hexdigest() != identity.get(
+        "reference_index_sha256"
+    ):
+        raise E50ContentReferenceIndexError("E50 reference index digest changed")
+    references = tuple(
+        _read_reference_line(line)
+        for line in index_payload.splitlines()
+        if line
+    )
+    _validate_reference_set(references)
+    if identity.get("reference_count") != len(references):
+        raise E50ContentReferenceIndexError("E50 reference count changed")
+
+    report = _read_json(resolved / E50_REPORT_NAME)
+    metrics = _object(report.get("metrics"), "E50 metrics")
+    unique_content = {str(row["content_id"]) for row in references}
+    if (
+        report.get("schema_version") != E50_REPORT_SCHEMA
+        or report.get("result_id") != resolved.name
+        or report.get("identity_sha256") != identity_sha256
+        or report.get("source_e44_result_id")
+        != identity.get("source_e44_result_id")
+        or metrics.get("logical_reference_count") != len(references)
+        or metrics.get("canonical_content_count") != len(unique_content)
+        or metrics.get("canonical_reference_count") != len(unique_content)
+        or report.get("decision", {}).get("content_references_indexed") is not True
+        or report.get("decision", {}).get("existing_artifacts_rewritten") is not False
+        or report.get("decision", {}).get("existing_artifacts_deleted") is not False
+        or report.get("decision", {}).get("physical_reclamation_applied") is not False
+        or report.get("decision", {}).get("storage_migration_authorized") is not False
+        or report.get("authority") != _AUTHORITY
+    ):
+        raise E50ContentReferenceIndexError("E50 report is invalid")
+    return E50ContentReferenceIndex(
+        result_id=resolved.name,
+        result_root=resolved,
+        manifest=manifest,
+        report=report,
+        references=references,
+    )
+
+
+def resolve_e50_content_reference(
+    index_root: Path,
+    artifact_roots: dict[str, Path],
+    *,
+    root_label: str,
+    relative_path: str,
+    prefer_canonical: bool = True,
+) -> Path:
+    """Resolve one indexed logical path through explicit trusted roots."""
+
+    index = read_e50_content_reference_index(index_root)
+    selected = next(
+        (
+            row
+            for row in index.references
+            if row["logical"]["root_label"] == root_label
+            and row["logical"]["path"] == relative_path
+        ),
+        None,
+    )
+    if selected is None:
+        raise E50ContentReferenceIndexError("E50 logical reference was not indexed")
+    locator = selected["canonical"] if prefer_canonical else selected["logical"]
+    label = str(locator["root_label"])
+    root = artifact_roots.get(label)
+    if root is None:
+        raise E50ContentReferenceIndexError("E50 trusted root is missing")
+    root_path = root.expanduser().absolute()
+    if root_path.is_symlink():
+        raise E50ContentReferenceIndexError("E50 trusted root is a symlink")
+    resolved_root = root_path.resolve(strict=True)
+    if (
+        not resolved_root.is_dir()
+        or resolved_root.name != locator["root_name"]
+    ):
+        raise E50ContentReferenceIndexError("E50 trusted root identity changed")
+    path = _regular_member(resolved_root, str(locator["path"]))
+    if (
+        path.stat().st_size != selected["byte_length"]
+        or _sha256(path) != selected["sha256"]
+    ):
+        raise E50ContentReferenceIndexError("E50 referenced content changed")
+    return path
+
+
+def _catalog_files(roots: dict[str, Path]) -> list[dict[str, Any]]:
+    files: list[dict[str, Any]] = []
+    for label, root in sorted(roots.items()):
+        for path in sorted(root.rglob("*")):
+            if path.is_symlink():
+                raise E50ContentReferenceIndexError("E50 input contains a symlink")
+            if not path.is_file():
+                continue
+            files.append(
+                {
+                    "root": label,
+                    "path": path.relative_to(root).as_posix(),
+                    "byte_length": path.stat().st_size,
+                    "sha256": _sha256(path),
+                    "kind": _artifact_kind(path),
+                }
+            )
+    if not files:
+        raise E50ContentReferenceIndexError("E50 artifact roots are empty")
+    return files
+
+
+def _build_references(
+    files: list[dict[str, Any]],
+    roots: dict[str, Path],
+) -> tuple[dict[str, Any], ...]:
+    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for row in files:
+        grouped[str(row["sha256"])].append(row)
+    canonical_by_digest = {
+        digest: min(rows, key=_canonical_sort_key)
+        for digest, rows in grouped.items()
+    }
+    references = []
+    for row in sorted(files, key=lambda item: (item["root"], item["path"])):
+        digest = str(row["sha256"])
+        canonical = canonical_by_digest[digest]
+        logical = _locator(row, roots)
+        canonical_locator = _locator(canonical, roots)
+        references.append(
+            {
+                "schema_version": CONTENT_REFERENCE_SCHEMA,
+                "content_id": f"sha256:{digest}",
+                "algorithm": "sha256",
+                "sha256": digest,
+                "byte_length": int(row["byte_length"]),
+                "kind": str(row["kind"]),
+                "logical": logical,
+                "canonical": canonical_locator,
+                "relation": (
+                    "canonical" if logical == canonical_locator else "exact-reference"
+                ),
+            }
+        )
+    return tuple(references)
+
+
+def _canonical_sort_key(row: dict[str, Any]) -> tuple[int, str, str]:
+    label = str(row["root"])
+    package_penalty = int(label.endswith("-package"))
+    return package_penalty, label, str(row["path"])
+
+
+def _locator(row: dict[str, Any], roots: dict[str, Path]) -> dict[str, str]:
+    label = str(row["root"])
+    return {
+        "root_label": label,
+        "root_name": roots[label].name,
+        "path": str(row["path"]),
+    }
+
+
+def _read_reference_line(line: bytes) -> dict[str, Any]:
+    try:
+        value = json.loads(line)
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise E50ContentReferenceIndexError("E50 reference row is invalid") from exc
+    if not isinstance(value, dict):
+        raise E50ContentReferenceIndexError("E50 reference row must be an object")
+    return value
+
+
+def _validate_reference_set(references: tuple[dict[str, Any], ...]) -> None:
+    logical_keys: set[tuple[str, str]] = set()
+    canonical_by_content: dict[str, dict[str, str]] = {}
+    canonical_counts: defaultdict[str, int] = defaultdict(int)
+    for row in references:
+        logical = _object(row.get("logical"), "E50 logical locator")
+        canonical = _object(row.get("canonical"), "E50 canonical locator")
+        digest = row.get("sha256")
+        content_id = row.get("content_id")
+        byte_length = row.get("byte_length")
+        relation = row.get("relation")
+        if (
+            row.get("schema_version") != CONTENT_REFERENCE_SCHEMA
+            or row.get("algorithm") != "sha256"
+            or not isinstance(digest, str)
+            or _SHA256.fullmatch(digest) is None
+            or content_id != f"sha256:{digest}"
+            or not isinstance(byte_length, int)
+            or isinstance(byte_length, bool)
+            or byte_length < 0
+            or not isinstance(row.get("kind"), str)
+            or relation not in {"canonical", "exact-reference"}
+        ):
+            raise E50ContentReferenceIndexError("E50 content reference is invalid")
+        for locator in (logical, canonical):
+            label = locator.get("root_label")
+            root_name = locator.get("root_name")
+            relative = locator.get("path")
+            if (
+                not isinstance(label, str)
+                or _LABEL.fullmatch(label) is None
+                or not isinstance(root_name, str)
+                or not root_name
+                or not isinstance(relative, str)
+                or not relative
+                or Path(relative).is_absolute()
+                or ".." in Path(relative).parts
+            ):
+                raise E50ContentReferenceIndexError("E50 locator is invalid")
+        logical_key = (str(logical["root_label"]), str(logical["path"]))
+        if logical_key in logical_keys:
+            raise E50ContentReferenceIndexError("E50 logical reference is duplicated")
+        logical_keys.add(logical_key)
+        previous = canonical_by_content.setdefault(str(content_id), canonical)
+        if previous != canonical:
+            raise E50ContentReferenceIndexError("E50 canonical reference changed")
+        if relation == "canonical":
+            if logical != canonical:
+                raise E50ContentReferenceIndexError("E50 canonical relation is invalid")
+            canonical_counts[str(content_id)] += 1
+        elif logical == canonical:
+            raise E50ContentReferenceIndexError("E50 exact reference is canonical")
+    if not references or any(count != 1 for count in canonical_counts.values()):
+        raise E50ContentReferenceIndexError("E50 canonical coverage is invalid")
+    if set(canonical_counts) != set(canonical_by_content):
+        raise E50ContentReferenceIndexError("E50 content coverage is invalid")
+
+
+def _regular_member(root: Path, relative: str) -> Path:
+    path = Path(relative)
+    if path.is_absolute() or not path.parts or ".." in path.parts:
+        raise E50ContentReferenceIndexError("E50 reference path is invalid")
+    candidate = root
+    for part in path.parts:
+        candidate = candidate / part
+        if candidate.is_symlink():
+            raise E50ContentReferenceIndexError("E50 reference crosses a symlink")
+    resolved = candidate.resolve(strict=True)
+    if not resolved.is_file() or not resolved.is_relative_to(root):
+        raise E50ContentReferenceIndexError("E50 reference is not a regular member")
+    return resolved
+
+
+def _artifact_kind(path: Path) -> str:
+    suffix = path.suffix.lower()
+    if suffix in {".jpg", ".jpeg", ".png", ".webp"}:
+        return "camera-image"
+    if suffix in {".npy", ".npz", ".las", ".laz", ".pcd", ".ply"}:
+        return "point-or-array"
+    if suffix in {".mp4", ".mkv", ".mov"}:
+        return "video"
+    if suffix == ".rrd":
+        return "rerun"
+    if suffix in {".json", ".jsonl", ".md", ".txt", ".yaml", ".yml"}:
+        return "metadata-or-report"
+    if suffix in {".py", ".ps1", ".sh"}:
+        return "runtime-source"
+    return "other"
+
+
+def _artifact(path: Path, role: str) -> dict[str, Any]:
+    return {
+        "path": path.name,
+        "role": role,
+        "byte_length": path.stat().st_size,
+        "sha256": _sha256(path),
+    }
+
+
+def _object(value: object, label: str) -> dict[str, Any]:
+    if not isinstance(value, dict):
+        raise E50ContentReferenceIndexError(f"{label} must be an object")
+    return value
+
+
+def _read_json(path: Path) -> dict[str, Any]:
+    value = json.loads(path.read_text(encoding="utf-8-sig"))
+    if not isinstance(value, dict):
+        raise E50ContentReferenceIndexError(f"JSON object expected: {path.name}")
+    return value
+
+
+def _write_json(path: Path, value: object) -> None:
+    with path.open("x", encoding="utf-8", newline="\n") as stream:
+        json.dump(value, stream, indent=2, sort_keys=True)
+        stream.write("\n")
+        stream.flush()
+        os.fsync(stream.fileno())
+
+
+def _write_bytes(path: Path, value: bytes) -> None:
+    with path.open("xb") as stream:
+        stream.write(value)
+        stream.flush()
+        os.fsync(stream.fileno())
+
+
+def _canonical_json(value: object) -> bytes:
+    return json.dumps(
+        value,
+        sort_keys=True,
+        separators=(",", ":"),
+        allow_nan=False,
+    ).encode()
+
+
+def _sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as stream:
+        while chunk := stream.read(1024 * 1024):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _utc_now() -> str:
+    return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")
@@ -669,6 +669,13 @@ app.include_router(
            / "e44"
            / "results"
        ),
+        e50_results_root_provider=lambda: (
+            REPOSITORY_ROOT
+            / ".runtime"
+            / "compute-experiments"
+            / "e50"
+            / "results"
+        ),
    )
 )
 app.include_router(
@@ -18,6 +18,10 @@ from k1link.compute.e44_data_amplification_audit import (
    E44DataAmplificationAuditError,
    read_e44_data_amplification_audit,
 )
+from k1link.compute.e50_content_reference_index import (
+    E50ContentReferenceIndexError,
+    read_e50_content_reference_index,
+)

 ARTIFACT_HEALTH_SCHEMA: Final = "missioncore.artifact-health/v1"
 DEFAULT_CACHE_SECONDS: Final = 30.0
@@ -77,8 +81,18 @@ class ExactAmplificationAudit(BaseModel):
 class ContentReferenceStatus(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)

-    state: Literal["not-indexed"] = "not-indexed"
-    storage_model: Literal["materialized-copies"] = "materialized-copies"
+    state: Literal["not-indexed", "indexed"]
+    storage_model: Literal[
+        "materialized-copies",
+        "materialized-copies-with-reference-index",
+    ]
+    result_id: str | None = None
+    indexed_at_utc: str | None = None
+    logical_reference_count: int = Field(default=0, ge=0)
+    canonical_content_count: int = Field(default=0, ge=0)
+    duplicate_reference_count: int = Field(default=0, ge=0)
+    exact_duplicate_bytes: int = Field(default=0, ge=0)
+    physical_reclamation_applied: Literal[False] = False
    storage_migration_authorized: Literal[False] = False
    next_gate: str

@@ -102,10 +116,12 @@ class ArtifactHealthService:
        runtime_root_provider: RootProvider,
        e44_results_root_provider: RootProvider,
        *,
+        e50_results_root_provider: RootProvider | None = None,
        cache_seconds: float = DEFAULT_CACHE_SECONDS,
    ) -> None:
        self._runtime_root_provider = runtime_root_provider
        self._e44_results_root_provider = e44_results_root_provider
+        self._e50_results_root_provider = e50_results_root_provider
        self._cache_seconds = max(cache_seconds, 0)
        self._lock = threading.Lock()
        self._cached_at = 0.0
@@ -124,6 +140,10 @@ class ArtifactHealthService:
            expected_roots = audit_source[2] if audit_source is not None else {}
            inventory, audited_roots = self._scan_runtime(expected_roots, previous)
            exact_audit = self._audit_document(audit_source, audited_roots)
+            content_references = self._content_reference_document(
+                self._read_latest_reference_index(),
+                exact_audit,
+            )
            overall_state: Literal["live", "attention", "unavailable"]
            if inventory.state == "unavailable":
                overall_state = "unavailable"
@@ -140,12 +160,7 @@ class ArtifactHealthService:
                overall_state=overall_state,
                inventory=inventory,
                exact_audit=exact_audit,
-                content_references=ContentReferenceStatus(
-                    next_gate=(
-                        "Select content-addressed references or chunking only from "
-                        "measured dominant duplicate classes."
-                    )
-                ),
+                content_references=content_references,
            )
            self._cached = document
            self._cached_at = now
@@ -190,6 +205,37 @@ class ArtifactHealthService:
                continue
        return None

+    def _read_latest_reference_index(
+        self,
+    ) -> tuple[dict[str, Any], dict[str, Any]] | None:
+        if self._e50_results_root_provider is None:
+            return None
+        results_root = self._e50_results_root_provider()
+        if not results_root.is_dir():
+            return None
+        candidates = sorted(
+            (
+                candidate
+                for candidate in results_root.glob("e50-content-reference-index-*")
+                if candidate.is_dir()
+            ),
+            key=lambda candidate: candidate.stat().st_mtime_ns,
+            reverse=True,
+        )
+        for candidate in candidates:
+            try:
+                index = read_e50_content_reference_index(candidate)
+                return index.manifest, index.report
+            except (
+                E50ContentReferenceIndexError,
+                OSError,
+                TypeError,
+                ValueError,
+                json.JSONDecodeError,
+            ):
+                continue
+        return None
+
    def _scan_runtime(
        self,
        expected_roots: dict[str, dict[str, int]],
@@ -359,17 +405,59 @@ class ArtifactHealthService:
            limitations=tuple(str(item) for item in report.get("limitations", ())),
        )

+    def _content_reference_document(
+        self,
+        source: tuple[dict[str, Any], dict[str, Any]] | None,
+        exact_audit: ExactAmplificationAudit,
+    ) -> ContentReferenceStatus:
+        if source is None:
+            return ContentReferenceStatus(
+                state="not-indexed",
+                storage_model="materialized-copies",
+                next_gate=(
+                    "Build an immutable exact-content reference index from the "
+                    "accepted amplification audit."
+                ),
+            )
+        manifest, report = source
+        if report.get("source_e44_result_id") != exact_audit.result_id:
+            return ContentReferenceStatus(
+                state="not-indexed",
+                storage_model="materialized-copies",
+                next_gate=(
+                    "Rebuild the content reference index from the current exact "
+                    "amplification audit."
+                ),
+            )
+        metrics = report["metrics"]
+        return ContentReferenceStatus(
+            state="indexed",
+            storage_model="materialized-copies-with-reference-index",
+            result_id=str(report["result_id"]),
+            indexed_at_utc=str(manifest["created_at_utc"]),
+            logical_reference_count=int(metrics["logical_reference_count"]),
+            canonical_content_count=int(metrics["canonical_content_count"]),
+            duplicate_reference_count=int(metrics["duplicate_reference_count"]),
+            exact_duplicate_bytes=int(metrics["exact_duplicate_bytes"]),
+            next_gate=(
+                "Adopt verified reference resolution in future producers before "
+                "any separately authorized physical reclamation."
+            ),
+        )
+

 def build_artifact_health_router(
    *,
    runtime_root_provider: RootProvider,
    e44_results_root_provider: RootProvider,
+    e50_results_root_provider: RootProvider | None = None,
    cache_seconds: float = DEFAULT_CACHE_SECONDS,
 ) -> APIRouter:
    router = APIRouter()
    service = ArtifactHealthService(
        runtime_root_provider,
        e44_results_root_provider,
+        e50_results_root_provider=e50_results_root_provider,
        cache_seconds=cache_seconds,
    )

@@ -6,6 +6,9 @@ from pathlib import Path
 from k1link.compute.e44_data_amplification_audit import (
    build_e44_data_amplification_audit,
 )
+from k1link.compute.e50_content_reference_index import (
+    build_e50_content_reference_index,
+)
 from k1link.web.artifact_health_api import ArtifactHealthService


@@ -23,14 +26,21 @@ def test_artifact_health_reports_live_inventory_and_exact_audit(tmp_path: Path)
    _write(second / "shared.bin", b"shared")
    _write(second / "only-b.bin", b"bb")
    e44_results = runtime / "compute-experiments" / "e44" / "results"
-    build_e44_data_amplification_audit(
+    e44 = build_e44_data_amplification_audit(
        artifact_roots={"e30": first, "e40": second},
        output_root=e44_results,
    )
+    e50_results = runtime / "compute-experiments" / "e50" / "results"
+    e50 = build_e50_content_reference_index(
+        e44_result_root=e44.result_root,
+        artifact_roots={"e30": first, "e40": second},
+        output_root=e50_results,
+    )

    service = ArtifactHealthService(
        lambda: runtime,
        lambda: e44_results,
+        e50_results_root_provider=lambda: e50_results,
        cache_seconds=0,
    )
    document = service.snapshot()
@@ -40,7 +50,13 @@ def test_artifact_health_reports_live_inventory_and_exact_audit(tmp_path: Path)
    assert document.inventory.logical_bytes > 0
    assert document.exact_audit.state == "exact-current"
    assert document.exact_audit.duplicate_bytes == len(b"shared")
-    assert document.content_references.state == "not-indexed"
+    assert document.content_references.state == "indexed"
+    assert document.content_references.result_id == e50.result_id
+    assert document.content_references.logical_reference_count == 4
+    assert document.content_references.canonical_content_count == 3
+    assert document.content_references.duplicate_reference_count == 1
+    assert document.content_references.exact_duplicate_bytes == len(b"shared")
+    assert document.content_references.physical_reclamation_applied is False
    assert document.content_references.storage_migration_authorized is False


@@ -81,3 +97,4 @@ def test_artifact_health_reports_unavailable_runtime(tmp_path: Path) -> None:
    assert document.overall_state == "unavailable"
    assert document.inventory.state == "unavailable"
    assert document.exact_audit.state == "unavailable"
+    assert document.content_references.state == "not-indexed"
@@ -0,0 +1,164 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from k1link.compute.e44_data_amplification_audit import (
+    build_e44_data_amplification_audit,
+)
+from k1link.compute.e50_content_reference_index import (
+    E50ContentReferenceIndexError,
+    build_e50_content_reference_index,
+    read_e50_content_reference_index,
+    resolve_e50_content_reference,
+)
+
+
+def _write(path: Path, payload: bytes) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(payload)
+
+
+def test_e50_indexes_exact_references_without_rewriting_sources(
+    tmp_path: Path,
+) -> None:
+    result_root = tmp_path / "e30-result"
+    package_root = tmp_path / "e40-package"
+    shared = b"shared-evidence"
+    _write(result_root / "frames/shared.bin", shared)
+    _write(result_root / "unique.bin", b"result-only")
+    _write(package_root / "input/shared.bin", shared)
+    before = {
+        path: (path.stat().st_size, path.stat().st_mtime_ns, path.read_bytes())
+        for path in (result_root / "frames/shared.bin", package_root / "input/shared.bin")
+    }
+    roots = {"e30": result_root, "e40-package": package_root}
+    e44 = build_e44_data_amplification_audit(
+        artifact_roots=roots,
+        output_root=tmp_path / "e44",
+    )
+
+    result = build_e50_content_reference_index(
+        e44_result_root=e44.result_root,
+        artifact_roots=roots,
+        output_root=tmp_path / "e50",
+    )
+    repeated = read_e50_content_reference_index(result.result_root)
+
+    assert repeated.result_id == result.result_id
+    assert result.report["metrics"]["logical_reference_count"] == 3
+    assert result.report["metrics"]["canonical_content_count"] == 2
+    assert result.report["metrics"]["duplicate_reference_count"] == 1
+    assert result.report["metrics"]["exact_duplicate_bytes"] == len(shared)
+    assert result.report["decision"] == {
+        "content_references_indexed": True,
+        "existing_artifacts_deleted": False,
+        "existing_artifacts_rewritten": False,
+        "physical_reclamation_applied": False,
+        "resolver_requires_explicit_trusted_roots": True,
+        "storage_migration_authorized": False,
+    }
+    assert before == {
+        path: (path.stat().st_size, path.stat().st_mtime_ns, path.read_bytes())
+        for path in before
+    }
+
+
+def test_e50_resolves_duplicate_to_verified_non_package_canonical(
+    tmp_path: Path,
+) -> None:
+    result_root = tmp_path / "e30-result"
+    package_root = tmp_path / "e40-package"
+    _write(result_root / "shared.bin", b"same")
+    _write(package_root / "shared.bin", b"same")
+    roots = {"e30": result_root, "e40-package": package_root}
+    e44 = build_e44_data_amplification_audit(
+        artifact_roots=roots,
+        output_root=tmp_path / "e44",
+    )
+    result = build_e50_content_reference_index(
+        e44_result_root=e44.result_root,
+        artifact_roots=roots,
+        output_root=tmp_path / "e50",
+    )
+
+    canonical = resolve_e50_content_reference(
+        result.result_root,
+        roots,
+        root_label="e40-package",
+        relative_path="shared.bin",
+    )
+    logical = resolve_e50_content_reference(
+        result.result_root,
+        roots,
+        root_label="e40-package",
+        relative_path="shared.bin",
+        prefer_canonical=False,
+    )
+
+    assert canonical == (result_root / "shared.bin").resolve()
+    assert logical == (package_root / "shared.bin").resolve()
+
+
+def test_e50_resolver_fails_closed_after_referenced_content_changes(
+    tmp_path: Path,
+) -> None:
+    first = tmp_path / "first-root"
+    second = tmp_path / "second-root"
+    _write(first / "shared.bin", b"same")
+    _write(second / "shared.bin", b"same")
+    roots = {"first": first, "second": second}
+    e44 = build_e44_data_amplification_audit(
+        artifact_roots=roots,
+        output_root=tmp_path / "e44",
+    )
+    result = build_e50_content_reference_index(
+        e44_result_root=e44.result_root,
+        artifact_roots=roots,
+        output_root=tmp_path / "e50",
+    )
+    _write(second / "shared.bin", b"changed")
+
+    with pytest.raises(E50ContentReferenceIndexError, match="changed"):
+        resolve_e50_content_reference(
+            result.result_root,
+            roots,
+            root_label="second",
+            relative_path="shared.bin",
+            prefer_canonical=False,
+        )
+
+
+def test_e50_rejects_symlinked_source_and_trusted_roots(tmp_path: Path) -> None:
+    first = tmp_path / "first-root"
+    second = tmp_path / "second-root"
+    _write(first / "shared.bin", b"same")
+    _write(second / "shared.bin", b"same")
+    roots = {"first": first, "second": second}
+    e44 = build_e44_data_amplification_audit(
+        artifact_roots=roots,
+        output_root=tmp_path / "e44",
+    )
+    first_link = tmp_path / "first-link"
+    first_link.symlink_to(first, target_is_directory=True)
+
+    with pytest.raises(E50ContentReferenceIndexError, match="symlink"):
+        build_e50_content_reference_index(
+            e44_result_root=e44.result_root,
+            artifact_roots={"first": first_link, "second": second},
+            output_root=tmp_path / "rejected-e50",
+        )
+
+    result = build_e50_content_reference_index(
+        e44_result_root=e44.result_root,
+        artifact_roots=roots,
+        output_root=tmp_path / "e50",
+    )
+    with pytest.raises(E50ContentReferenceIndexError, match="symlink"):
+        resolve_e50_content_reference(
+            result.result_root,
+            {"first": first_link, "second": second},
+            root_label="first",
+            relative_path="shared.bin",
+        )