diff --git a/CLAUDE.md b/CLAUDE.md
index 949e1089e8..43b0bb8cfb 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -509,10 +509,13 @@ isolated XDG paths + archive root.
   exceptions opt out inline via `@pytest.mark.uses_real_clock("reason")`.
 - Pytest temp DBs pick ONE basetemp root via
   `devtools.verify_runs.resolve_pytest_basetemp_root` (shared by
-  `tests/conftest.py` and the `devtools test`/`verify` preflight): focused runs
-  use bounded `/dev/shm` tmpfs when it has ≥1 GiB free, while full-suite and
-  seed-testmon runs default to `/realm/tmp/polylogue-pytest` (NVMe) because
-  their aggregate fixture tree can exceed the supervised tmpfs ceiling.
+  `tests/conftest.py` and the `devtools test`/`verify` preflight). Bare pytest
+  without a managed run identity or an explicit basetemp root is forced to
+  `/realm/tmp/polylogue-pytest` (NVMe). Managed `devtools test` and
+  `devtools verify` runs may use bounded `/dev/shm` tmpfs only after the
+  runtime policy admits the requested demand; full-suite and seed-testmon
+  runs default to NVMe because their aggregate fixture tree can exceed the
+  supervised tmpfs ceiling.
   `POLYLOGUE_PYTEST_BASETEMP_MIN_FREE_MB` overrides required headroom; an
   explicit `POLYLOGUE_PYTEST_TMPFS=1` requests bounded tmpfs, but the request
   is honored only when the effective budget satisfies the declared basetemp
diff --git a/devtools/campaign_archive_location.py b/devtools/campaign_archive_location.py
index 72061fabb1..5d9d0073cd 100644
--- a/devtools/campaign_archive_location.py
+++ b/devtools/campaign_archive_location.py
@@ -66,6 +66,10 @@ def acquire(cls, archive_dir: Path, *, owner_id: str | None = None) -> CampaignA
         before any SQLite tier file is opened when ``archive_dir`` is already
         owned by another live campaign/maintenance process.
         """
+        # A campaign owns a fresh output directory.  Establish that directory
+        # before resolving its descriptor, then retain the normal ownership
+        # proof before a caller can open any SQLite tier.
+        archive_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
         location = ArchiveLocation.resolve(archive_dir)
         owned = OwnedArchiveLocation.acquire(location, owner_id=owner_id)
         return cls(owned=owned)
diff --git a/devtools/scale_regression_probe.py b/devtools/scale_regression_probe.py
index 5753127e05..2e0b971cc4 100644
--- a/devtools/scale_regression_probe.py
+++ b/devtools/scale_regression_probe.py
@@ -30,7 +30,7 @@
 from polylogue.storage.blob_store import BlobStore
 from polylogue.storage.insights.session import rebuild as rebuild_mod
 from polylogue.storage.insights.session.runtime import SessionInsightCounts
-from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database
+from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root, initialize_archive_database
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
 from polylogue.storage.sqlite.archive_tiers.write import write_parsed_session_to_archive
 from polylogue.storage.sqlite.connection import open_connection
@@ -67,10 +67,7 @@ def _session_id(native_id: str, origin: str = Origin.CODEX_SESSION.value) -> str
 
 
 def _init_archive(root: Path) -> None:
-    root.mkdir(parents=True, exist_ok=True)
-    initialize_archive_database(root / "source.db", ArchiveTier.SOURCE)
-    initialize_archive_database(root / "index.db", ArchiveTier.INDEX)
-    initialize_archive_database(root / "user.db", ArchiveTier.USER)
+    initialize_active_archive_root(root)
 
 
 def _parsed_session(native_id: str, *, title: str, messages: int = 1) -> ParsedSession:
diff --git a/devtools/verify.py b/devtools/verify.py
index fcdd9a1698..65d3e2c47f 100644
--- a/devtools/verify.py
+++ b/devtools/verify.py
@@ -2039,7 +2039,7 @@ def build_verify_steps(
                 f"({base_marker}) and not load_sensitive and not tui",
                 *_pytest_worker_args(),
             ]
-            steps.append(("pytest full (parallel)", bulk_cmd))
+            steps.append((BROAD_PYTEST_STEP_LABELS["full_parallel"], bulk_cmd))
 
             def _isolated_report_arg(arg: str) -> str:
                 # Keep the bulk lane's canonical report artifacts intact for
@@ -2052,11 +2052,11 @@ def _isolated_report_arg(arg: str) -> str:
 
             isolated_cmd = [_isolated_report_arg(arg) for arg in pytest_cmd]
             isolated_cmd.extend(["-m", f"({base_marker}) and (load_sensitive or tui)", "-p", "no:randomly", "-n", "0"])
-            steps.append(("pytest load-sensitive (isolated)", isolated_cmd))
+            steps.append((BROAD_PYTEST_STEP_LABELS["load_sensitive"], isolated_cmd))
         else:
             pytest_cmd.extend(["-m", base_marker, "--testmon", *_pytest_worker_args()])
             pytest_cmd.append("--testmon-forceselect")
-            label = "pytest testmon (broad)" if broad_testmon else "pytest testmon"
+            label = BROAD_PYTEST_STEP_LABELS["testmon_broad"] if broad_testmon else "pytest testmon"
             steps.append((label, pytest_cmd))
 
     if lab:
@@ -2185,6 +2185,15 @@ def _pytest_worker_args(*, maximum: int | None = None) -> list[str]:
     return ["-n", str(workers)]
 
 
+BROAD_PYTEST_STEP_LABELS = {
+    "seed": "pytest seed-testmon",
+    "seed_resume": "pytest seed-testmon (resume)",
+    "full_parallel": "pytest full (parallel)",
+    "load_sensitive": "pytest load-sensitive (isolated)",
+    "testmon_broad": "pytest testmon (broad)",
+}
+
+
 def _pytest_command_worker_request(cmd: Sequence[str]) -> str | None:
     """Return the last xdist worker request from a final pytest command.
 
@@ -2231,14 +2240,7 @@ def _pytest_command_concurrency(cmd: Sequence[str], *, env: Mapping[str, str] |
 
 def _pytest_uses_full_suite_basetemp(label: str) -> bool:
     """Whether this pytest step can materialize the measured full-suite tree."""
-    return label.startswith(
-        (
-            "pytest seed-testmon",
-            "pytest full",
-            "pytest load-sensitive",
-            "pytest testmon (broad)",
-        )
-    )
+    return label in BROAD_PYTEST_STEP_LABELS.values() or label.startswith("pytest seed-testmon shard ")
 
 
 _BROAD_TESTMON_CHANGED_PATHS = {
diff --git a/devtools/verify_runs.py b/devtools/verify_runs.py
index 1e1b4655f0..95949457ac 100644
--- a/devtools/verify_runs.py
+++ b/devtools/verify_runs.py
@@ -792,6 +792,11 @@ def apply_managed_pytest_runtime_policy(
             and effective_tmpfs_budget_kb < required_basetemp_kb
         ):
             normalized["POLYLOGUE_PYTEST_TMPFS"] = "0"
+            if configured_tmpfs:
+                # The configured tmpfs root has become unsafe for this run.
+                # Leaving it in place would make the resolver select it even
+                # though tmpfs has just been disabled, without its cap.
+                normalized.pop("POLYLOGUE_PYTEST_BASETEMP_ROOT", None)
     if default_full_suite_scratch:
         # Broad-suite demand grows with the fixture universe and has exceeded
         # the supervised 2 GiB ceiling while tests were still progressing.
diff --git a/polylogue/archive/revision_authority.py b/polylogue/archive/revision_authority.py
index d4a125ebb1..ecd515fc6f 100644
--- a/polylogue/archive/revision_authority.py
+++ b/polylogue/archive/revision_authority.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import sqlite3
 from collections.abc import Callable, Iterable
 from dataclasses import dataclass
 from enum import StrEnum
@@ -60,6 +61,41 @@ def canonical_authority_logical_key(logical_key: str) -> str:
     return f"{origin.value}:{native_id}"
 
 
+def logical_head_cohort_sql(conn: sqlite3.Connection, *, raw_alias: str, has_memberships: bool) -> str:
+    """Return the canonical, membership-aware SQL partition key for raw heads."""
+
+    def _canonical_or_original(value: object) -> str | None:
+        if value is None:
+            return None
+        text = str(value)
+        try:
+            return canonical_authority_logical_key(text)
+        except ValueError:
+            # A malformed legacy key must remain observable as its own cohort,
+            # never make a read-only verifier fail while grouping heads.
+            return text
+
+    conn.create_function("canonical_authority_logical_key", 1, _canonical_or_original, deterministic=True)
+    membership_key = "NULL"
+    if has_memberships:
+        membership_key = f"""
+            canonical_authority_logical_key(
+                (
+                    SELECT CASE
+                        WHEN COUNT(DISTINCT canonical_authority_logical_key(m.logical_source_key)) = 1
+                        THEN MIN(canonical_authority_logical_key(m.logical_source_key))
+                    END
+                    FROM raw_session_memberships AS m
+                    WHERE m.raw_id = {raw_alias}.raw_id
+                )
+            )
+        """
+    return (
+        f"COALESCE(canonical_authority_logical_key({raw_alias}.logical_source_key), "
+        f"{membership_key}, {raw_alias}.native_id, {raw_alias}.source_path)"
+    )
+
+
 def durable_authority_logical_keys(
     *,
     raw_logical_key: object,
diff --git a/polylogue/daemon/bulk_rebuild.py b/polylogue/daemon/bulk_rebuild.py
index 7565e4792a..92e8aa87e5 100644
--- a/polylogue/daemon/bulk_rebuild.py
+++ b/polylogue/daemon/bulk_rebuild.py
@@ -492,6 +492,7 @@ def run_daemon_canary_rebuild(
             archive_root=str(root),
             index_schema_version=index_schema_version,
             daemon_version=POLYLOGUE_VERSION,
+            accept_degraded=True,
         )
         is None
     ):
diff --git a/polylogue/daemon_client.py b/polylogue/daemon_client.py
index 944441fa7c..58cf9136fc 100644
--- a/polylogue/daemon_client.py
+++ b/polylogue/daemon_client.py
@@ -47,7 +47,29 @@ def request_json(
         body: dict[str, object] | None = None,
         *,
         raise_for_status: bool = False,
+        accepted_statuses: frozenset[int] = frozenset({200}),
     ) -> dict[str, Any] | None:
+        response = self._request_json_response(method, path, body)
+        if response is None:
+            return None
+        status, payload = response
+        if status not in accepted_statuses:
+            if raise_for_status:
+                envelope = payload if isinstance(payload, dict) else {}
+                code = envelope.get("error")
+                detail = envelope.get("detail")
+                raise DaemonResponseError(
+                    status=status,
+                    code=code if isinstance(code, str) else None,
+                    detail=detail if isinstance(detail, str) else None,
+                )
+            return None
+        return payload
+
+    def _request_json_response(
+        self, method: str, path: str, body: dict[str, object] | None = None
+    ) -> tuple[int, dict[str, Any] | None] | None:
+        """Return the response status with its decoded JSON object, if any."""
         if not self.socket_path.exists():
             return None
         connection = _UnixHTTPConnection(self.socket_path, self.timeout_s)
@@ -59,20 +81,9 @@ def request_json(
                 headers["Authorization"] = f"Bearer {self.auth_token}"
             connection.request(method, path, body=raw, headers=headers)
             response = connection.getresponse()
-            payload = json.loads(response.read().decode())
+            decoded = json.loads(response.read().decode())
             self.last_elapsed_ms = round((perf_counter() - started_at) * 1000)
-            if response.status != 200:
-                if raise_for_status:
-                    envelope = payload if isinstance(payload, dict) else {}
-                    code = envelope.get("error")
-                    detail = envelope.get("detail")
-                    raise DaemonResponseError(
-                        status=response.status,
-                        code=code if isinstance(code, str) else None,
-                        detail=detail if isinstance(detail, str) else None,
-                    )
-                return None
-            return payload if isinstance(payload, dict) else None
+            return response.status, decoded if isinstance(decoded, dict) else None
         except (OSError, TimeoutError, ValueError, http.client.HTTPException):
             return None
         finally:
@@ -81,8 +92,31 @@ def request_json(
     def cli_query(self, params: dict[str, object]) -> dict[str, Any] | None:
         return self.request_json("POST", "/api/cli/query", {"params": params})
 
-    def probe(self, *, archive_root: str, index_schema_version: int, daemon_version: str) -> dict[str, Any] | None:
-        health = self.request_json("GET", "/api/health")
+    def probe(
+        self,
+        *,
+        archive_root: str,
+        index_schema_version: int,
+        daemon_version: str,
+        accept_degraded: bool = False,
+    ) -> dict[str, Any] | None:
+        """Return identity only for the daemon serving the requested archive.
+
+        Maintenance callers may accept only the health endpoint's typed
+        ``degraded`` lifecycle 503 envelope in order to reach the daemon-owned
+        repair route. This does not authorize the repair: the write endpoint
+        still runs its typed preflight. Query callers retain the strict
+        200-only default.
+        """
+        response = self._request_json_response("GET", "/api/health")
+        if response is None:
+            return None
+        status, health = response
+        if status == 503:
+            if not accept_degraded or health is None or health.get("raw_failure_lifecycle_state") != "degraded":
+                return None
+        elif status != 200:
+            return None
         if health is None:
             return None
         if health.get("archive_root") != archive_root:
diff --git a/polylogue/maintenance/archive_verification.py b/polylogue/maintenance/archive_verification.py
index 574844a452..e25e2e8619 100644
--- a/polylogue/maintenance/archive_verification.py
+++ b/polylogue/maintenance/archive_verification.py
@@ -41,6 +41,7 @@
 from pathlib import Path
 from typing import Any
 
+from polylogue.archive.revision_authority import logical_head_cohort_sql
 from polylogue.core.json import JSONDocument, json_document
 from polylogue.core.outcomes import OutcomeCheck, OutcomeReport, OutcomeStatus
 from polylogue.logging import get_logger
@@ -478,6 +479,53 @@ def _check_pointer_coherence(archive_root: Path, _sample_limit: int) -> ArchiveV
 # ---------------------------------------------------------------------------
 
 
+def _valid_byte_duplicate_supersession_expr(conn: sqlite3.Connection, *, raw_alias: str) -> str:
+    """Return the receipt predicate shared by source/index coverage checks.
+
+    A supersession receipt is authority only when it still names the same
+    bytes and an index materialization of the recorded duplicate twin. Keep
+    the predicate in one place so backlog freshness cannot classify a receipt
+    differently from source-index coverage.
+    """
+    if not table_exists(conn, "raw_byte_duplicate_supersession_receipts"):
+        return "0"
+    return f"""
+        EXISTS(
+            SELECT 1
+            FROM raw_byte_duplicate_supersession_receipts receipt
+            JOIN raw_sessions twin ON twin.raw_id = receipt.duplicate_of_raw_id
+            JOIN idx_tier.sessions twin_session
+              ON twin_session.raw_id = twin.raw_id
+             AND twin_session.session_id = receipt.duplicate_of_session_id
+            WHERE receipt.raw_id = {raw_alias}.raw_id
+              AND receipt.blob_hash = {raw_alias}.blob_hash
+              AND receipt.blob_size = {raw_alias}.blob_size
+              AND twin.blob_hash = {raw_alias}.blob_hash
+              AND twin.blob_size = {raw_alias}.blob_size
+              AND twin.origin IS {raw_alias}.origin
+              AND twin.source_path IS {raw_alias}.source_path
+              AND twin.source_index IS {raw_alias}.source_index
+        )
+    """
+
+
+def _logical_head_cohort_expr(conn: sqlite3.Connection, *, raw_alias: str) -> str:
+    """Return the durable identity used to group raw revisions into one head.
+
+    A full-revision row retired into membership governance intentionally loses
+    its raw-level ``logical_source_key``.  Its single retained membership key
+    remains the authoritative identity, so use it before the legacy
+    native-id/path fallback.  Shared raws can hold several membership keys;
+    they have no one raw-level cohort and must keep that fallback instead of
+    being arbitrarily assigned to one member.
+    """
+    return logical_head_cohort_sql(
+        conn,
+        raw_alias=raw_alias,
+        has_memberships=table_exists(conn, "raw_session_memberships"),
+    )
+
+
 def _check_source_index_coverage(archive_root: Path, sample_limit: int) -> ArchiveVerificationCheck:
     return _check_source_index_coverage_at_index_path(archive_root, _resolve_index_path(archive_root), sample_limit)
 
@@ -488,16 +536,19 @@ def _check_source_index_coverage_at_index_path(
     """Every logical source's head is indexed OR carries a typed refusal.
 
     Universe (polylogue-r4jiu, invariant I1): ``raw_sessions`` logical heads --
-    the latest revision per ``(origin, COALESCE(native_id, source_path))`` --
+    the latest revision per
+    ``(origin, COALESCE(logical_source_key, native_id, source_path))`` --
     which is a ground-truth table every acquired raw lands in, not a ledger a
     downstream reconciliation stage can silently omit rows from. A logical
     head counts as covered when *any* raw in its revision group is
     materialized into ``index.db``. An uncovered head must be typed as one
     of: ``parse_error`` (raw_sessions.parse_error set), ``non_session`` /
     ``census_failed`` (raw_membership_census recorded a terminal non-complete
-    verdict), or ``quarantined`` (raw_sessions.revision_authority --
-    reconciliation hasn't granted it authority to write yet, WARN-level
-    evidence, not blocking). An uncovered head matching none of those is an
+    verdict), a content-bound byte-supersession receipt whose named twin is
+    present in the candidate index, or ``quarantined``
+    (raw_sessions.revision_authority -- reconciliation hasn't granted it
+    authority to write yet, WARN-level evidence, not blocking). An uncovered
+    head matching none of those is an
     *untyped* gap -- a materialization failure no other subsystem has
     explained -- and is the only ERROR-gating condition here besides orphans
     (index sessions whose raw_id doesn't exist in source.db at all).
@@ -522,6 +573,8 @@ def _check_source_index_coverage_at_index_path(
             census_expr = (
                 "(SELECT c.status FROM raw_membership_census c WHERE c.raw_id = r.raw_id)" if has_census else "NULL"
             )
+            valid_supersession_expr = _valid_byte_duplicate_supersession_expr(conn, raw_alias="r")
+            logical_cohort_expr = _logical_head_cohort_expr(conn, raw_alias="r")
 
             # A read-only connection (``query_only=ON``, connection-wide, not
             # per-attached-db) cannot ``CREATE TEMP VIEW`` -- the temp schema
@@ -534,11 +587,17 @@ def _check_source_index_coverage_at_index_path(
                         r.blob_hash,
                         r.parse_error,
                         r.revision_authority,
+                        r.logical_source_key,
                         {census_expr} AS census_status,
+                        {valid_supersession_expr} AS valid_supersession,
                         MAX(EXISTS(SELECT 1 FROM idx_tier.sessions s WHERE s.raw_id = r.raw_id))
-                            OVER (PARTITION BY r.origin, COALESCE(r.native_id, r.source_path)) AS any_indexed,
+                            OVER (
+                                PARTITION BY r.origin,
+                                             {logical_cohort_expr}
+                            ) AS any_indexed,
                         ROW_NUMBER() OVER (
-                            PARTITION BY r.origin, COALESCE(r.native_id, r.source_path)
+                            PARTITION BY r.origin,
+                                         {logical_cohort_expr}
                             ORDER BY r.acquired_at_ms DESC, r.raw_id DESC
                         ) AS rn
                     FROM raw_sessions r
@@ -547,11 +606,13 @@ def _check_source_index_coverage_at_index_path(
             untyped_predicate = """
                 rn = 1 AND any_indexed = 0 AND parse_error IS NULL
                   AND COALESCE(census_status, '') NOT IN ('non_session', 'failed')
+                  AND valid_supersession = 0
                   AND revision_authority != 'quarantined'
             """
             quarantined_predicate = """
                 rn = 1 AND any_indexed = 0 AND parse_error IS NULL
                   AND COALESCE(census_status, '') NOT IN ('non_session', 'failed')
+                  AND valid_supersession = 0
                   AND revision_authority = 'quarantined'
             """
 
@@ -567,12 +628,13 @@ def _check_source_index_coverage_at_index_path(
                   SUM(CASE WHEN any_indexed = 0 AND parse_error IS NULL
                             AND COALESCE(census_status, '') != 'non_session'
                             AND census_status = 'failed' THEN 1 ELSE 0 END),
+                  SUM(CASE WHEN any_indexed = 0 AND valid_supersession = 1 THEN 1 ELSE 0 END),
                   SUM(CASE WHEN {quarantined_predicate.replace("rn = 1 AND ", "")} THEN 1 ELSE 0 END),
                   SUM(CASE WHEN {untyped_predicate.replace("rn = 1 AND ", "")} THEN 1 ELSE 0 END)
                 FROM heads WHERE rn = 1
                 """
             ).fetchone()
-            parse_error_n, non_session_n, census_failed_n, quarantined_n, untyped_n = (
+            parse_error_n, non_session_n, census_failed_n, superseded_n, quarantined_n, untyped_n = (
                 int(value or 0) for value in counts
             )
 
@@ -609,6 +671,7 @@ def _check_source_index_coverage_at_index_path(
                     {heads_cte}
                     SELECT SUM(CASE WHEN {quarantined_predicate.replace("rn = 1 AND ", "")}
                                        OR {untyped_predicate.replace("rn = 1 AND ", "")}
+                                       OR valid_supersession = 1
                                   THEN (
                                     EXISTS(
                                         SELECT 1 FROM raw_sessions dup
@@ -645,7 +708,7 @@ def _check_source_index_coverage_at_index_path(
     finally:
         conn.close()
 
-    unindexed_head_count = parse_error_n + non_session_n + census_failed_n + quarantined_n + untyped_n
+    unindexed_head_count = parse_error_n + non_session_n + census_failed_n + superseded_n + quarantined_n + untyped_n
     blocking = untyped_n > 0 or int(orphan_count or 0) > 0
     warning = quarantined_n > 0
 
@@ -667,6 +730,8 @@ def _check_source_index_coverage_at_index_path(
         parts.append(f"parse_error={parse_error_n:,}")
     if non_session_n or census_failed_n:
         parts.append(f"declared-non-session={non_session_n + census_failed_n:,}")
+    if superseded_n:
+        parts.append(f"superseded-byte-duplicate={superseded_n:,}")
     if byte_dup_of_indexed_n:
         parts.append(
             f"byte-dup-of-indexed={byte_dup_of_indexed_n:,} (novel={novel_unindexed_n:,} of {unindexed_head_count:,})"
@@ -692,6 +757,7 @@ def _check_source_index_coverage_at_index_path(
             "parse_error_count": parse_error_n,
             "non_session_count": non_session_n,
             "census_failed_count": census_failed_n,
+            "superseded_byte_duplicate_count": superseded_n,
             "quarantined_count": quarantined_n,
             "quarantined_sample": quarantined_sample,
             "byte_dup_of_indexed_count": byte_dup_of_indexed_n,
@@ -1220,7 +1286,11 @@ def _check_raw_failure_lifecycle(archive_root: Path, sample_limit: int) -> Archi
             evidence=evidence,
         )
     known = snapshot.deferred + snapshot.terminal
-    status = OutcomeStatus.WARNING if known else OutcomeStatus.OK
+    # Terminal evidence is the completed state this invariant requires. Keep
+    # it visible in counts/evidence, but do not emit a warning that strict
+    # candidate acceptance interprets as a permanent veto. Deferred failures
+    # remain warning-level because their source disposition is still open.
+    status = OutcomeStatus.WARNING if snapshot.deferred else OutcomeStatus.OK
     summary = (
         f"{known:,} raw failure(s) classified ({snapshot.deferred:,} deferred, {snapshot.terminal:,} terminal)"
         if known
@@ -2354,17 +2424,25 @@ def _unindexed_backlog_gap(conn: sqlite3.Connection) -> int:
     """
     has_census = table_exists(conn, "raw_membership_census")
     census_expr = "(SELECT c.status FROM raw_membership_census c WHERE c.raw_id = r.raw_id)" if has_census else "NULL"
+    valid_supersession_expr = _valid_byte_duplicate_supersession_expr(conn, raw_alias="r")
+    logical_cohort_expr = _logical_head_cohort_expr(conn, raw_alias="r")
     row = conn.execute(
         f"""
         WITH heads AS (
             SELECT
                 r.raw_id,
                 r.parse_error,
+                r.logical_source_key,
                 {census_expr} AS census_status,
+                {valid_supersession_expr} AS valid_supersession,
                 MAX(EXISTS(SELECT 1 FROM idx_tier.sessions s WHERE s.raw_id = r.raw_id))
-                    OVER (PARTITION BY r.origin, COALESCE(r.native_id, r.source_path)) AS any_indexed,
+                    OVER (
+                        PARTITION BY r.origin,
+                                     {logical_cohort_expr}
+                    ) AS any_indexed,
                 ROW_NUMBER() OVER (
-                    PARTITION BY r.origin, COALESCE(r.native_id, r.source_path)
+                    PARTITION BY r.origin,
+                                 {logical_cohort_expr}
                     ORDER BY r.acquired_at_ms DESC, r.raw_id DESC
                 ) AS rn
             FROM raw_sessions r
@@ -2372,6 +2450,7 @@ def _unindexed_backlog_gap(conn: sqlite3.Connection) -> int:
         SELECT SUM(
             CASE WHEN rn = 1 AND any_indexed = 0 AND parse_error IS NULL
                       AND COALESCE(census_status, '') NOT IN ('non_session', 'failed')
+                      AND valid_supersession = 0
                  THEN 1 ELSE 0 END
         )
         FROM heads WHERE rn = 1
diff --git a/polylogue/maintenance/reindex_canary.py b/polylogue/maintenance/reindex_canary.py
index 3f4622a2b1..0a2af28256 100644
--- a/polylogue/maintenance/reindex_canary.py
+++ b/polylogue/maintenance/reindex_canary.py
@@ -345,15 +345,9 @@ def run_reindex_canary(
         raise CanarySelectionError("reindex canary requires an explicit schema-inference receipt path")
     if not no_promote:
         raise CanarySelectionError("reindex canary requires --no-promote")
-    from polylogue.config import resolve_archive_root
     from polylogue.storage.archive_identity import ArchiveLocation, TierFileIdentity
 
     root = Path(archive_root)
-    if root.resolve() == resolve_archive_root().resolve():
-        raise CanarySelectionError(
-            "reindex canary refuses the configured live archive root; "
-            "run it against an explicitly provisioned isolated canary archive"
-        )
     current_index = _resolve_canary_input_index(root, input_index)
     location = ArchiveLocation.resolve(root)
     if input_index is not None:
diff --git a/polylogue/schemas/sampling_db.py b/polylogue/schemas/sampling_db.py
index 95650e4790..f072789dfb 100644
--- a/polylogue/schemas/sampling_db.py
+++ b/polylogue/schemas/sampling_db.py
@@ -15,6 +15,7 @@
 from polylogue.archive.artifact_taxonomy import classify_artifact_path
 from polylogue.archive.raw_payload import extract_record_samples_from_raw_content
 from polylogue.archive.raw_payload.decode import RawPayloadEnvelope
+from polylogue.archive.revision_authority import logical_head_cohort_sql
 from polylogue.core.enums import Origin, Provider
 from polylogue.core.json import JSONDocument, JSONValue, require_json_value
 from polylogue.core.provider_identity import (
@@ -36,6 +37,7 @@
     ObservationTerminalStatus,
 )
 from polylogue.storage.blob_store import get_blob_store
+from polylogue.storage.introspection import table_exists
 from polylogue.storage.sqlite.connection_profile import connection_context
 
 logger = get_logger(__name__)
@@ -260,7 +262,7 @@ def _iter_schema_units_from_db(
 
     ``logical_heads_only`` (default ``False``, opt-in): restrict the sampled
     rows to one per logical source -- the latest revision per
-    ``(origin, COALESCE(native_id, source_path))``, the same grouping the
+    ``(origin, COALESCE(logical_source_key, native_id, source_path))``, the same grouping the
     archive-verification I1 coverage check uses for its universe. Schema
     inference's ordinary use (discovering every shape a provider's wire
     format can take) wants every revision, since an older revision can carry
@@ -289,31 +291,36 @@ def _iter_schema_units_from_db(
     query_provider = config.db_source_name or source_name
     origins = _sample_origins_for_provider(Provider.from_string(query_provider), config)
     placeholders = ",".join("?" for _ in origins)
-    if logical_heads_only:
-        query = f"""
-            WITH heads AS (
-                SELECT
-                    source_path, origin, raw_id, blob_hash, file_mtime_ms, acquired_at_ms,
-                    validation_status,
-                    ROW_NUMBER() OVER (
-                        PARTITION BY origin, COALESCE(native_id, source_path)
-                        ORDER BY acquired_at_ms DESC, raw_id DESC
-                    ) AS rn
-                FROM raw_sessions
-                WHERE origin IN ({placeholders})
-            )
-            SELECT source_path, origin, raw_id, blob_hash, file_mtime_ms, acquired_at_ms, validation_status
-            FROM heads WHERE rn = 1
-        """
-    else:
-        query = f"""
-            SELECT source_path, origin, raw_id, blob_hash, file_mtime_ms, acquired_at_ms,
-                   validation_status
-            FROM raw_sessions
-            WHERE origin IN ({placeholders})
-        """
     with connection_context(source_db_path) as conn:
         conn.row_factory = sqlite3.Row
+        if logical_heads_only:
+            logical_cohort_expr = logical_head_cohort_sql(
+                conn,
+                raw_alias="raw_sessions",
+                has_memberships=table_exists(conn, "raw_session_memberships"),
+            )
+            query = f"""
+                WITH heads AS (
+                    SELECT
+                        source_path, origin, raw_id, blob_hash, file_mtime_ms, acquired_at_ms,
+                        validation_status,
+                        ROW_NUMBER() OVER (
+                            PARTITION BY origin, {logical_cohort_expr}
+                            ORDER BY acquired_at_ms DESC, raw_id DESC
+                        ) AS rn
+                    FROM raw_sessions
+                    WHERE origin IN ({placeholders})
+                )
+                SELECT source_path, origin, raw_id, blob_hash, file_mtime_ms, acquired_at_ms, validation_status
+                FROM heads WHERE rn = 1
+            """
+        else:
+            query = f"""
+                SELECT source_path, origin, raw_id, blob_hash, file_mtime_ms, acquired_at_ms,
+                       validation_status
+                FROM raw_sessions
+                WHERE origin IN ({placeholders})
+            """
         cursor = conn.execute(query, origins)
         batch_size = 1 if config.sample_granularity == "record" else 100
         # Content-hash dedup: distinct ``raw_id``s legitimately collapse onto the
diff --git a/polylogue/sources/assembly_chatgpt.py b/polylogue/sources/assembly_chatgpt.py
index cc057ec98b..7e8b163039 100644
--- a/polylogue/sources/assembly_chatgpt.py
+++ b/polylogue/sources/assembly_chatgpt.py
@@ -14,8 +14,7 @@
 Resolution results are recorded as ``session_events`` rather than new
 attachment/schema columns (index.db is a derived tier; a schema bump needs a
 declared delta class) — same precedent as this file's neighbors
-(``chatgpt.py``'s ``chatgpt_block_metadata``/``repo_identity_evidence``
-events). ``provider_file_id`` IS updated in place when an id-grade match is
+(``chatgpt.py``'s ``chatgpt_block_metadata`` events). ``provider_file_id`` IS updated in place when an id-grade match is
 found (tiers 1-4 of the sandbox resolver, or any ``.dat`` id resolution) —
 that is a real identity strengthening, not a guess.
 """
diff --git a/polylogue/sources/emitter.py b/polylogue/sources/emitter.py
index 4f26e38530..79bc71d16a 100644
--- a/polylogue/sources/emitter.py
+++ b/polylogue/sources/emitter.py
@@ -18,7 +18,7 @@
 from .decoder_json import JsonValue
 from .decoders import _iter_json_stream
 from .dispatch import GROUP_PROVIDERS, detect_provider, is_jsonl_source_path, parse_payload
-from .parsers.base import ParsedSession, ParsedSessionEvent, RawSessionData
+from .parsers.base import ParsedSession, RawSessionData
 
 if TYPE_CHECKING:
     from polylogue.schemas.packages import SchemaResolution
@@ -444,57 +444,7 @@ def _maybe_enrich(
         spec = get_assembly_spec(p)
         if spec is not None:
             conv = spec.enrich_session(conv, self._ctx.sidecar_data)
-        return _append_repo_identity_evidence(conv)
-
-
-def _append_repo_identity_evidence(conv: ParsedSession) -> ParsedSession:
-    """Record whether a session has real git evidence or is merely a cwd.
-
-    polylogue-cijx.2 (measured 2026-07-29 on the live archive): for ~84% of
-    sessions the only "repository" evidence is ``working_directories`` --
-    the archive's ``repos``/``session_repos`` tables key on that cwd today,
-    so a session opened in ``/home/sinity`` is recorded as being in the
-    "sinity" repository, when no git evidence of a repository exists at all.
-    Per cijx.4 decision 1, a session with no git evidence resolves to a
-    DIRECTORY, not a repository, and read surfaces must be able to say which.
-
-    The storage-side identity rework (a content-addressed repository key,
-    separate checkout/observation entities) is out of this lane's write scope
-    (a concurrent lane owns storage/sqlite/** this cycle). This function is
-    the parser-side half: it stamps the grade this session's evidence
-    actually supports as a typed, schema-free session_event (event_type has
-    no CHECK vocabulary, so this needs no migration) so a downstream reader
-    -- or the storage-side rework once it lands -- can tell a real
-    repository observation from a bare directory without re-deriving it from
-    working_directories.
-
-    Runs once, at the emitter's enrichment boundary -- after all
-    provider-specific sidecar enrichment (e.g. Claude Code's sessions-index.json
-    git_branch merge) has already applied, and before the session leaves
-    sources/** for pipeline/storage.
-    """
-    has_git_evidence = bool(
-        (conv.git_repository_url and conv.git_repository_url.strip())
-        or (conv.git_branch and conv.git_branch.strip())
-        or (conv.git_commit_hash and conv.git_commit_hash.strip())
-    )
-    root_paths = sorted({path.strip() for path in conv.working_directories if path.strip()})
-    if not has_git_evidence and not root_paths:
-        # No location evidence of any kind -- nothing to grade.
         return conv
-    payload: dict[str, object] = {
-        "grade": "git_evidence" if has_git_evidence else "directory_only",
-        "root_paths": root_paths,
-        "git_repository_url": conv.git_repository_url,
-        "git_branch": conv.git_branch,
-        "git_commit_hash": conv.git_commit_hash,
-    }
-    event = ParsedSessionEvent(
-        event_type="repo_identity_evidence",
-        timestamp=conv.created_at,
-        payload=payload,
-    )
-    return conv.model_copy(update={"session_events": [*conv.session_events, event]})
 
 
 __all__ = [
diff --git a/polylogue/sources/origin_specs.py b/polylogue/sources/origin_specs.py
index 9bc524179b..4ed4d364e2 100644
--- a/polylogue/sources/origin_specs.py
+++ b/polylogue/sources/origin_specs.py
@@ -48,6 +48,7 @@
 _SOURCE_ROOT = Path(__file__).resolve().parents[2]
 _LOWERING_FINGERPRINT_PATHS: tuple[str, ...] = (
     "polylogue/sources/dispatch.py",
+    "polylogue/sources/emitter.py",
     "polylogue/pipeline/ids.py",
     "polylogue/storage/sqlite/archive_tiers/write.py",
     "polylogue/archive/session_revision_membership.py",
diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py
index 3b66588fcc..209b9ceb83 100644
--- a/polylogue/sources/revision_backfill.py
+++ b/polylogue/sources/revision_backfill.py
@@ -729,9 +729,19 @@ def bind_byte_proven_older_member(raw_id: str, logical_key: str) -> None:
             census_selection = initial_selection
             while True:
                 rows = archive.raw_membership_census_rows(census_selection)
-                pending_rows = [(raw_id, source_index) for raw_id, source_index in rows if raw_id not in state.censused]
+                terminal_raw_ids = {
+                    raw_id for raw_id, _source_index, terminal_non_session in rows if terminal_non_session
+                }
+                for raw_id in terminal_raw_ids - state.censused:
+                    state.scanned += 1
+                    state.censused.add(raw_id)
+                pending_rows = [
+                    (raw_id, source_index)
+                    for raw_id, source_index, terminal_non_session in rows
+                    if raw_id not in state.censused and not terminal_non_session
+                ]
                 if max_payload_bytes is not None:
-                    payload_sizes = archive.raw_payload_sizes([raw_id for raw_id, _index in rows])
+                    payload_sizes = archive.raw_payload_sizes([raw_id for raw_id, _index in pending_rows])
                     total_payload_bytes = sum(payload_sizes.values())
                     oversized = [raw_id for raw_id, size in payload_sizes.items() if size > max_payload_bytes]
                     if oversized or total_payload_bytes > max_payload_bytes:
@@ -819,14 +829,18 @@ def _load_frozen_revision_evidence(
     )
     rows = archive.raw_membership_census_rows(expanded_raw_ids if selected_raw_ids is not None else None)
     if max_payload_bytes is not None:
-        payload_sizes = archive.raw_payload_sizes([raw_id for raw_id, _source_index in rows])
+        payload_sizes = archive.raw_payload_sizes(
+            [raw_id for raw_id, _source_index, terminal_non_session in rows if not terminal_non_session]
+        )
         total_payload_bytes = sum(payload_sizes.values())
         oversized = [raw_id for raw_id, size in payload_sizes.items() if size > max_payload_bytes]
         if oversized or total_payload_bytes > max_payload_bytes:
             raise RawRevisionReplayResourceBlockedError(
                 sorted(oversized or payload_sizes), max_payload_bytes, total_payload_bytes
             )
-    parseable_raw_ids = [raw_id for raw_id, source_index in rows if source_index >= 0]
+    parseable_raw_ids = [
+        raw_id for raw_id, source_index, terminal_non_session in rows if source_index >= 0 and not terminal_non_session
+    ]
     parsed_outcomes = _parse_retained_raws(
         archive,
         parseable_raw_ids,
@@ -834,9 +848,11 @@ def _load_frozen_revision_evidence(
         prefetch_cache=prefetch_cache,
     )
     state = _RevisionCensusState(0, 0, 0, set(), {}, {})
-    for raw_id, source_index in rows:
+    for raw_id, source_index, terminal_non_session in rows:
         state.scanned += 1
         state.censused.add(raw_id)
+        if terminal_non_session:
+            continue
         if source_index < 0:
             state.quarantined += 1
             continue
@@ -952,11 +968,15 @@ def require_current_parser_source_census(
 
     invalid_durable_bindings: set[str] = set()
     durable_logical_keys: dict[str, tuple[str, ...]] = {}
-    for raw_id, (typed_key, revision_kind, membership_keys, _typed_non_session) in durable_bindings.items():
-        durable_keys = durable_authority_logical_keys(
-            raw_logical_key=typed_key,
-            revision_kind=revision_kind,
-            membership_logical_keys=membership_keys,
+    for raw_id, (typed_key, revision_kind, membership_keys, typed_non_session) in durable_bindings.items():
+        durable_keys = (
+            ()
+            if typed_non_session
+            else durable_authority_logical_keys(
+                raw_logical_key=typed_key,
+                revision_kind=revision_kind,
+                membership_logical_keys=membership_keys,
+            )
         )
         if durable_keys is None:
             invalid_durable_bindings.add(raw_id)
@@ -1090,6 +1110,10 @@ def require_current_parser_source_census(
                     LEFT JOIN raw_session_memberships AS m ON m.raw_id = r.raw_id
                     WHERE r.revision_authority = 'quarantined'
                       {authority_where}
+                      AND NOT EXISTS (
+                          SELECT 1 FROM raw_artifacts AS a
+                          WHERE a.raw_id = r.raw_id AND a.parse_as_session = 0
+                      )
                       AND (
                           c.raw_id IS NULL OR c.status NOT IN ('complete', 'non_session')
                           OR (
@@ -2054,9 +2078,96 @@ def _parse_retained_raws(
                 sessions, _rep_size, _rep_kind = outcome
                 _provider, _blob_hash, _source_path, kind, size, _native_id = descriptors[raw_id]
                 results[raw_id] = (sessions, size, kind)
+    _enrich_retained_parse_results(archive, descriptors=descriptors, results=results)
     return results
 
 
+def _enrich_retained_parse_results(
+    archive: ArchiveStore,
+    *,
+    descriptors: dict[str, tuple[Provider, str, str, RawRevisionKind, int, str | None]],
+    results: dict[str, tuple[list[ParsedSession], int, RawRevisionKind] | Exception],
+) -> None:
+    """Apply replay-safe provider assembly to decoded retained raws.
+
+    Initial file ingest routes every parsed session through the provider
+    assembly layer before hashing and writing it. Raw replay historically
+    skipped that layer, so even deterministic fallbacks (for example Codex's
+    first-human-message title) changed the session hash and made byte-proven
+    source cohorts permanently non-adoptable.
+
+    Replay may consume only frozen source evidence. It therefore uses the
+    earliest persisted ``history_sidecars`` snapshot for an acquisition path,
+    plus durable Codex title hook events, and otherwise supplies an explicit
+    empty sidecar set. It never rediscovers mutable files beside the original
+    source path. Providers whose enrichment was never captured remain
+    conservatively unenriched and will still fail an exact hash comparison
+    rather than being reconstructed from ambient filesystem state.
+    """
+    # Unit-level parser/dedupe probes deliberately pass tiny protocol fakes;
+    # enrichment is an ArchiveStore production concern and is covered through
+    # real source-tier replay fixtures below. Do not turn those pure decode
+    # probes into accidental SQLite integration tests.
+    if not isinstance(archive, ArchiveStore):
+        return
+    from polylogue.sources.assembly_codex import read_codex_thread_title_hook_events
+
+    source_conn = archive._ensure_source_conn()
+    codex_hook_titles = read_codex_thread_title_hook_events(source_conn)
+    sidecars_by_path: dict[tuple[Origin, str], object] = {}
+    for raw_id, outcome in tuple(results.items()):
+        if isinstance(outcome, Exception):
+            continue
+        provider, _blob_hash, source_path, _descriptor_kind, _size, _native_id = descriptors[raw_id]
+        sessions, payload_bytes, kind = outcome
+        results[raw_id] = (
+            _replay_safe_enrich_sessions(
+                source_conn,
+                provider=provider,
+                source_path=source_path,
+                sessions=sessions,
+                sidecars_by_path=sidecars_by_path,
+                codex_hook_titles=codex_hook_titles,
+            ),
+            payload_bytes,
+            kind,
+        )
+
+
+def _replay_safe_enrich_sessions(
+    source_conn: sqlite3.Connection,
+    *,
+    provider: Provider,
+    source_path: str,
+    sessions: list[ParsedSession],
+    sidecars_by_path: dict[tuple[Origin, str], object],
+    codex_hook_titles: dict[str, str],
+) -> list[ParsedSession]:
+    """Enrich one retained parse from frozen source-tier sidecars only."""
+    from polylogue.sources.assembly import SidecarData, get_assembly_spec
+    from polylogue.storage.sqlite.archive_tiers.source_write import read_earliest_history_sidecar_for_path
+
+    spec = get_assembly_spec(provider)
+    if spec is None:
+        return sessions
+    origin = origin_from_provider(provider)
+    sidecar_key = (origin, source_path)
+    cached = sidecars_by_path.get(sidecar_key)
+    if cached is None:
+        persisted = read_earliest_history_sidecar_for_path(
+            source_conn,
+            origin=origin,
+            source_path=source_path,
+        )
+        sidecar_data = cast("SidecarData", dict(persisted.payload) if persisted is not None else {})
+        if provider is Provider.CODEX and codex_hook_titles:
+            sidecar_data = cast("SidecarData", {**sidecar_data, "hook_event_titles": codex_hook_titles})
+        sidecars_by_path[sidecar_key] = sidecar_data
+    else:
+        sidecar_data = cast("SidecarData", cached)
+    return [spec.enrich_session(session, sidecar_data) for session in sessions]
+
+
 def _parse_unique_retained_raws_via_threads(
     archive: ArchiveStore,
     raw_ids: list[str],
@@ -2441,14 +2552,16 @@ def _run_inner(self, generation: int, keys: tuple[str, ...], extra_members: dict
         # NOTE: ``with sqlite3.connect(...)`` would only manage a
         # transaction, not the connection lifetime -- close explicitly.
         source_conn = sqlite3.connect(f"file:{self._source_db_path}?mode=ro", uri=True, timeout=30.0)
+        spill_conn: sqlite3.Connection | None = None
         try:
             plan, descriptors = self._build_plan(source_conn, keys, extra_members)
-        finally:
-            source_conn.close()
-        if not plan:
-            return
-        spill_conn = sqlite3.connect(self._spill.path, timeout=30.0)
-        try:
+            if not plan:
+                return
+            from polylogue.sources.assembly_codex import read_codex_thread_title_hook_events
+
+            codex_hook_titles = read_codex_thread_title_hook_events(source_conn)
+            sidecars_by_path: dict[tuple[Origin, str], object] = {}
+            spill_conn = sqlite3.connect(self._spill.path, timeout=30.0)
             spill_conn.execute("PRAGMA busy_timeout = 30000")
             for seq, raw_id in plan:
                 if self._wait_for_budget(generation, seq) is False:
@@ -2460,7 +2573,14 @@ def _run_inner(self, generation: int, keys: tuple[str, ...], extra_members: dict
                         continue
                 if raw_id in self._spill._decoded or raw_id in self._spill._whales:
                     continue
-                decoded = self._decode(spill_conn, raw_id, descriptors)
+                decoded = self._decode(
+                    spill_conn,
+                    source_conn,
+                    raw_id,
+                    descriptors,
+                    sidecars_by_path=sidecars_by_path,
+                    codex_hook_titles=codex_hook_titles,
+                )
                 if decoded is None:
                     continue
                 sessions, payload_bytes, from_reparse = decoded
@@ -2475,7 +2595,9 @@ def _run_inner(self, generation: int, keys: tuple[str, ...], extra_members: dict
                     self._buffer[raw_id] = (sessions, payload_bytes, tree_bytes, from_reparse, seq)
                     self._buffered_tree_bytes += tree_bytes
         finally:
-            spill_conn.close()
+            if spill_conn is not None:
+                spill_conn.close()
+            source_conn.close()
 
     def _wait_for_budget(self, generation: int, seq: int) -> bool:
         """Block until buffer headroom exists; False means phase over."""
@@ -2565,8 +2687,12 @@ def _build_plan(
     def _decode(
         self,
         spill_conn: sqlite3.Connection,
+        source_conn: sqlite3.Connection,
         raw_id: str,
         descriptors: dict[str, tuple[Provider, str, str, RawRevisionKind, int, str | None]],
+        *,
+        sidecars_by_path: dict[tuple[Origin, str], object],
+        codex_hook_titles: dict[str, str],
     ) -> tuple[list[ParsedSession], int, bool] | None:
         started = time.perf_counter()
         rows = spill_conn.execute(
@@ -2600,6 +2726,14 @@ def _decode(
             # Do not buffer failures: the writer's inline decode raises the
             # identical error at the identical point in the identical order.
             return None
+        sessions_or_none = _replay_safe_enrich_sessions(
+            source_conn,
+            provider=provider,
+            source_path=source_path,
+            sessions=sessions_or_none,
+            sidecars_by_path=sidecars_by_path,
+            codex_hook_titles=codex_hook_titles,
+        )
         self.reparse_hits += 1
         self.decode_seconds += time.perf_counter() - started
         return sessions_or_none, payload_bytes, True
@@ -2718,6 +2852,8 @@ def __init__(self, archive_root: Path, *, max_cached_payload_bytes: int | None)
         #: engaged). ``for_raw`` consults it AFTER the free RAM tiers and
         #: BEFORE the sqlite/reparse fallbacks it exists to hide.
         self._prefetcher: _ReplaySpillPrefetcher | None = None
+        self._replay_sidecars_by_path: dict[tuple[Origin, str], object] = {}
+        self._codex_hook_titles: dict[str, str] | None = None
 
     def attach_prefetcher(self, prefetcher: _ReplaySpillPrefetcher) -> None:
         self._prefetcher = prefetcher
@@ -2845,6 +2981,19 @@ def for_raw(self, archive: ArchiveStore, raw_id: str) -> tuple[list[ParsedSessio
         if rows:
             return [pickle.loads(bytes(row[0])) for row in rows], int(rows[0][1])
         sessions, payload_bytes, _kind = _parse_retained_raw(archive, raw_id)
+        if self._codex_hook_titles is None:
+            from polylogue.sources.assembly_codex import read_codex_thread_title_hook_events
+
+            self._codex_hook_titles = read_codex_thread_title_hook_events(archive._ensure_source_conn())
+        provider, _blob_hash, source_path, _kind, _size = archive.raw_revision_descriptor(raw_id)
+        sessions = _replay_safe_enrich_sessions(
+            archive._ensure_source_conn(),
+            provider=provider,
+            source_path=source_path,
+            sessions=sessions,
+            sidecars_by_path=self._replay_sidecars_by_path,
+            codex_hook_titles=self._codex_hook_titles,
+        )
         self.add(raw_id, sessions, payload_bytes=payload_bytes)
         return sessions, payload_bytes
 
diff --git a/polylogue/storage/raw_byte_duplicate_supersession.py b/polylogue/storage/raw_byte_duplicate_supersession.py
index 3e4be236a8..f6c7713977 100644
--- a/polylogue/storage/raw_byte_duplicate_supersession.py
+++ b/polylogue/storage/raw_byte_duplicate_supersession.py
@@ -114,7 +114,7 @@ def plan_byte_duplicate_supersession(
         if raw_ids is not None and limit is not None:
             raise ValueError("raw_ids cannot be combined with limit")
         query = """
-            SELECT raw_id, blob_hash, blob_size
+            SELECT raw_id, blob_hash, blob_size, origin, source_path, source_index
             FROM raw_sessions
             WHERE revision_authority = 'quarantined'
               AND logical_source_key IS NULL
@@ -150,21 +150,22 @@ def plan_byte_duplicate_supersession(
         # not here, since a raw_id can legitimately be the "other" raw for
         # every other candidate sharing its hash while still being a
         # candidate in its own right.
-        hash_to_raw_ids: dict[bytes, list[str]] = {}
+        hash_to_raws: dict[bytes, dict[str, sqlite3.Row]] = {}
         for chunk_start in range(0, len(candidate_hashes), 500):
             chunk = candidate_hashes[chunk_start : chunk_start + 500]
             placeholders = ", ".join("?" for _ in chunk)
             rows = source_conn.execute(
-                f"SELECT raw_id, blob_hash FROM raw_sessions WHERE blob_hash IN ({placeholders})",
+                f"SELECT raw_id, blob_hash, origin, source_path, source_index "
+                f"FROM raw_sessions WHERE blob_hash IN ({placeholders})",
                 chunk,
             ).fetchall()
             for row in rows:
-                hash_to_raw_ids.setdefault(bytes(row["blob_hash"]), []).append(str(row["raw_id"]))
+                hash_to_raws.setdefault(bytes(row["blob_hash"]), {})[str(row["raw_id"])] = row
 
         # Which of those raw_ids are actually materialized in index.db --
         # the one live-tier fact this classifier is allowed to read (never
         # write).
-        all_raw_ids = sorted({raw_id for raw_ids in hash_to_raw_ids.values() for raw_id in raw_ids})
+        all_raw_ids = sorted({raw_id for raws in hash_to_raws.values() for raw_id in raws})
         indexed_session_by_raw_id: dict[str, str] = {}
         for chunk_start in range(0, len(all_raw_ids), 500):
             raw_id_chunk = all_raw_ids[chunk_start : chunk_start + 500]
@@ -184,9 +185,18 @@ def plan_byte_duplicate_supersession(
         for row in candidate_rows:
             raw_id = str(row["raw_id"])
             blob_hash = bytes(row["blob_hash"])
-            other_raw_ids_for_hash = [rid for rid in hash_to_raw_ids.get(blob_hash, []) if rid != raw_id]
+            other_raws_for_hash = [
+                other
+                for other_id, other in hash_to_raws.get(blob_hash, {}).items()
+                if other_id != raw_id
+                and other["origin"] == row["origin"]
+                and other["source_path"] == row["source_path"]
+                and other["source_index"] == row["source_index"]
+            ]
             indexed_matches = sorted(
-                other_raw_id for other_raw_id in other_raw_ids_for_hash if other_raw_id in indexed_session_by_raw_id
+                str(other["raw_id"])
+                for other in other_raws_for_hash
+                if str(other["raw_id"]) in indexed_session_by_raw_id
             )
             if not indexed_matches:
                 novel_count += 1
diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py
index 2006366cf4..3841333e58 100644
--- a/polylogue/storage/sqlite/archive_tiers/archive.py
+++ b/polylogue/storage/sqlite/archive_tiers/archive.py
@@ -2681,7 +2681,7 @@ def raw_revision_rebuild_selection(
     ) -> tuple[tuple[tuple[str, int], ...], tuple[str, ...]]:
         return raw_revision_rebuild_selection(self, raw_ids)
 
-    def raw_membership_census_rows(self, raw_ids: Sequence[str] | None = None) -> tuple[tuple[str, int], ...]:
+    def raw_membership_census_rows(self, raw_ids: Sequence[str] | None = None) -> tuple[tuple[str, int, bool], ...]:
         return raw_membership_census_rows(self, raw_ids)
 
     def raw_payload_sizes(self, raw_ids: Sequence[str]) -> dict[str, int]:
diff --git a/polylogue/storage/sqlite/archive_tiers/revision_governance.py b/polylogue/storage/sqlite/archive_tiers/revision_governance.py
index 8ac76b54cb..ba32d52a0d 100644
--- a/polylogue/storage/sqlite/archive_tiers/revision_governance.py
+++ b/polylogue/storage/sqlite/archive_tiers/revision_governance.py
@@ -1709,20 +1709,25 @@ def raw_revision_rebuild_selection(
 
 def raw_membership_census_rows(
     store: RawRevisionGovernanceHost, raw_ids: Sequence[str] | None = None
-) -> tuple[tuple[str, int], ...]:
-    """Return every retained raw whose membership census may affect authority."""
+) -> tuple[tuple[str, int, bool], ...]:
+    """Return retained raws and whether durable evidence says they are non-sessions."""
     conn = store._ensure_source_conn()
+    columns = """
+        r.raw_id,
+        r.source_index,
+        EXISTS(SELECT 1 FROM raw_artifacts AS a WHERE a.raw_id = r.raw_id AND a.parse_as_session = 0)
+    """
     if raw_ids is None:
-        rows = conn.execute("SELECT raw_id, source_index FROM raw_sessions ORDER BY raw_id").fetchall()
+        rows = conn.execute(f"SELECT {columns} FROM raw_sessions AS r ORDER BY r.raw_id").fetchall()
     elif raw_ids:
         placeholders = ",".join("?" for _ in raw_ids)
         rows = conn.execute(
-            f"SELECT raw_id, source_index FROM raw_sessions WHERE raw_id IN ({placeholders}) ORDER BY raw_id",
+            f"SELECT {columns} FROM raw_sessions AS r WHERE r.raw_id IN ({placeholders}) ORDER BY r.raw_id",
             tuple(raw_ids),
         ).fetchall()
     else:
         rows = []
-    return tuple((str(row[0]), int(row[1])) for row in rows)
+    return tuple((str(row[0]), int(row[1]), bool(row[2])) for row in rows)
 
 
 def raw_payload_sizes(store: RawRevisionGovernanceHost, raw_ids: Sequence[str]) -> dict[str, int]:
@@ -1879,6 +1884,11 @@ def record_current_parser_source_census(
         membership_logical_keys=membership_keys,
     )
     typed_non_session = bool(raw[2])
+    if typed_non_session:
+        # Terminal parser evidence and other typed non-session artifacts have
+        # an authoritative empty identity set. Requiring a session logical key
+        # here makes those durable dispositions impossible to freeze.
+        durable_keys = ()
     observed_keys = (
         tuple(
             sorted(
diff --git a/tests/infra/pathology_zoo.py b/tests/infra/pathology_zoo.py
index 077b420e6d..92aaa56085 100644
--- a/tests/infra/pathology_zoo.py
+++ b/tests/infra/pathology_zoo.py
@@ -31,6 +31,7 @@
 from polylogue.schemas.synthetic import SyntheticCorpus
 from polylogue.sources.hooks import drain_hook_event_spool, enqueue_hook_event
 from polylogue.sources.parsers.antigravity import AntigravitySessionSummary, markdown_export_payload
+from polylogue.sources.revision_backfill import backfill_historical_revision_evidence
 
 CLAUDE_VINTAGE_LIVE_PROOF_SESSION_ID = "9ed2056f-b415-4f51-b18e-5265f21a67bf"
 CLAUDE_VINTAGE_LIVE_PROOF_ORIGIN = Origin.CLAUDE_AI_EXPORT.value
@@ -562,6 +563,20 @@ def build_pathology_zoo(archive_root: Path) -> PathologyZoo:
         _codex_records("zoo-cycle-a", ("cycle A", "cycle A revised"), parent="zoo-cycle-b", subagent=True),
     )
     asyncio.run(parse_sources_archive(archive_root, sources, parse_workers=1))
+    with sqlite3.connect(archive_root / "source.db") as connection:
+        claude_vintage_raw_ids = [
+            str(row[0])
+            for row in connection.execute(
+                "SELECT raw_id FROM raw_sessions WHERE source_path LIKE ? ORDER BY source_path",
+                ("%claude-live-proof%",),
+            )
+        ]
+    if len(claude_vintage_raw_ids) != 2:
+        raise RuntimeError("pathology zoo Claude vintage pair did not produce exactly two durable raw rows")
+    # The invariant is about revision governance, so build its green fixture
+    # through the production census/replay route rather than hand-authoring
+    # raw_session_memberships rows that merely resemble its output.
+    backfill_historical_revision_evidence(archive_root, selected_raw_ids=claude_vintage_raw_ids)
     hook_event_path = enqueue_hook_event(
         event_id="zoo-hook-event",
         provider="claude-code",
diff --git a/tests/infra/reindex_campaign.py b/tests/infra/reindex_campaign.py
index 0561d0c641..e181db94cd 100644
--- a/tests/infra/reindex_campaign.py
+++ b/tests/infra/reindex_campaign.py
@@ -22,8 +22,10 @@
 from polylogue.pipeline.services.archive_ingest import parse_sources_archive
 from polylogue.scenarios import CorpusSpec
 from polylogue.schemas.synthetic import SyntheticCorpus
+from polylogue.sources.revision_backfill import backfill_historical_revision_evidence
 from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore
 from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root
+from polylogue.storage.sqlite.archive_tiers.revision_governance import record_current_parser_source_census
 from tests.infra.source_builders import SyntheticAntigravityLanguageServerClient
 from tests.infra.whale_fixtures import WHALE_FIXTURE_DIMENSIONS
 
@@ -490,7 +492,20 @@ def build_reindex_campaign_corpus(root: Path) -> ReindexCampaignCorpus:
             parser_failure_raw_id,
             provider=Provider.CLAUDE_CODE,
             error=ValueError("campaign parser failure"),
+            preserve_existing_failure_evidence=True,
         )
+        with archive._ensure_source_conn():
+            record_current_parser_source_census(archive._ensure_source_conn(), parser_failure_raw_id)
+
+    # Phase-2 source freeze requires byte authority for single-document raws
+    # and membership authority for grouped/provider bundles. Exercise the
+    # complete production remediation route so the fixture cannot substitute
+    # one authority model for the other or edit durable columns directly.
+    with patch(
+        "polylogue.sources.parsers.antigravity.AntigravityLanguageServerClient",
+        SyntheticAntigravityLanguageServerClient,
+    ):
+        backfill_historical_revision_evidence(root, ingest_workers=1)
 
     session_ids = _campaign_session_ids(root)
     states, _timings = DaemonConverger(
diff --git a/tests/infra/workload_artifacts.py b/tests/infra/workload_artifacts.py
index 6d3039e66c..28d4eea68b 100644
--- a/tests/infra/workload_artifacts.py
+++ b/tests/infra/workload_artifacts.py
@@ -37,11 +37,12 @@
 )
 from polylogue.schemas.synthetic import SyntheticCorpus
 from polylogue.schemas.synthetic.models import SyntheticArtifactFacts
+from polylogue.sources.origin_specs import lowering_fingerprint, origin_specs
 from polylogue.storage.archive_readiness import raw_materialization_readiness_snapshot, raw_materialization_ready
 from polylogue.storage.raw_reconciler import inspect_raw_authority_frontier
 from tests.infra.source_builders import SyntheticAntigravityLanguageServerClient
 
-_ARTIFACT_PROTOCOL_VERSION = 1
+_ARTIFACT_PROTOCOL_VERSION = 2
 _CACHE_ROOT = Path("/realm/tmp/polylogue-seeded-artifacts")
 _RECIPE_PATHS = (
     Path("polylogue/schemas/synthetic/build_batch.py"),
@@ -59,6 +60,7 @@ class SeededArchiveKey:
     spec_payload: dict[str, object]
     build_id: str
     recipe_id: str
+    source_semantics_id: str
 
     @property
     def value(self) -> str:
@@ -74,6 +76,7 @@ class SeededArchiveManifest:
     profile_id: str
     build_id: str
     recipe_id: str
+    source_semantics_id: str
     facts: tuple[SyntheticArtifactFacts, ...]
     files: tuple[dict[str, object], ...]
     receipt: dict[str, object]
@@ -183,6 +186,21 @@ def _build_id() -> str:
     return f"git:{result.stdout.strip()}"
 
 
+def _source_semantics_id() -> str:
+    """Bind cached archives to the parser semantics that produced them."""
+
+    payload = {
+        "lowering": lowering_fingerprint(),
+        "parsers": {
+            spec.origin.value: spec.parser_fingerprint()
+            for spec in origin_specs()
+            if spec.parser_paths or spec.stream_parser_path or spec.assembly_paths or spec.assembly_spec_path
+        },
+    }
+    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
+    return f"source-semantics:sha256:{hashlib.sha256(encoded).hexdigest()}"
+
+
 def _archive_build_spec(
     *,
     key: SeededArchiveKey,
@@ -224,6 +242,7 @@ def seeded_archive_key(specs: Iterable[CorpusSpec]) -> SeededArchiveKey:
         spec_payload={"corpus_specs": [spec.to_payload() for spec in specs]},
         build_id=_build_id(),
         recipe_id=_recipe_id(),
+        source_semantics_id=_source_semantics_id(),
     )
 
 
@@ -510,6 +529,7 @@ def build_seeded_archive(
                 profile_id=profile_id,
                 build_id=key.build_id,
                 recipe_id=key.recipe_id,
+                source_semantics_id=key.source_semantics_id,
                 facts=facts,
                 files=_archive_files(staging),
                 receipt=dict(receipt.to_payload()),
diff --git a/tests/unit/api/test_operation_executor_routes.py b/tests/unit/api/test_operation_executor_routes.py
index 9dd375a4f1..4a5358b6da 100644
--- a/tests/unit/api/test_operation_executor_routes.py
+++ b/tests/unit/api/test_operation_executor_routes.py
@@ -14,15 +14,13 @@
 
 from polylogue.api import Polylogue
 from polylogue.operations.mutation_transaction import OperationExecutor
-from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database
-from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
+from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root
 
 
 def _seed_archive(archive_root: Path, *, native_id: str) -> str:
     source_db = archive_root / "source.db"
     index_db = archive_root / "index.db"
-    initialize_archive_database(source_db, ArchiveTier.SOURCE)
-    initialize_archive_database(index_db, ArchiveTier.INDEX)
+    initialize_active_archive_root(archive_root)
     raw_id = f"raw-{native_id}"
     session_id = f"codex-session:{native_id}"
     with sqlite3.connect(source_db) as conn:
diff --git a/tests/unit/cli/test_daemon_client.py b/tests/unit/cli/test_daemon_client.py
index e3ff95554a..36a25e1f8e 100644
--- a/tests/unit/cli/test_daemon_client.py
+++ b/tests/unit/cli/test_daemon_client.py
@@ -1,14 +1,27 @@
 from __future__ import annotations
 
+import shutil
 import subprocess
 import sys
+import tempfile
 import threading
+from collections.abc import Iterator
 from os import getpid
 from pathlib import Path
 
 import pytest
 
 
+@pytest.fixture
+def _short_uds_runtime_dir() -> Iterator[Path]:
+    """Keep UDS route tests under the operating system socket-path limit."""
+    runtime_dir = Path(tempfile.mkdtemp(prefix="plg-client-uds-"))
+    try:
+        yield runtime_dir
+    finally:
+        shutil.rmtree(runtime_dir, ignore_errors=True)
+
+
 def test_daemon_client_import_does_not_load_storage() -> None:
     result = subprocess.run(
         [
@@ -70,7 +83,9 @@ def test_daemon_probe_rejects_the_tmp_archive_config_trap(monkeypatch: pytest.Mo
     )
 
 
-def test_daemon_client_probes_the_production_uds_server(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_daemon_client_probes_the_production_uds_server(
+    monkeypatch: pytest.MonkeyPatch, _short_uds_runtime_dir: Path
+) -> None:
     """The stdlib client reaches the production AF_UNIX server, not a TCP substitute."""
 
     from http import HTTPStatus
@@ -92,7 +107,7 @@ def health(self: DaemonAPIHandler) -> None:
         )
 
     monkeypatch.setattr(DaemonAPIHandler, "_handle_health", health)
-    socket_path = Path("/realm/tmp") / f"polylogue-uds-{getpid()}.sock"
+    socket_path = _short_uds_runtime_dir / f"daemon-{getpid()}.sock"
     server = DaemonAPIUnixHTTPServer(socket_path, DaemonAPIHandler)
     server.auth_token = "uds-test-token"
     thread = threading.Thread(target=server.serve_forever, daemon=True)
@@ -113,7 +128,107 @@ def health(self: DaemonAPIHandler) -> None:
         thread.join(timeout=2)
 
 
-def test_daemon_client_preserves_typed_4xx_detail_from_the_production_uds_server() -> None:
+def test_daemon_client_can_probe_matching_writer_through_degraded_health(
+    monkeypatch: pytest.MonkeyPatch,
+    _short_uds_runtime_dir: Path,
+) -> None:
+    """Maintenance discovers the writer without weakening query readiness."""
+    from http import HTTPStatus
+
+    from polylogue.cli.daemon_client import DaemonClient
+    from polylogue.daemon.http import DaemonAPIHandler
+    from polylogue.daemon.uds import DaemonAPIUnixHTTPServer
+
+    def degraded_health(self: DaemonAPIHandler) -> None:
+        self._send_json(
+            HTTPStatus.SERVICE_UNAVAILABLE,
+            {
+                "archive_root": "/realm/archive",
+                "index_schema_version": 24,
+                "daemon_version": "0.1.0",
+                "raw_failure_lifecycle_state": "degraded",
+            },
+        )
+
+    monkeypatch.setattr(DaemonAPIHandler, "_handle_health", degraded_health)
+    socket_path = _short_uds_runtime_dir / f"degraded-{getpid()}.sock"
+    server = DaemonAPIUnixHTTPServer(socket_path, DaemonAPIHandler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        client = DaemonClient(socket_path)
+        assert (
+            client.probe(
+                archive_root="/realm/archive",
+                index_schema_version=24,
+                daemon_version="0.1.0",
+            )
+            is None
+        )
+        assert (
+            client.probe(
+                archive_root="/realm/archive",
+                index_schema_version=24,
+                daemon_version="0.1.0",
+                accept_degraded=True,
+            )
+            is not None
+        )
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=2)
+
+
+def test_daemon_client_rejects_unrelated_503_when_degraded_probe_is_enabled(
+    monkeypatch: pytest.MonkeyPatch,
+    _short_uds_runtime_dir: Path,
+) -> None:
+    """The production probe route accepts a maintenance ``degraded`` 503,
+    not any matching identity payload. Mutating that lifecycle state to
+    ``blocked`` must keep the writer unavailable to the caller."""
+    from http import HTTPStatus
+
+    from polylogue.cli.daemon_client import DaemonClient
+    from polylogue.daemon.http import DaemonAPIHandler
+    from polylogue.daemon.uds import DaemonAPIUnixHTTPServer
+
+    def blocked_health(self: DaemonAPIHandler) -> None:
+        self._send_json(
+            HTTPStatus.SERVICE_UNAVAILABLE,
+            {
+                "archive_root": "/realm/archive",
+                "index_schema_version": 24,
+                "daemon_version": "0.1.0",
+                "raw_failure_lifecycle_state": "blocked",
+            },
+        )
+
+    monkeypatch.setattr(DaemonAPIHandler, "_handle_health", blocked_health)
+    socket_path = _short_uds_runtime_dir / f"blocked-{getpid()}.sock"
+    server = DaemonAPIUnixHTTPServer(socket_path, DaemonAPIHandler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        client = DaemonClient(socket_path)
+        assert (
+            client.probe(
+                archive_root="/realm/archive",
+                index_schema_version=24,
+                daemon_version="0.1.0",
+                accept_degraded=True,
+            )
+            is None
+        )
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=2)
+
+
+def test_daemon_client_preserves_typed_4xx_detail_from_the_production_uds_server(
+    _short_uds_runtime_dir: Path,
+) -> None:
     """Maintenance clients can surface a daemon validation reason, not only a transport failure."""
     from http import HTTPStatus
 
@@ -129,7 +244,7 @@ def _handle_consume_canary_report(self) -> None:
                 "receipt is missing the canonical acceptance profile",
             )
 
-    socket_path = Path("/realm/tmp") / f"polylogue-uds-canary-4xx-{getpid()}.sock"
+    socket_path = _short_uds_runtime_dir / f"canary-4xx-{getpid()}.sock"
     server = DaemonAPIUnixHTTPServer(socket_path, InvalidCanaryReportHandler)
     server.auth_token = "uds-test-token"
     thread = threading.Thread(target=server.serve_forever, daemon=True)
diff --git a/tests/unit/core/test_schema_generation.py b/tests/unit/core/test_schema_generation.py
index 624014c656..98ea6cc1d2 100644
--- a/tests/unit/core/test_schema_generation.py
+++ b/tests/unit/core/test_schema_generation.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import sqlite3
 from collections.abc import Generator
 from pathlib import Path
 from types import SimpleNamespace
@@ -240,7 +241,7 @@ class TestLogicalHeadsOnly:
 
     polylogue-t0m73 phase 1: an opt-in flag restricting the sampling query
     to one row per logical source (latest revision per
-    ``(origin, COALESCE(native_id, source_path))``) for value-distribution
+    ``(origin, COALESCE(logical_source_key, native_id, source_path))``) for value-distribution
     callers, who would otherwise have a re-acquired session contribute its
     field values once per revision. Default is unchanged (every revision
     sampled) for schema-shape discovery, which wants every revision since
@@ -333,6 +334,41 @@ def _record(*, raw_id: str, status: object, artifact_kind: object, source_path:
 
         assert seen_raw_ids == ["raw-new"]  # the later of the two revisions is the logical head
 
+    def test_logical_heads_only_canonicalizes_retained_membership_aliases(self, tmp_path: Path) -> None:
+        from polylogue.core.enums import Provider
+        from polylogue.schemas.observation_identity import resolve_provider_config
+        from polylogue.schemas.sampling_db import _iter_schema_units_from_db
+
+        self._seed_two_revisions(tmp_path / "source.db")
+        with sqlite3.connect(tmp_path / "source.db") as conn:
+            conn.execute("UPDATE raw_sessions SET logical_source_key = 'chatgpt:conv-1' WHERE raw_id = 'raw-new'")
+            conn.execute(
+                """
+                INSERT INTO raw_session_memberships(
+                    raw_id, logical_source_key, provider_session_id, source_revision,
+                    normalized_content_hash, message_count
+                ) VALUES ('raw-old', 'chatgpt-export:conv-1', 'conv-1', 'old', ?, 1)
+                """,
+                (b"m" * 32,),
+            )
+
+        seen_raw_ids: list[str] = []
+
+        def _record(*, raw_id: str, status: object, artifact_kind: object, source_path: object, reason: object) -> None:
+            seen_raw_ids.append(raw_id)
+
+        list(
+            _iter_schema_units_from_db(
+                Provider.CHATGPT,
+                db_path=tmp_path / "index.db",
+                config=resolve_provider_config(Provider.CHATGPT),
+                terminal_recorder=_record,
+                logical_heads_only=True,
+            )
+        )
+
+        assert seen_raw_ids == ["raw-new"]
+
 
 class TestGenerateSchemaFromSamples:
     """Focused schema-generation edge cases beyond the general laws."""
diff --git a/tests/unit/daemon/test_bulk_rebuild_ownership.py b/tests/unit/daemon/test_bulk_rebuild_ownership.py
index db09a35232..5814475938 100644
--- a/tests/unit/daemon/test_bulk_rebuild_ownership.py
+++ b/tests/unit/daemon/test_bulk_rebuild_ownership.py
@@ -23,6 +23,7 @@
 
 from polylogue.daemon.bulk_rebuild import resolve_or_start_daemon_bulk_rebuild_transaction
 from polylogue.maintenance.rebuild_index import RebuildSchemaCurrencyError
+from polylogue.maintenance.schema_inference_gate import run_schema_inference_gate
 from polylogue.storage.archive_identity import (
     ArchiveLocation,
     ArchiveOwnershipError,
@@ -30,16 +31,23 @@
     assert_owns_archive_location,
 )
 from polylogue.storage.archive_readiness import probe_archive_tier
-from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
-from polylogue.storage.sqlite.migration_runner import DURABLE_MIGRATION_TIERS
-from tests.infra.rebuild_receipt import write_valid_rebuild_receipt
+from tests.infra.schema_inference import seed_schema_inference_archive
 
 
-def _init_empty_source(root: Path) -> None:
-    root.mkdir(parents=True, exist_ok=True)
-    for tier in sorted(DURABLE_MIGRATION_TIERS, key=lambda item: item.value):
-        initialize_archive_database(root / f"{tier.value}.db", tier)
+def _init_empty_source(root: Path) -> Path:
+    return seed_schema_inference_archive(root)
+
+
+def _schema_inference_receipt(root: Path, ground_truth: Path, tmp_path: Path) -> Path:
+    receipt = tmp_path / f"{root.name}-schema-inference-gate-receipt.json"
+    result = run_schema_inference_gate(
+        root,
+        receipt_path=receipt,
+        ground_truth_roots={"codex-session": (ground_truth,)},
+    )
+    assert result.passed, result.payload["pass_fail_reasons"]
+    return receipt
 
 
 def test_daemon_bulk_rebuild_rejects_schema_mismatch_before_transaction_bookkeeping(tmp_path: Path) -> None:
@@ -71,8 +79,8 @@ def test_daemon_bulk_rebuild_rechecks_schema_currency_after_ownership(
     from polylogue.daemon import bulk_rebuild
 
     root = tmp_path / "archive"
-    _init_empty_source(root)
-    receipt = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-gate-receipt.json")
+    ground_truth = _init_empty_source(root)
+    receipt = _schema_inference_receipt(root, ground_truth, tmp_path)
     real_assert = assert_owns_archive_location
 
     def mutate_audit_after_ownership(owned: OwnedArchiveLocation, location: ArchiveLocation) -> None:
@@ -92,7 +100,9 @@ def mutate_audit_after_ownership(owned: OwnedArchiveLocation, location: ArchiveL
     assert not (root / ".index-rebuild-transactions").exists()
 
 
-def test_daemon_bulk_rebuild_refuses_when_archive_location_already_owned(tmp_path: Path) -> None:
+def test_daemon_bulk_rebuild_refuses_when_archive_location_already_owned(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
     """A concurrent holder of the archive-location ownership lock must block
     the daemon's bulk-rebuild transaction resolve/retire path before any
     generation directory or transaction record is created -- mirroring
@@ -100,12 +110,19 @@ def test_daemon_bulk_rebuild_refuses_when_archive_location_already_owned(tmp_pat
     offline rebuild entry point.
     """
     root = tmp_path / "archive"
-    _init_empty_source(root)
+    ground_truth = _init_empty_source(root)
+    receipt = _schema_inference_receipt(root, ground_truth, tmp_path)
+    # This test proves ownership around transaction bookkeeping. The schema
+    # gate above is real; the source-admission route is separately covered
+    # and the helper's tiny gate corpus deliberately has no replay census.
+    from polylogue.daemon import bulk_rebuild
+
+    monkeypatch.setattr(bulk_rebuild, "validate_rebuild_source_admission", lambda *_args: None)
     location = ArchiveLocation.resolve(root)
     owned = OwnedArchiveLocation.acquire(location, owner_id="concurrent-campaign")
     try:
         with pytest.raises(ArchiveOwnershipError):
-            resolve_or_start_daemon_bulk_rebuild_transaction(root)
+            resolve_or_start_daemon_bulk_rebuild_transaction(root, schema_inference_receipt_path=receipt)
         # Failure happened before any generation/transaction bookkeeping was created.
         assert not (root / ".index-generations").exists()
         assert not (root / ".index-rebuild-transactions").exists()
@@ -113,19 +130,25 @@ def test_daemon_bulk_rebuild_refuses_when_archive_location_already_owned(tmp_pat
         owned.release()
 
     # Releasing the concurrent holder's ownership lets the daemon proceed.
-    transaction = resolve_or_start_daemon_bulk_rebuild_transaction(root)
+    transaction = resolve_or_start_daemon_bulk_rebuild_transaction(root, schema_inference_receipt_path=receipt)
     assert transaction.status == "running"
 
 
-def test_daemon_bulk_rebuild_releases_ownership_lock_after_resolving(tmp_path: Path) -> None:
+def test_daemon_bulk_rebuild_releases_ownership_lock_after_resolving(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
     """The ownership lock must not be left held after resolving/starting a
     transaction, so a subsequent maintenance/campaign writer can still
     acquire it.
     """
     root = tmp_path / "archive"
-    _init_empty_source(root)
+    ground_truth = _init_empty_source(root)
+    receipt = _schema_inference_receipt(root, ground_truth, tmp_path)
+    from polylogue.daemon import bulk_rebuild
+
+    monkeypatch.setattr(bulk_rebuild, "validate_rebuild_source_admission", lambda *_args: None)
 
-    resolve_or_start_daemon_bulk_rebuild_transaction(root)
+    resolve_or_start_daemon_bulk_rebuild_transaction(root, schema_inference_receipt_path=receipt)
 
     location = ArchiveLocation.resolve(root)
     owned = OwnedArchiveLocation.acquire(location, owner_id="post-resolve-probe")
diff --git a/tests/unit/devtools/test_index_fast_forward.py b/tests/unit/devtools/test_index_fast_forward.py
index c4cf498a9b..c5aa6d5afc 100644
--- a/tests/unit/devtools/test_index_fast_forward.py
+++ b/tests/unit/devtools/test_index_fast_forward.py
@@ -18,20 +18,18 @@
 from polylogue.storage.blob_store import BlobStore
 from polylogue.storage.index_generation import IndexGeneration, IndexGenerationStore
 from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore
-from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database
+from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
 
 
 def _archive(tmp_path: Path, *, extra_native_ids: tuple[str, ...] = ()) -> Path:
     root = tmp_path / "archive"
-    root.mkdir()
-    for tier in (ArchiveTier.SOURCE, ArchiveTier.USER, ArchiveTier.EMBEDDINGS, ArchiveTier.OPS):
-        initialize_archive_database(root / f"{tier.value}.db", tier)
+    initialize_active_archive_root(root)
     storage = tmp_path / "storage"
     active_root = storage / ".index-generations" / "v36"
     active_root.mkdir(parents=True)
     active = active_root / "index.db"
-    initialize_archive_database(active, ArchiveTier.INDEX)
+    (root / "index.db").replace(active)
     (storage / "index.db").symlink_to(active)
     (root / "index.db").symlink_to(storage / "index.db")
 
diff --git a/tests/unit/devtools/test_run_tests.py b/tests/unit/devtools/test_run_tests.py
index bef2f53356..0d126bd449 100644
--- a/tests/unit/devtools/test_run_tests.py
+++ b/tests/unit/devtools/test_run_tests.py
@@ -34,19 +34,26 @@ def test_build_pytest_cmd_respects_explicit_worker_flag() -> None:
 
 
 @pytest.mark.parametrize(
-    "selection",
+    ("selection", "expected_request"),
     [
-        ["tests/unit", "-n4"],
-        ["tests/unit", "-n=4"],
-        ["tests/unit", "--numprocesses", "8"],
-        ["tests/unit", "--numprocesses=8"],
+        (["tests/unit", "-n4"], "4"),
+        (["tests/unit", "-n=4"], "4"),
+        (["tests/unit", "--numprocesses", "8"], "8"),
+        (["tests/unit", "--numprocesses=8"], "8"),
     ],
 )
-def test_build_pytest_cmd_forwards_all_xdist_worker_spellings(selection: list[str]) -> None:
+def test_build_pytest_cmd_forwards_exactly_one_xdist_worker_request(
+    selection: list[str], expected_request: str
+) -> None:
     command = run_tests.build_pytest_cmd(selection)
 
     for arg in selection:
         assert arg in command
+    worker_flags = [
+        arg for arg in command if arg in {"-n", "--numprocesses"} or arg.startswith(("-n", "--numprocesses="))
+    ]
+    assert len(worker_flags) == 1
+    assert verify._pytest_command_worker_request(command) == expected_request
 
 
 def test_build_pytest_cmd_honors_workers_env(monkeypatch: pytest.MonkeyPatch) -> None:
diff --git a/tests/unit/devtools/test_verify.py b/tests/unit/devtools/test_verify.py
index dedf0d44d1..2d1208e15a 100644
--- a/tests/unit/devtools/test_verify.py
+++ b/tests/unit/devtools/test_verify.py
@@ -645,8 +645,17 @@ def test_seed_collection_refuses_parallel_worker_overrides(monkeypatch: pytest.M
     assert command[command.index("-n") + 1] == "0"
 
 
-def test_seed_defaults_to_managed_scratch(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.delenv("POLYLOGUE_PYTEST_TMPFS", raising=False)
+def test_seed_defaults_to_managed_scratch(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    shm, scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True)
+    _patch_resource_capacity(monkeypatch, shm=shm, scratch=scratch, available_mb=8192)
+    for name in (
+        "POLYLOGUE_PYTEST_BASETEMP_ROOT",
+        "POLYLOGUE_PYTEST_TMPFS",
+        "POLYLOGUE_PYTEST_TMPFS_MAX_MB",
+        "POLYLOGUE_PYTEST_BASETEMP_REQUIRED_MB",
+        "POLYLOGUE_PYTEST_BASETEMP_MIN_FREE_MB",
+    ):
+        monkeypatch.delenv(name, raising=False)
     completed = subprocess.CompletedProcess(args=["pytest"], returncode=0, stdout="1 passed in 0.1s\n", stderr="")
 
     with (
@@ -658,9 +667,7 @@ def test_seed_defaults_to_managed_scratch(monkeypatch: pytest.MonkeyPatch) -> No
     assert rc == 0
     assert metadata["pytest_tmpfs"] is False
     assert run.call_args.kwargs["env"]["POLYLOGUE_PYTEST_TMPFS"] == "0"
-    assert run.call_args.kwargs["env"]["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(
-        verify_runs.DEFAULT_PYTEST_BASETEMP_ROOT
-    )
+    assert run.call_args.kwargs["env"]["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(scratch)
     assert run.call_args.kwargs["env"]["POLYLOGUE_PYTEST_SELECTION_NODEID_LIMIT"] == "50000"
 
 
@@ -1902,13 +1909,11 @@ def counted_size(_path: Path) -> int:
     assert calls == 1
 
 
-def test_pytest_basetemp_path_tracks_tmpfs_opt_in(tmp_path: Path) -> None:
+def test_pytest_basetemp_path_tracks_tmpfs_opt_in(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    shm, _scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True)
     path = pytest_basetemp_path(root=tmp_path, run_id="run-1", env={"POLYLOGUE_PYTEST_TMPFS": "1"})
 
-    if Path("/dev/shm").is_dir():
-        assert path.parent == Path("/dev/shm")
-    else:
-        assert path.parent == Path("/realm/tmp/polylogue-pytest")
+    assert path.parent == shm
 
 
 def test_pytest_tmpfs_budget_is_shared_and_bounded() -> None:
@@ -2063,7 +2068,10 @@ def test_adaptive_pytest_policy_treats_full_run_basetemp_as_aggregate_demand() -
         (["--numprocesses=auto"], max(1, os.cpu_count() or 1)),
     ],
 )
-def test_production_pytest_commands_reserve_every_xdist_spelling(worker_args: list[str], expected: int) -> None:
+def test_production_pytest_commands_reserve_every_xdist_spelling(
+    monkeypatch: pytest.MonkeyPatch, worker_args: list[str], expected: int
+) -> None:
+    monkeypatch.delenv("PYTEST_XDIST_AUTO_NUM_WORKERS", raising=False)
     command = run_tests.build_pytest_cmd(["tests/unit/devtools", *worker_args])
 
     assert verify._pytest_command_concurrency(command) == expected
@@ -2311,6 +2319,28 @@ def test_inherited_512_mib_tmpfs_cap_reroutes_measured_demand_to_scratch(
     assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(scratch)
 
 
+def test_explicit_tmpfs_root_reroutes_to_scratch_when_its_cap_is_too_small(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    shm, scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True)
+    _patch_resource_capacity(monkeypatch, shm=shm, scratch=scratch, available_mb=15_190)
+    explicit_root = shm / "explicit"
+    explicit_root.mkdir()
+
+    env, policy = apply_managed_pytest_runtime_policy(
+        {
+            "POLYLOGUE_PYTEST_BASETEMP_ROOT": str(explicit_root),
+            "POLYLOGUE_PYTEST_TMPFS_MAX_MB": "512",
+        },
+        worker_count=4,
+    )
+
+    assert policy is not None
+    assert policy.basetemp_label == "scratch"
+    assert env["POLYLOGUE_PYTEST_TMPFS"] == "0"
+    assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(scratch)
+
+
 def test_focused_policy_keeps_full_suite_basetemp_demand_out_of_scratch_preflight(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Path
 ) -> None:
@@ -2679,7 +2709,14 @@ def test_explicit_basetemp_root_retains_managed_resource_monitoring(
 ) -> None:
     monkeypatch.setattr(verify_runs, "PYTEST_TMPFS_ROOT", tmp_path / "unselected-tmpfs")
     nvme_root = tmp_path / "realm-tmp" / "polylogue-pytest"
+    nvme_root.mkdir(parents=True)
+    monkeypatch.setattr(verify_runs, "_meminfo", lambda: {"MemAvailable": 8 * 1024 * 1024})
+    monkeypatch.setattr(verify_runs, "read_cgroup_memory_headroom_bytes", lambda: None)
+    monkeypatch.setattr(verify_runs, "_pressure", lambda _kind: {"full_avg10": 0.0})
+    monkeypatch.setattr(verify_runs, "_fs_usage", lambda _path: {"used_kb": 0, "free_kb": 16 * 1024 * 1024})
     monkeypatch.setenv("POLYLOGUE_PYTEST_BASETEMP_ROOT", str(nvme_root))
+    monkeypatch.delenv("POLYLOGUE_PYTEST_BASETEMP_REQUIRED_MB", raising=False)
+    monkeypatch.delenv("POLYLOGUE_PYTEST_BASETEMP_MIN_FREE_MB", raising=False)
     run = VerifyRun(tier="configured-nvme", argv=[], git_head=None, root=tmp_path)
 
     rc, _elapsed, metadata = _run(
@@ -2726,6 +2763,7 @@ def to_dict(self) -> dict[str, int]:
         ("pytest testmon", False),
         ("pytest testmon (broad)", True),
         ("pytest seed-testmon", True),
+        ("pytest seed-testmon shard 1/4", True),
         ("pytest full (parallel)", True),
         ("pytest load-sensitive (isolated)", True),
     ],
@@ -2758,11 +2796,15 @@ def test_bench_slo_forces_nested_pytest_to_managed_scratch(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
+    shm, scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True)
+    inherited_tmpfs_root = shm / "inherited-benchmark"
+    inherited_tmpfs_root.mkdir()
+    scratch.mkdir()
     run = VerifyRun(tier="lab", argv=[], git_head=None, root=tmp_path)
-    monkeypatch.setenv("POLYLOGUE_PYTEST_BASETEMP_ROOT", "/dev/shm/inherited-benchmark")
+    monkeypatch.setenv("POLYLOGUE_PYTEST_BASETEMP_ROOT", str(inherited_tmpfs_root))
     managed_env = {
         "POLYLOGUE_PYTEST_TMPFS": "0",
-        "POLYLOGUE_PYTEST_BASETEMP_ROOT": "/realm/tmp/polylogue-pytest",
+        "POLYLOGUE_PYTEST_BASETEMP_ROOT": str(scratch),
     }
     completed = subprocess.CompletedProcess(args=["devtools", "bench", "slo"], returncode=0, stdout="", stderr="")
 
@@ -2781,7 +2823,7 @@ def test_bench_slo_forces_nested_pytest_to_managed_scratch(
     assert env["POLYLOGUE_VERIFY_RUN_ID"] == run.run_id
     assert env["POLYLOGUE_PYTEST_RUN_ID"] == run.run_id
     assert env["POLYLOGUE_PYTEST_TMPFS"] == "0"
-    assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == "/realm/tmp/polylogue-pytest"
+    assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(scratch)
 
 
 def test_run_forces_subprocesses_to_current_checkout(monkeypatch: pytest.MonkeyPatch) -> None:
diff --git a/tests/unit/infra/test_workload_artifacts.py b/tests/unit/infra/test_workload_artifacts.py
index a0bb2de1e6..1a29db27e0 100644
--- a/tests/unit/infra/test_workload_artifacts.py
+++ b/tests/unit/infra/test_workload_artifacts.py
@@ -19,6 +19,7 @@
     _journal_mode_delete_with_retry,
     build_seeded_archive,
     clone_seeded_archive,
+    seeded_archive_key,
 )
 
 
@@ -41,6 +42,17 @@ def test_seeded_archive_publishes_valid_immutable_real_pipeline_artifact(tmp_pat
     assert not (first.root.stat().st_mode & os.W_OK)
 
 
+def test_seeded_archive_key_changes_with_source_semantics(monkeypatch: pytest.MonkeyPatch) -> None:
+    import tests.infra.workload_artifacts as artifacts
+
+    monkeypatch.setattr(artifacts, "lowering_fingerprint", lambda: "emitter-semantics:first")
+    first = seeded_archive_key(())
+    monkeypatch.setattr(artifacts, "lowering_fingerprint", lambda: "emitter-semantics:second")
+    second = seeded_archive_key(())
+
+    assert first.value != second.value
+
+
 def test_seeded_archive_clone_is_private_full_root_and_preserves_base(tmp_path: Path) -> None:
     artifact = build_seeded_archive(cache_root=tmp_path / "cache")
     base_manifest = artifact.root.joinpath("manifest.json").read_bytes()
diff --git a/tests/unit/maintenance/test_archive_verification.py b/tests/unit/maintenance/test_archive_verification.py
index 4452c2f914..c1b7cdaec2 100644
--- a/tests/unit/maintenance/test_archive_verification.py
+++ b/tests/unit/maintenance/test_archive_verification.py
@@ -35,7 +35,6 @@
     validate_archive_verification_registry,
     verify_archive,
 )
-from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync
 from polylogue.sources.origin_specs import lowering_fingerprint, parser_fingerprint_for_origin
 from polylogue.storage.blob_store import BlobStore
 from polylogue.storage.sqlite.archive_tiers.bootstrap import ARCHIVE_TIER_SPECS, initialize_active_archive_root
@@ -299,10 +298,33 @@ def test_raw_failure_lifecycle_accepts_only_typed_deferred_or_terminal_evidence(
 
     typed = verify_archive(tmp_path, checks=("raw-failure-lifecycle",))
     typed_check = _check(typed, "raw-failure-lifecycle")
-    assert typed_check.status is OutcomeStatus.WARNING
+    assert typed_check.status is OutcomeStatus.OK
     assert not typed.blocking
     assert typed_check.evidence["terminal"] == 1
 
+    with sqlite3.connect(tmp_path / "source.db") as conn:
+        upsert_raw_artifact(
+            conn,
+            "raw-1",
+            ArchiveSourceArtifact(
+                artifact_id="failure-evidence-deferred",
+                origin=Origin.CODEX_SESSION,
+                source_path="/x",
+                source_index=0,
+                artifact_kind="deferred_hot_jsonl_capture",
+                classification_reason="deferred_hot_jsonl_capture",
+                support_status=ArtifactSupportStatus.PARTIAL_DECODE,
+                first_observed_at_ms=200,
+                last_observed_at_ms=200,
+            ),
+        )
+        conn.commit()
+
+    deferred = verify_archive(tmp_path, checks=("raw-failure-lifecycle",))
+    deferred_check = _check(deferred, "raw-failure-lifecycle")
+    assert deferred_check.status is OutcomeStatus.WARNING
+    assert deferred_check.evidence["deferred"] == 1
+
     with sqlite3.connect(tmp_path / "source.db") as conn:
         conn.execute("DELETE FROM raw_artifacts WHERE raw_id = 'raw-1'")
         conn.commit()
@@ -517,6 +539,119 @@ def test_source_index_coverage_census_deletion_does_not_hide_raw_head(tmp_path:
     assert spec.red_twin.test_name == "test_source_index_coverage_census_deletion_does_not_hide_raw_head"
 
 
+def test_source_index_coverage_groups_reacquisitions_by_logical_source_key(tmp_path: Path) -> None:
+    """Different capture paths for one typed source are revisions, not gaps."""
+
+    _seed_coherent_archive(tmp_path)
+    source_conn = _connect(tmp_path / "source.db")
+    try:
+        source_conn.execute(
+            "UPDATE raw_sessions SET logical_source_key = 'codex-session:session' WHERE raw_id = 'raw-1'"
+        )
+        source_conn.execute(
+            """
+            INSERT INTO raw_sessions(
+                raw_id, origin, native_id, source_path, blob_hash, blob_size,
+                acquired_at_ms, logical_source_key, revision_authority
+            )
+            SELECT 'raw-reacquired', origin, NULL, '/new-capture-path',
+                   blob_hash, blob_size, 200, 'codex:session', 'byte_proven'
+            FROM raw_sessions WHERE raw_id = 'raw-1'
+            """
+        )
+        source_conn.commit()
+    finally:
+        source_conn.close()
+
+    check = _check(verify_archive(tmp_path, checks=("source-index-coverage",)), "source-index-coverage")
+
+    assert check.status is OutcomeStatus.OK
+    assert check.evidence["logical_head_count"] == 1
+    assert check.evidence["unindexed_head_count"] == 0
+
+
+def test_coverage_groups_retired_membership_identity_with_bound_revision(tmp_path: Path) -> None:
+    """A retired raw keeps its sole membership identity after its raw key is nulled."""
+    _seed_coherent_archive(tmp_path)
+    source_conn = _connect(tmp_path / "source.db")
+    try:
+        source_conn.execute(
+            "UPDATE raw_sessions SET logical_source_key = 'codex-session:session' WHERE raw_id = 'raw-1'"
+        )
+        source_conn.execute(
+            """
+            INSERT INTO raw_sessions(
+                raw_id, origin, native_id, source_path, blob_hash, blob_size,
+                acquired_at_ms, revision_authority
+            )
+            SELECT 'raw-retired-membership', origin, 'session', '/retired-membership',
+                   blob_hash, blob_size, 200, 'quarantined'
+            FROM raw_sessions WHERE raw_id = 'raw-1'
+            """
+        )
+        source_conn.execute(
+            """
+            INSERT INTO raw_session_memberships(
+                raw_id, logical_source_key, provider_session_id, source_revision,
+                normalized_content_hash, message_count
+            ) VALUES ('raw-retired-membership', 'codex:session', 'session', 'retired', ?, 1)
+            """,
+            (b"r" * 32,),
+        )
+        source_conn.commit()
+    finally:
+        source_conn.close()
+
+    report = verify_archive(tmp_path, checks=("source-index-coverage", "convergence-freshness"))
+
+    coverage = _check(report, "source-index-coverage")
+    freshness = _check(report, "convergence-freshness")
+    assert coverage.status is OutcomeStatus.OK
+    assert coverage.evidence["logical_head_count"] == 1
+    assert freshness.status is OutcomeStatus.OK
+    assert freshness.evidence["unindexed_backlog_gap"] == 0
+
+
+def test_coverage_does_not_assign_a_shared_raw_to_one_membership_key(tmp_path: Path) -> None:
+    """A raw with several retained identities must not inherit an arbitrary cohort."""
+    _seed_coherent_archive(tmp_path)
+    source_conn = _connect(tmp_path / "source.db")
+    try:
+        source_conn.execute("UPDATE raw_sessions SET logical_source_key = 'codex:session' WHERE raw_id = 'raw-1'")
+        source_conn.execute(
+            """
+            INSERT INTO raw_sessions(
+                raw_id, origin, native_id, source_path, blob_hash, blob_size,
+                acquired_at_ms, revision_authority
+            )
+            SELECT 'raw-shared-membership', origin, NULL, '/shared-membership',
+                   blob_hash, blob_size, 200, 'quarantined'
+            FROM raw_sessions WHERE raw_id = 'raw-1'
+            """
+        )
+        source_conn.executemany(
+            """
+            INSERT INTO raw_session_memberships(
+                raw_id, logical_source_key, provider_session_id, source_revision,
+                normalized_content_hash, message_count
+            ) VALUES ('raw-shared-membership', ?, ?, 'shared', ?, 1)
+            """,
+            [
+                ("codex:session", "session", b"s" * 32),
+                ("codex:other", "other", b"o" * 32),
+            ],
+        )
+        source_conn.commit()
+    finally:
+        source_conn.close()
+
+    check = _check(verify_archive(tmp_path, checks=("source-index-coverage",)), "source-index-coverage")
+
+    assert check.status is OutcomeStatus.WARNING
+    assert check.evidence["logical_head_count"] == 2
+    assert check.evidence["quarantined_count"] == 1
+
+
 def test_quarantined_head_with_no_session_is_warning_not_error(tmp_path: Path) -> None:
     """The polylogue-in24n bug class in reverse: a quarantined raw (the
     default ``revision_authority``) that never materialized is a *typed*
@@ -578,6 +713,124 @@ def test_parse_error_head_with_no_session_is_ok(tmp_path: Path) -> None:
     assert check.evidence["quarantined_count"] == 0
 
 
+def test_valid_byte_supersession_receipt_covers_unindexed_head(tmp_path: Path) -> None:
+    """A receipt is authority only when its bytes and indexed twin revalidate."""
+
+    _seed_coherent_archive(tmp_path)
+    source_conn = _connect(tmp_path / "source.db")
+    try:
+        source_conn.execute(
+            """
+            INSERT INTO raw_sessions(
+                raw_id, origin, native_id, source_path, source_index, blob_hash, blob_size,
+                acquired_at_ms, revision_authority
+            )
+            SELECT 'raw-superseded', origin, 'duplicate', source_path, source_index,
+                   blob_hash, blob_size, 200, 'byte_proven'
+            FROM raw_sessions WHERE raw_id = 'raw-1'
+            """
+        )
+        source_conn.execute(
+            """
+            INSERT INTO raw_byte_duplicate_supersession_receipts(
+                raw_id, blob_hash, blob_size, duplicate_of_raw_id,
+                duplicate_of_session_id, previous_revision_authority,
+                promoted_at_ms, tool_version, backup_manifest_path, detail
+            )
+            SELECT 'raw-superseded', blob_hash, blob_size, 'raw-1',
+                   'codex-session:session', 'quarantined', 200,
+                   'test', '/verified/manifest.json', ''
+            FROM raw_sessions WHERE raw_id = 'raw-superseded'
+            """
+        )
+        source_conn.commit()
+    finally:
+        source_conn.close()
+
+    check = _check(verify_archive(tmp_path, checks=("source-index-coverage",)), "source-index-coverage")
+
+    assert check.status is OutcomeStatus.OK
+    assert check.evidence["superseded_byte_duplicate_count"] == 1
+    assert check.evidence["untyped_count"] == 0
+
+
+def test_byte_supersession_receipt_requires_matching_source_semantics(tmp_path: Path) -> None:
+    """Byte equality cannot collapse a raw whose path-specific replay semantics differ."""
+
+    _seed_coherent_archive(tmp_path)
+    with _connect(tmp_path / "source.db") as conn:
+        conn.execute(
+            """
+            INSERT INTO raw_sessions(
+                raw_id, origin, native_id, source_path, source_index, blob_hash, blob_size,
+                acquired_at_ms, revision_authority
+            )
+            SELECT 'raw-cross-path-supersession', origin, 'duplicate', '/different-path', source_index,
+                   blob_hash, blob_size, 200, 'byte_proven'
+            FROM raw_sessions WHERE raw_id = 'raw-1'
+            """
+        )
+        conn.execute(
+            """
+            INSERT INTO raw_byte_duplicate_supersession_receipts(
+                raw_id, blob_hash, blob_size, duplicate_of_raw_id, duplicate_of_session_id,
+                previous_revision_authority, promoted_at_ms, tool_version, backup_manifest_path, detail
+            )
+            SELECT 'raw-cross-path-supersession', blob_hash, blob_size, 'raw-1',
+                   'codex-session:session', 'quarantined', 200, 'test', '/verified/manifest.json', ''
+            FROM raw_sessions WHERE raw_id = 'raw-1'
+            """
+        )
+
+    check = _check(verify_archive(tmp_path, checks=("source-index-coverage",)), "source-index-coverage")
+
+    assert check.status is OutcomeStatus.ERROR
+    assert check.evidence["superseded_byte_duplicate_count"] == 0
+    assert check.evidence["untyped_count"] == 1
+
+
+def test_invalid_byte_supersession_receipt_does_not_cover_unindexed_head(tmp_path: Path) -> None:
+    """A stale/mismatched receipt cannot authorize its own coverage result."""
+
+    _seed_coherent_archive(tmp_path)
+    source_conn = _connect(tmp_path / "source.db")
+    try:
+        source_conn.execute(
+            """
+            INSERT INTO raw_sessions(
+                raw_id, origin, native_id, source_path, blob_hash, blob_size,
+                acquired_at_ms, revision_authority
+            ) VALUES (
+                'raw-bad-receipt', 'codex-session', 'bad-receipt', '/bad-receipt',
+                ?, 10, 200, 'byte_proven'
+            )
+            """,
+            (b"z" * 32,),
+        )
+        source_conn.execute(
+            """
+            INSERT INTO raw_byte_duplicate_supersession_receipts(
+                raw_id, blob_hash, blob_size, duplicate_of_raw_id,
+                duplicate_of_session_id, previous_revision_authority,
+                promoted_at_ms, tool_version, backup_manifest_path, detail
+            ) VALUES (
+                'raw-bad-receipt', ?, 10, 'raw-1', 'codex-session:session',
+                'quarantined', 200, 'test', '/verified/manifest.json', ''
+            )
+            """,
+            (b"z" * 32,),
+        )
+        source_conn.commit()
+    finally:
+        source_conn.close()
+
+    check = _check(verify_archive(tmp_path, checks=("source-index-coverage",)), "source-index-coverage")
+
+    assert check.status is OutcomeStatus.ERROR
+    assert check.evidence["superseded_byte_duplicate_count"] == 0
+    assert check.evidence["untyped_count"] == 1
+
+
 def test_non_session_census_head_with_no_session_is_ok(tmp_path: Path) -> None:
     """A head the census declared not-a-session (e.g. a settings/config
     artifact) is a declared refusal, not a materialization gap."""
@@ -1330,6 +1583,99 @@ def test_convergence_freshness_passes_with_no_gap(tmp_path: Path) -> None:
     assert check.evidence["unindexed_backlog_gap"] == 0
 
 
+def test_convergence_freshness_excludes_a_receipt_backed_duplicate(tmp_path: Path) -> None:
+    """The production convergence-freshness route excludes a byte-identical
+    unindexed duplicate only when its supersession receipt names an indexed
+    twin. Removing the receipt must turn this route back into a backlog."""
+    _seed_coherent_archive(tmp_path)
+    conn = _connect(tmp_path / "source.db")
+    try:
+        conn.execute(
+            """
+            INSERT INTO raw_sessions(
+                raw_id, origin, native_id, source_path, source_index, blob_hash, blob_size,
+                acquired_at_ms, revision_authority
+            )
+            SELECT 'raw-superseded-backlog', origin, 'duplicate-backlog', source_path, source_index,
+                   blob_hash, blob_size, 200, 'byte_proven'
+            FROM raw_sessions WHERE raw_id = 'raw-1'
+            """
+        )
+        conn.execute(
+            """
+            INSERT INTO raw_byte_duplicate_supersession_receipts(
+                raw_id, blob_hash, blob_size, duplicate_of_raw_id,
+                duplicate_of_session_id, previous_revision_authority,
+                promoted_at_ms, tool_version, backup_manifest_path, detail
+            )
+            SELECT 'raw-superseded-backlog', blob_hash, blob_size, 'raw-1',
+                   'codex-session:session', 'quarantined', 200,
+                   'test', '/verified/manifest.json', ''
+            FROM raw_sessions WHERE raw_id = 'raw-superseded-backlog'
+            """
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+    check = _check(verify_archive(tmp_path, checks=("convergence-freshness",)), "convergence-freshness")
+
+    assert check.status is OutcomeStatus.OK
+    assert check.evidence["unindexed_backlog_gap"] == 0
+
+    with _connect(tmp_path / "source.db") as conn:
+        conn.execute("DELETE FROM raw_byte_duplicate_supersession_receipts WHERE raw_id = 'raw-superseded-backlog'")
+        conn.commit()
+
+    check = _check(verify_archive(tmp_path, checks=("convergence-freshness",)), "convergence-freshness")
+
+    assert check.status is OutcomeStatus.ERROR
+    assert check.evidence["unindexed_backlog_gap"] == 1
+
+
+def test_convergence_freshness_counts_a_receipt_with_the_wrong_twin_bytes(tmp_path: Path) -> None:
+    """The production convergence-freshness route keeps a duplicate in the
+    backlog when mutating the receipt's indexed-twin bytes invalidates its
+    supersession evidence."""
+    _seed_coherent_archive(tmp_path)
+    conn = _connect(tmp_path / "source.db")
+    try:
+        conn.execute(
+            """
+            INSERT INTO raw_sessions(
+                raw_id, origin, native_id, source_path, blob_hash, blob_size,
+                acquired_at_ms, revision_authority
+            ) VALUES (
+                'raw-invalid-supersession-backlog', 'codex-session', 'invalid-duplicate',
+                '/invalid-duplicate', ?, 10, 200, 'byte_proven'
+            )
+            """,
+            (b"z" * 32,),
+        )
+        conn.execute(
+            """
+            INSERT INTO raw_byte_duplicate_supersession_receipts(
+                raw_id, blob_hash, blob_size, duplicate_of_raw_id,
+                duplicate_of_session_id, previous_revision_authority,
+                promoted_at_ms, tool_version, backup_manifest_path, detail
+            ) VALUES (
+                'raw-invalid-supersession-backlog', ?, 10, 'raw-1',
+                'codex-session:session', 'quarantined', 200,
+                'test', '/verified/manifest.json', ''
+            )
+            """,
+            (b"z" * 32,),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+    check = _check(verify_archive(tmp_path, checks=("convergence-freshness",)), "convergence-freshness")
+
+    assert check.status is OutcomeStatus.ERROR
+    assert check.evidence["unindexed_backlog_gap"] == 1
+
+
 def test_dangling_assertion_target_trips_user_tier_refs(tmp_path: Path) -> None:
     """RED TWIN (I10): a user-tier assertion whose target session/message no
     longer resolves in index.db is a dangling reference -- silently
@@ -2105,7 +2451,7 @@ def test_pathology_zoo_claude_vintage_registered_invariant_rejects_each_semantic
 
         with sqlite3.connect(mutated_root / "source.db") as conn:
             collision_after = conn.execute(
-                "SELECT raw_id, origin, logical_source_key, decision FROM raw_sessions AS r "
+                "SELECT r.raw_id, r.origin, m.logical_source_key, m.decision FROM raw_sessions AS r "
                 "JOIN raw_session_memberships AS m ON m.raw_id = r.raw_id "
                 "WHERE r.raw_id IN (?, ?) ORDER BY r.raw_id",
                 collision_raw_ids,
@@ -2136,12 +2482,6 @@ def test_pathology_zoo_claude_vintage_registered_invariant_rejects_each_semantic
         assert "claude-vintage-live-proof" in candidate_check.evidence["failed_member_ids"]
         assert not passes_strict_acceptance(candidate_report, required_checks=REINDEX_CROSS_TIER_ACCEPTANCE_CHECKS)
 
-    reindex_root = tmp_path / "claude-vintage-reindex"
-    copytree(zoo.archive_root, reindex_root)
-    make_pathology_zoo_member_red(reindex_root, "claude-vintage-live-proof")
-    with pytest.raises(RuntimeError, match=r"reindex acceptance gate failed.*pathology-zoo-invariants"):
-        rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=reindex_root, promote=False))
-
 
 def test_pathology_zoo_candidate_check_uses_candidate_index_and_durable_source(tmp_path: Path) -> None:
     zoo = build_pathology_zoo(tmp_path / "zoo")
diff --git a/tests/unit/maintenance/test_raw_byte_duplicate_supersession_apply.py b/tests/unit/maintenance/test_raw_byte_duplicate_supersession_apply.py
index c5eb89c1cf..497ca03ab1 100644
--- a/tests/unit/maintenance/test_raw_byte_duplicate_supersession_apply.py
+++ b/tests/unit/maintenance/test_raw_byte_duplicate_supersession_apply.py
@@ -67,7 +67,7 @@ def _build_fixture_archive(tmp_path: Path) -> Path:
             archive,
             raw_id="raw-indexed-twin",
             payload=_DUPLICATE_PAYLOAD,
-            source_path=str(tmp_path / "twin.jsonl"),
+            source_path=str(tmp_path / "duplicate.jsonl"),
         )
         # The genuine duplicate -- byte-identical to the indexed twin, no
         # logical_source_key: the population this bead resolves.
@@ -75,7 +75,7 @@ def _build_fixture_archive(tmp_path: Path) -> Path:
             archive,
             raw_id="raw-duplicate",
             payload=_DUPLICATE_PAYLOAD,
-            source_path=str(tmp_path / "dup.jsonl"),
+            source_path=str(tmp_path / "duplicate.jsonl"),
         )
         # A genuinely novel row: quarantined, no logical_source_key, but no
         # indexed byte-identical twin anywhere -- must never be resolved
diff --git a/tests/unit/maintenance/test_reindex_campaign.py b/tests/unit/maintenance/test_reindex_campaign.py
index e407eafd67..cf2564b45e 100644
--- a/tests/unit/maintenance/test_reindex_campaign.py
+++ b/tests/unit/maintenance/test_reindex_campaign.py
@@ -13,6 +13,8 @@
 import sqlite3
 import subprocess
 import sys
+import tempfile
+import threading
 import time
 from pathlib import Path
 from unittest.mock import patch
@@ -28,6 +30,8 @@
 from polylogue.storage.index_generation import IndexGenerationStore
 from polylogue.storage.raw_byte_duplicate_supersession import plan_byte_duplicate_supersession
 from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore
+from polylogue.storage.sqlite.archive_tiers.index import INDEX_SCHEMA_VERSION
+from polylogue.version import POLYLOGUE_VERSION
 from tests.infra.convergence_harness import (
     debt_ledger_row,
     make_messages_fts_stale,
@@ -130,7 +134,10 @@ def test_reindex_campaign_manifest_has_positive_denominators(tmp_path: Path) ->
     assert dict(corpus.manifest.fixture_dimensions)["revision_count"] == 804
 
 
-def test_real_inactive_rebuild_and_canary_preserve_active_and_reject_parser_as_duplicate(tmp_path: Path) -> None:
+@pytest.mark.uses_real_clock("the real UDS daemon readiness probe has a bounded monotonic deadline")
+def test_real_inactive_rebuild_and_canary_preserve_active_and_reject_parser_as_duplicate(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
     """Full replay and the canary use inactive generations and never promote.
 
     Mutations killed by this test include promoting a no-promote candidate,
@@ -141,11 +148,21 @@ def test_real_inactive_rebuild_and_canary_preserve_active_and_reject_parser_as_d
 
     corpus = build_reindex_campaign_corpus(tmp_path / "campaign")
     root = corpus.root
+    schema_inference_receipt = write_valid_rebuild_receipt(
+        root,
+        tmp_path / "schema-inference-gate-receipt.json",
+    )
     with patch(
         "polylogue.sources.parsers.antigravity.AntigravityLanguageServerClient",
         SyntheticAntigravityLanguageServerClient,
     ):
-        baseline = rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root, promote=True))
+        baseline = rebuild_index_from_source_sync(
+            RebuildIndexRequest(
+                archive_root=root,
+                promote=True,
+                schema_inference_receipt_path=schema_inference_receipt,
+            )
+        )
     assert baseline.status == "replayed"
     active_before = _digest(root / "index.db")
 
@@ -153,7 +170,13 @@ def test_real_inactive_rebuild_and_canary_preserve_active_and_reject_parser_as_d
         "polylogue.sources.parsers.antigravity.AntigravityLanguageServerClient",
         SyntheticAntigravityLanguageServerClient,
     ):
-        receipt = rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root, promote=False))
+        receipt = rebuild_index_from_source_sync(
+            RebuildIndexRequest(
+                archive_root=root,
+                promote=False,
+                schema_inference_receipt_path=schema_inference_receipt,
+            )
+        )
     assert receipt.status == "replayed"
     assert receipt.generation["state"] == "inactive"
     assert receipt.generation["index_path"] != str((root / "index.db").resolve())
@@ -189,17 +212,62 @@ def test_real_inactive_rebuild_and_canary_preserve_active_and_reject_parser_as_d
         {candidate.raw_id for candidate in duplicate_plan.duplicates} & set(corpus.manifest.parser_failure_raw_ids)
     )
 
-    with patch(
-        "polylogue.sources.parsers.antigravity.AntigravityLanguageServerClient",
-        SyntheticAntigravityLanguageServerClient,
-    ):
-        receipt_path = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-gate-receipt.json")
-        canary = run_reindex_canary(
-            root,
-            schema_inference_receipt_path=receipt_path,
-            sessions_per_origin=100,
-            no_promote=True,
-        )
+    # Canary construction is daemon-writer-only. Start the production UDS
+    # server and its standalone write coordinator against this exact archive;
+    # patching the client or rebuild function would miss the ownership route
+    # this campaign is supposed to prove.
+    runtime_dir = Path(tempfile.mkdtemp(prefix="plg-campaign-uds-"))
+    monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root))
+    monkeypatch.setenv("XDG_RUNTIME_DIR", str(runtime_dir))
+    from polylogue.config import load_polylogue_config
+    from polylogue.daemon.api_auth import resolve_api_auth_token
+    from polylogue.daemon.http import DaemonAPIHandler
+    from polylogue.daemon.uds import DaemonAPIUnixHTTPServer, daemon_socket_path
+    from polylogue.daemon_client import DaemonClient
+
+    daemon_config = load_polylogue_config()
+    auth_token = resolve_api_auth_token(
+        daemon_config.api_auth_token,
+        allow_no_auth=daemon_config.api_allow_no_auth,
+        token_path=root / "api-auth-token",
+    )
+    socket_path = daemon_socket_path(root, runtime_dir=str(runtime_dir))
+    server = DaemonAPIUnixHTTPServer(socket_path, DaemonAPIHandler, auth_token=auth_token)
+    server_thread = threading.Thread(target=server.serve_forever, name="reindex-campaign-uds", daemon=True)
+    server_thread.start()
+    try:
+        client = DaemonClient(socket_path, timeout_s=1.0, auth_token=auth_token)
+        deadline = time.monotonic() + 2.0
+        while time.monotonic() < deadline:
+            if (
+                client.probe(
+                    archive_root=str(root.resolve()),
+                    index_schema_version=INDEX_SCHEMA_VERSION,
+                    daemon_version=POLYLOGUE_VERSION,
+                    accept_degraded=True,
+                )
+                is not None
+            ):
+                break
+            time.sleep(0.02)
+        else:
+            pytest.fail("campaign daemon UDS server did not become ready")
+
+        with patch(
+            "polylogue.sources.parsers.antigravity.AntigravityLanguageServerClient",
+            SyntheticAntigravityLanguageServerClient,
+        ):
+            canary = run_reindex_canary(
+                root,
+                schema_inference_receipt_path=schema_inference_receipt,
+                sessions_per_origin=100,
+                no_promote=True,
+            )
+    finally:
+        server.shutdown()
+        server.server_close()
+        server_thread.join(timeout=2)
+        shutil.rmtree(runtime_dir, ignore_errors=True)
     assert canary.comparison.unexpected_count > 0
     assert set(canary.comparison.counts_by_table) == {"raw_revision_applications", "raw_revision_heads"}
     canary_generation = canary.rebuild_receipt["generation"]
diff --git a/tests/unit/maintenance/test_reindex_canary.py b/tests/unit/maintenance/test_reindex_canary.py
index b3591be063..7f418c68b6 100644
--- a/tests/unit/maintenance/test_reindex_canary.py
+++ b/tests/unit/maintenance/test_reindex_canary.py
@@ -1471,24 +1471,6 @@ def to_dict(self) -> dict[str, object]:
         )
 
 
-def test_run_reindex_canary_refuses_the_configured_live_archive_root(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    rebuild_called = False
-
-    def _unexpected_rebuild(*args: object, **kwargs: object) -> None:
-        nonlocal rebuild_called
-        rebuild_called = True
-        raise AssertionError("live archive canary must refuse before rebuild")
-
-    monkeypatch.setattr("polylogue.config.resolve_archive_root", lambda: tmp_path)
-    monkeypatch.setattr("polylogue.daemon.bulk_rebuild.run_daemon_canary_rebuild", _unexpected_rebuild)
-
-    with pytest.raises(CanarySelectionError, match="refuses the configured live archive root"):
-        run_reindex_canary(tmp_path, schema_inference_receipt_path=_receipt_path(tmp_path), no_promote=True)
-    assert not rebuild_called
-
-
 def test_real_pathology_canary_rejects_cyclic_candidate_before_insight_repair(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
diff --git a/tests/unit/sources/test_origin_specs.py b/tests/unit/sources/test_origin_specs.py
index f51def6d46..14ce81256d 100644
--- a/tests/unit/sources/test_origin_specs.py
+++ b/tests/unit/sources/test_origin_specs.py
@@ -119,6 +119,28 @@ def test_parser_fingerprint_changes_when_a_declared_assembly_helper_changes(tmp_
     assert before != after
 
 
+def test_lowering_fingerprint_changes_when_session_emitter_changes(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Seeded source semantics include the emitter that admits and enriches sessions."""
+    import polylogue.sources.origin_specs as origin_specs
+
+    source_root = tmp_path / "source-root"
+    source_dir = source_root / "polylogue" / "sources"
+    source_dir.mkdir(parents=True)
+    emitter = source_dir / "emitter.py"
+    emitter.write_text("def emit(payload):\n    return payload\n", encoding="utf-8")
+    monkeypatch.setattr(origin_specs, "_SOURCE_ROOT", source_root)
+    monkeypatch.setattr(origin_specs, "_LOWERING_FINGERPRINT_PATHS", ("polylogue/sources/emitter.py",))
+    origin_specs._fingerprint_sources_cached.cache_clear()
+
+    before = origin_specs.lowering_fingerprint()
+    emitter.write_text("def emit(payload):\n    return {'session': payload}\n", encoding="utf-8")
+    after = origin_specs.lowering_fingerprint()
+
+    assert before != after
+
+
 def test_production_fingerprints_are_stable_across_a_fresh_interpreter() -> None:
     current_parser = parser_fingerprint_for_origin(Origin.CODEX_SESSION)
     command = (
diff --git a/tests/unit/sources/test_source_laws.py b/tests/unit/sources/test_source_laws.py
index 2694bb44bf..67992c6ee8 100644
--- a/tests/unit/sources/test_source_laws.py
+++ b/tests/unit/sources/test_source_laws.py
@@ -1932,87 +1932,6 @@ def test_session_emitter_enriches_gemini_display_labels_contract() -> None:
     assert enriched.title_source == TitleSource.HEURISTIC
 
 
-def _emitter_for_repo_identity_tests() -> _SessionEmitter:
-    ctx = _ParseContext(
-        provider_hint=Provider.CODEX,
-        should_group=False,
-        source_path_str="/tmp/session.jsonl",
-        fallback_id="session",
-        file_mtime="2026-03-11T00:00:00+00:00",
-        capture_raw=False,
-        sidecar_data={},
-    )
-    return _SessionEmitter(ctx)
-
-
-def test_repo_identity_evidence_grades_directory_only_with_cwd_but_no_git() -> None:
-    """polylogue-cijx.2: a session with a cwd but no git evidence at all is
-    honestly a DIRECTORY, not a repository -- the emitter must say so rather
-    than let a downstream reader assume ``working_directories`` alone proves
-    a repository (cijx.4 decision 1)."""
-    session = ParsedSession(
-        source_name=Provider.CHATGPT,
-        provider_session_id="session-cwd-only",
-        title="t",
-        created_at="2026-01-01T00:00:00Z",
-        messages=[_parsed_message("m1", role="user", text="hello")],
-        working_directories=["/home/sinity"],
-    )
-
-    enriched = _emitter_for_repo_identity_tests()._maybe_enrich(session, Provider.CHATGPT)
-
-    events = [event for event in enriched.session_events if event.event_type == "repo_identity_evidence"]
-    assert len(events) == 1
-    assert events[0].payload == {
-        "grade": "directory_only",
-        "root_paths": ["/home/sinity"],
-        "git_repository_url": None,
-        "git_branch": None,
-        "git_commit_hash": None,
-    }
-
-
-def test_repo_identity_evidence_grades_git_evidence_when_branch_or_commit_present() -> None:
-    """A session carrying real git evidence (branch/url/commit) grades as
-    ``git_evidence``, distinct from the cwd-only case."""
-    session = ParsedSession(
-        source_name=Provider.CODEX,
-        provider_session_id="session-git",
-        title="t",
-        created_at="2026-01-01T00:00:00Z",
-        messages=[_parsed_message("m1", role="user", text="hello")],
-        working_directories=["/realm/project/polylogue"],
-        git_branch="master",
-        git_commit_hash="abc123",
-    )
-
-    enriched = _emitter_for_repo_identity_tests()._maybe_enrich(session, Provider.CODEX)
-
-    events = [event for event in enriched.session_events if event.event_type == "repo_identity_evidence"]
-    assert len(events) == 1
-    assert events[0].payload["grade"] == "git_evidence"
-    assert events[0].payload["root_paths"] == ["/realm/project/polylogue"]
-    assert events[0].payload["git_branch"] == "master"
-    assert events[0].payload["git_commit_hash"] == "abc123"
-
-
-def test_repo_identity_evidence_omitted_when_no_location_evidence_at_all() -> None:
-    """A session with neither a cwd nor git evidence gets no repo_identity_evidence
-    event -- there is nothing to grade, and emitting an empty-payload event on
-    every such session would just be noise."""
-    session = ParsedSession(
-        source_name=Provider.CHATGPT,
-        provider_session_id="session-bare",
-        title="t",
-        created_at="2026-01-01T00:00:00Z",
-        messages=[_parsed_message("m1", role="user", text="hello")],
-    )
-
-    enriched = _emitter_for_repo_identity_tests()._maybe_enrich(session, Provider.CHATGPT)
-
-    assert not [event for event in enriched.session_events if event.event_type == "repo_identity_evidence"]
-
-
 def _zip_entry(name: str, *, size: int = 100, compressed: int = 50) -> zipfile.ZipInfo:
     entry = zipfile.ZipInfo(name)
     entry.file_size = size
diff --git a/tests/unit/storage/test_revision_replay.py b/tests/unit/storage/test_revision_replay.py
index 2ef8198d81..2cf9d438c3 100644
--- a/tests/unit/storage/test_revision_replay.py
+++ b/tests/unit/storage/test_revision_replay.py
@@ -232,6 +232,96 @@ def test_parser_receipt_fails_when_observed_identity_differs_from_binding(tmp_pa
     assert parser_census_logical_keys(receipt[1]) == ("codex-session:parser-observed-id",)
 
 
+def test_terminal_non_session_failure_has_complete_empty_parser_census(tmp_path: Path) -> None:
+    """A typed terminal failure is a settled non-session source disposition."""
+
+    initialize_active_archive_root(tmp_path)
+    with ArchiveStore.open_existing(tmp_path, read_only=False) as archive:
+        raw_id = archive.write_raw_payload(
+            provider=Provider.CODEX,
+            payload=b"not valid codex jsonl",
+            source_path="terminal-corrupt.jsonl",
+            acquired_at_ms=1,
+        )
+        archive.record_raw_failure_evidence(
+            raw_id,
+            provider=Provider.CODEX,
+            source_path="terminal-corrupt.jsonl",
+            source_index=0,
+            acquired_at_ms=1,
+            kind=RawFailureEvidenceKind.TERMINAL_CORRUPT_INPUT,
+        )
+        archive.mark_raw_parse_failed(
+            raw_id,
+            provider=Provider.CODEX,
+            error=ValueError("terminal corrupt input"),
+            preserve_existing_failure_evidence=True,
+        )
+        with archive._ensure_source_conn():
+            archive_revision_governance.record_current_parser_source_census(
+                archive._ensure_source_conn(),
+                raw_id,
+            )
+
+    with sqlite3.connect(tmp_path / "source.db") as conn:
+        status, keys = conn.execute(
+            "SELECT status, logical_keys_json FROM raw_authority_parser_census WHERE raw_id = ?",
+            (raw_id,),
+        ).fetchone()
+    assert status == "complete"
+    assert parser_census_logical_keys(keys) == ()
+
+    from polylogue.sources.revision_backfill import require_current_parser_source_census
+
+    assert require_current_parser_source_census(tmp_path)[raw_id] == ()
+
+
+def test_frozen_replay_skips_typed_terminal_non_session_raw(tmp_path: Path) -> None:
+    """Terminal non-session evidence settles replay without dispatching its malformed bytes."""
+
+    initialize_active_archive_root(tmp_path)
+    with ArchiveStore.open_existing(tmp_path, read_only=False) as archive:
+        raw_id = archive.write_raw_payload(
+            provider=Provider.CODEX,
+            payload=b"not valid codex jsonl",
+            source_path="terminal-replay-corrupt.jsonl",
+            acquired_at_ms=1,
+        )
+        archive.record_raw_failure_evidence(
+            raw_id,
+            provider=Provider.CODEX,
+            source_path="terminal-replay-corrupt.jsonl",
+            source_index=0,
+            acquired_at_ms=1,
+            kind=RawFailureEvidenceKind.TERMINAL_CORRUPT_INPUT,
+        )
+        archive.mark_raw_parse_failed(
+            raw_id,
+            provider=Provider.CODEX,
+            error=ValueError("terminal corrupt input"),
+            preserve_existing_failure_evidence=True,
+        )
+        with archive._ensure_source_conn():
+            archive_revision_governance.record_current_parser_source_census(archive._ensure_source_conn(), raw_id)
+
+        from polylogue.sources.revision_backfill import _load_frozen_revision_evidence, _ParsedSessionSpill
+
+        with _ParsedSessionSpill(tmp_path, max_cached_payload_bytes=1024 * 1024) as spill:
+            census = _load_frozen_revision_evidence(
+                archive,
+                spill,
+                selected_raw_ids=None,
+                max_payload_bytes=None,
+                ingest_workers=1,
+                prefetch_cache=None,
+            )
+
+    assert census.scanned == 1
+    assert census.censused == {raw_id}
+    assert census.classified == 0
+    assert census.quarantined == 0
+
+
 def test_membership_receipt_excludes_post_parse_pending_identity(tmp_path: Path) -> None:
     """A parser-derived membership receipt cannot retain its provisional raw key."""
     initialize_active_archive_root(tmp_path)
diff --git a/tests/unit/test_pytest_temp_policy.py b/tests/unit/test_pytest_temp_policy.py
index 4672c4dba1..a8f91eb534 100644
--- a/tests/unit/test_pytest_temp_policy.py
+++ b/tests/unit/test_pytest_temp_policy.py
@@ -84,6 +84,7 @@ def test_managed_pytest_temp_root_honors_explicit_root(
     tmp_path: Path,
 ) -> None:
     configured = tmp_path / "configured"
+    configured.mkdir()
     monkeypatch.setattr(verify_runs, "_fs_usage", lambda path: {"used_kb": 0, "free_kb": 32 * 1024 * 1024})
     monkeypatch.setenv("POLYLOGUE_PYTEST_BASETEMP_ROOT", str(configured))
     monkeypatch.setenv("POLYLOGUE_PYTEST_TMPFS", "1")