fix(live): follow active index authority in every route

devtools/verify.py

@@ -3861,24 +3861,32 @@ def _main(argv: list[str] | None = None) -> int:
     final_checkout_fingerprint = worktree_fingerprint(ROOT)
     mutation_observation = finish_checkout_mutation_monitor(mutation_monitor)
     checkout_stable = True
-    if (
-        changed_path_authority_failed
-        or head is None
+    checkout_fingerprint_unavailable = (
+        head is None
         or final_head is None
-        or "unavailable" in {checkout_fingerprint, final_checkout_fingerprint}
-        or mutation_observation.unavailable
-    ):
+        or "unavailable"
+        in {
+            checkout_fingerprint,
+            final_checkout_fingerprint,
+        }
+    )
+    if changed_path_authority_failed or checkout_fingerprint_unavailable or mutation_observation.unavailable:
         checkout_stable = False
+        diagnosis = (
+            "testmon_changed_path_authority_unavailable"
+            if changed_path_authority_failed
+            else (
+                "checkout_fingerprint_unavailable"
+                if checkout_fingerprint_unavailable
+                else "checkout_mutation_monitor_unavailable"
+            )
+        )
         step_results.append(
             {
                 "name": "checkout stability",
                 "duration_s": 0.0,
                 "exit": 125,
-                "diagnosis": (
-                    "testmon_changed_path_authority_unavailable"
-                    if changed_path_authority_failed
-                    else "checkout_fingerprint_unavailable"
-                ),
+                "diagnosis": diagnosis,
                 "initial_git_head": head,
                 "final_git_head": final_head,
                 "initial_worktree_fingerprint": checkout_fingerprint,
@@ -3887,7 +3895,7 @@ def _main(argv: list[str] | None = None) -> int:
         )
         if exit_code == 0:
             exit_code = 125
-        sys.stderr.write("verify: checkout fingerprint unavailable; evidence is not exact-head.\n")
+        sys.stderr.write(f"verify: {diagnosis.replace('_', ' ')}; evidence is not exact-head.\n")
     elif final_head != head or mutation_observation.changed or final_checkout_fingerprint != checkout_fingerprint:
         checkout_stable = False
         step_results.append(

devtools/verify_runs.py

@@ -277,12 +277,15 @@ def __init__(self, root: Path) -> None:
         self._unavailable = False
         self._stop = threading.Event()
         self._ready = threading.Event()
+        self._initialized = threading.Event()
         self._thread: threading.Thread | None = None
         self._state_lock = threading.Lock()
         self._tracked_paths: frozenset[Path] = frozenset()
         self._tracked_directories: frozenset[Path] = frozenset()
         self._ignored_roots: frozenset[Path] = frozenset()
         self._git_index_path: Path | None = None
+        self._git_current_ref_path: Path | None = None
+        self._git_current_ref_was_loose: bool | None = None
         self._git_authority_paths: dict[Path, str] = {}
         self._directory_topology_fingerprint: frozenset[str] | None = None
 
@@ -292,14 +295,35 @@ def start(self) -> None:
             with self._state_lock:
                 self._unavailable = True
             self._ready.set()
+            self._initialized.set()
             return
-        self._thread = threading.Thread(target=self._watch, name="checkout-mutation-monitor", daemon=True)
+        # Repository enumeration and Git authority discovery are synchronous
+        # preflight, not native watcher startup. Keeping them outside the
+        # backend deadline prevents a slow CI checkout from consuming the
+        # entire readiness budget before watchfiles can initialize.
+        watched_directories = self._watched_directories()
+        if self._unavailable:
+            self._ready.set()
+            self._initialized.set()
+            return
+        self._thread = threading.Thread(
+            target=self._watch,
+            args=(watched_directories,),
+            name="checkout-mutation-monitor",
+            daemon=True,
+        )
         self._thread.start()
         if not self._ready.wait(timeout=self._WATCH_START_TIMEOUT_S):
             with self._state_lock:
                 self._unavailable = True
             self._stop.set()
             self._thread.join(timeout=self._WATCH_START_TIMEOUT_S)
+            return
+        # The one-second deadline proves only native backend startup. The
+        # protected topology recheck is ordinary repository discovery and may
+        # legitimately take longer on a cold CI checkout; complete it before
+        # the verification command can mutate the tree.
+        self._initialized.wait()
 
     def finish(self) -> CheckoutMutationObservation:
         """Stop monitoring only after the caller took its final fingerprint."""
@@ -320,11 +344,8 @@ def finish(self) -> CheckoutMutationObservation:
                 observed_path=self._observed_path,
             )
 
-    def _watch(self) -> None:
+    def _watch(self, watched_directories: Sequence[Path]) -> None:
         try:
-            watched_directories = self._watched_directories()
-            if self._unavailable:
-                return
             for changes in watchfiles.watch(
                 *watched_directories,
                 watch_filter=None,
@@ -338,12 +359,17 @@ def _watch(self) -> None:
                 recursive=False,
             ):
                 # An empty timeout batch proves the backend initialized before
-                # a verification command starts, closing the startup race.
-                if not self._ready.is_set() and not self._directory_topology_is_stable(watched_directories):
-                    with self._state_lock:
-                        self._unavailable = True
-                    return
-                self._ready.set()
+                # a verification command starts, closing the startup race. The
+                # active watcher protects the following topology recheck. The
+                # native-ready event has its own bounded startup deadline;
+                # ``start`` waits separately for repository discovery.
+                if not self._ready.is_set():
+                    self._ready.set()
+                    if not self._directory_topology_is_stable(watched_directories):
+                        with self._state_lock:
+                            self._unavailable = True
+                        return
+                    self._initialized.set()
                 for _change, raw_path in changes:
                     self._record_change(Path(raw_path))
                     if self._changed or self._unavailable:
@@ -356,6 +382,7 @@ def _watch(self) -> None:
                 self._unavailable = True
         finally:
             self._ready.set()
+            self._initialized.set()
 
     @classmethod
     def _polling_backend_requested(cls) -> bool:
@@ -459,6 +486,12 @@ def _resolve_git_head_paths(self) -> dict[Path, str]:
                 self._unavailable = True
             return paths
         paths[Path(raw_head_path)] = ".git/HEAD"
+        if symbolic_result.returncode == 1:
+            # A detached checkout's complete revision authority is the
+            # worktree-specific HEAD file. packed-refs is shared by every
+            # linked worktree, so unrelated fetch/pack maintenance cannot
+            # mutate this checkout and must not invalidate its verification.
+            return paths
         packed_result = self._git_command(["rev-parse", "--path-format=absolute", "--git-path", "packed-refs"])
         if packed_result is None:
             return paths
@@ -468,16 +501,18 @@ def _resolve_git_head_paths(self) -> dict[Path, str]:
                 self._unavailable = True
             return paths
         paths[Path(raw_packed_path)] = ".git/packed-refs"
-        if symbolic_result.returncode == 0:
-            ref_result = self._git_command(["rev-parse", "--path-format=absolute", "--git-path", symbolic_ref])
-            if ref_result is None:
-                return paths
-            raw_ref_path = os.fsdecode(ref_result.stdout).strip()
-            if not raw_ref_path:
-                with self._state_lock:
-                    self._unavailable = True
-                return paths
-            paths[Path(raw_ref_path)] = f".git/{symbolic_ref}"
+        ref_result = self._git_command(["rev-parse", "--path-format=absolute", "--git-path", symbolic_ref])
+        if ref_result is None:
+            return paths
+        raw_ref_path = os.fsdecode(ref_result.stdout).strip()
+        if not raw_ref_path:
+            with self._state_lock:
+                self._unavailable = True
+            return paths
+        self._git_current_ref_path = Path(raw_ref_path)
+        if self._git_current_ref_was_loose is None:
+            self._git_current_ref_was_loose = self._git_current_ref_path.exists()
+        paths[self._git_current_ref_path] = f".git/{symbolic_ref}"
         return paths
 
     def _git_command(
@@ -527,6 +562,14 @@ def _record_change(self, candidate: Path) -> None:
         if not candidate.is_absolute():
             candidate = self.root / candidate
         for authority_path, label in self._git_authority_paths.items():
+            if label == ".git/packed-refs" and self._git_current_ref_was_loose is True:
+                # packed-refs is shared by linked worktrees. When this
+                # worktree's current branch has a loose ref, unrelated fetch
+                # maintenance cannot change its HEAD through the packed file.
+                # A real pack transition remains visible when the loose ref
+                # is removed or replaced. Preserve the startup state so a
+                # packed-to-loose transition cannot hide its own first event.
+                continue
             if candidate != authority_path and authority_path.is_relative_to(candidate):
                 with self._state_lock:
                     self._changed = True

docs/plans/classifier-fingerprints.json

@@ -10,11 +10,11 @@
       }
     },
     "polylogue/archive/artifact_taxonomy/runtime.py:classify_artifact_path": {
-      "fingerprint": "10f54d1827fdca585985cf3180e54c27098de26121530076e51a26cef7b2de24",
+      "fingerprint": "32b6f26516b4cc9ed0342c262e492fca469aaca9fb47a66e4ca1c76b7f58987a",
       "covered_by": {
         "kind": "acknowledged_safe",
-        "reason": "Baseline snapshot at classifier-fingerprint gate introduction (polylogue-gucv).",
-        "ref": "polylogue-gucv"
+        "reason": "Strong-only admission helper preserves this classifier output; no archived payload classification changes.",
+        "ref": "#3952"
       }
     },
     "polylogue/archive/artifact_taxonomy/support.py:looks_like_beads_interaction": {

docs/plans/layering.yaml

@@ -60,7 +60,7 @@ writer_modules:
           interruption: atomic
       entrypoints:
         [apply_source_raw_state_update, bind_source_raw_revision, record_capture_mode_observation,
-        record_excised_blob_hash, write_history_sidecar,
+        record_excised_blob_hash, record_raw_container_coordinate, write_history_sidecar,
         delete_source_hook_event, write_source_blob_refs, write_source_hook_event, write_source_raw_session,
         write_source_raw_session_blob_ref, upsert_raw_artifact]
     - path: polylogue/storage/sqlite/archive_tiers/write.py

polylogue/archive/artifact_taxonomy/__init__.py

@@ -8,11 +8,16 @@
 from __future__ import annotations
 
 from polylogue.archive.artifact_taxonomy.models import ArtifactClassification, ArtifactKind
-from polylogue.archive.artifact_taxonomy.runtime import classify_artifact, classify_artifact_path
+from polylogue.archive.artifact_taxonomy.runtime import (
+    classify_artifact,
+    classify_artifact_path,
+    strong_path_classification,
+)
 
 __all__ = [
     "ArtifactClassification",
     "ArtifactKind",
     "classify_artifact",
     "classify_artifact_path",
+    "strong_path_classification",
 ]

polylogue/archive/artifact_taxonomy/runtime.py

@@ -102,6 +102,21 @@ def classify_artifact_path(
     """
     if weak := _self_generated_artifact_dir_classification(source_path, provider=provider):
         return weak
+    return strong_path_classification(source_path, provider=provider)
+
+
+def strong_path_classification(
+    source_path: str | Path | None,
+    *,
+    provider: str | Provider,
+) -> ArtifactClassification | None:
+    """Classify only definitive path rules.
+
+    Live admission uses this before deciding whether a payload may enter a
+    bounded streaming route. The weak ``analysis/`` location heuristic is
+    deliberately excluded there because it must yield to bounded payload
+    evidence or the streaming policy.
+    """
     return _classify_artifact_path_strong(source_path, provider=provider)