fix(reindex): restore source-authoritative acceptance

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

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)

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:

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 = {

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.

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,

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
     ):

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: