fix(governance): RED-1 — gate governed close on current-code-vs-attach drift (filigree-owned)

src/filigree/core.py

@@ -53,6 +53,7 @@
     DEFAULT_LOOMWEAVE_TOKEN_ENV,
     BatchQuery,
     BatchResolution,
+    EntityHashResolution,
     LocalRegistry,
     LoomweaveCapabilities,
     LoomweaveRegistry,
@@ -1454,6 +1455,22 @@ def resolve_files_batch(
             )
             return self._fallback.resolve_files_batch(queries, actor=actor)
 
+    def resolve_entity_content_hashes(self, entity_ids: list[str]) -> EntityHashResolution:
+        """Delegate entity content-hash resolution to the Loomweave primary.
+
+        There is no *local* analogue (a local registry cannot compute a drift
+        hash), so this only forwards. A whole-backend availability failure
+        propagates as :class:`RegistryUnavailableError` — the closure-gate
+        consumer catches it and degrades to a discriminated freshness UNKNOWN
+        without hard-blocking the close (enrich-only). A primary that predates
+        this surface (an older injected fake) resolves every id to UNKNOWN.
+        """
+        primary_method = getattr(self._primary, "resolve_entity_content_hashes", None)
+        if primary_method is None:
+            return EntityHashResolution(resolved={}, unresolved=list(entity_ids))
+        result: EntityHashResolution = primary_method(entity_ids)
+        return result
+
     def is_displaced(self) -> bool:
         return self._primary.is_displaced()
 

src/filigree/db_base.py

@@ -8,7 +8,7 @@
 import logging
 import sqlite3
 import time
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, ParamSpec, Protocol, TypeVar, cast
@@ -510,7 +510,12 @@ def add_entity_association(
         _skip_begin: bool = False,
     ) -> EntityAssociationRow: ...
 
-    def list_entity_associations(self, issue_id: IssueId) -> list[EntityAssociationRow]: ...
+    def list_entity_associations(
+        self,
+        issue_id: IssueId,
+        *,
+        current_content_hashes: Mapping[str, str] | None = None,
+    ) -> list[EntityAssociationRow]: ...
 
     # -- ObservationsMixin ---------------------------------------------------
 

src/filigree/db_entity_associations.py

@@ -399,14 +399,29 @@ def remove_entity_association(
             )
         return cursor.rowcount > 0
 
-    def list_entity_associations(self, issue_id: IssueId) -> list[EntityAssociationRow]:
+    def list_entity_associations(
+        self,
+        issue_id: IssueId,
+        *,
+        current_content_hashes: Mapping[str, str] | None = None,
+    ) -> list[EntityAssociationRow]:
         """Return all entity associations for an issue.
 
         Returns raw rows in attach-time order. Drift detection is the
         caller's job — Filigree does not compute or surface
         ``drift_warning`` here per ADR-029 §"Decision 3"; that's the
         consumer's (Loomweave's ``issues_for``) responsibility after
         fetching the rows.
+
+        ``current_content_hashes`` is an OPTIONAL ``entity_id -> current
+        content_hash`` map a caller may supply when it has already resolved live
+        hashes (e.g. from ``registry.resolve_entity_content_hashes``). When
+        provided, each row's ``freshness_status`` is computed against the matching
+        current hash (``fresh`` / ``stale``); rows with no entry — and the default
+        ``None`` map — keep the ``"unknown"`` sentinel. The data layer itself
+        never performs a network resolve (mirrors the reverse-lookup
+        ``list_associations_by_entity`` convention): resolution stays on the
+        network-allowed registry layer, the read surface only joins it in.
         """
         issue_id = make_issue_id(issue_id)
         self._check_id_prefix(issue_id)
@@ -421,7 +436,8 @@ def list_entity_associations(self, issue_id: IssueId) -> list[EntityAssociationR
             """,
             (issue_id,),
         ).fetchall()
-        return [_row_to_entity_association(r) for r in rows]
+        hashes = current_content_hashes or {}
+        return [_row_to_entity_association(r, current_content_hash=hashes.get(str(r["loomweave_entity_id"]))) for r in rows]
 
     def _row_to_by_entity_association(
         self, r: Mapping[str, Any], *, current_content_hash: str | None = None

src/filigree/governance.py

@@ -26,14 +26,18 @@
 
 from __future__ import annotations
 
+import logging
 from dataclasses import dataclass
 from enum import Enum
 from typing import Any, Protocol
 
 from filigree import legis_client
 from filigree.legis_client import LegisGateResult, LegisGateStatus
+from filigree.registry import RegistryUnavailableError, RegistryVersionMismatchError
 from filigree.types.core import make_issue_id
 
+logger = logging.getLogger(__name__)
+
 
 class GateOutcome(Enum):
     """What the close surface should do with a gate decision."""
@@ -107,6 +111,79 @@ def _signed_row_is_stale(row: Any) -> bool:
     return bool(signed != row.get("content_hash_at_attach"))
 
 
+def _row_entity_id(row: Any) -> str:
+    """The opaque entity id a binding row points at (forward or by-entity row)."""
+    return str(row.get("loomweave_entity_id") or row.get("entity_id") or "")
+
+
+def _evaluate_current_drift(db: Any, issue_id: str, signed_rows: list[Any]) -> GateDecision | None:
+    """Compare each governed binding's CURRENT content against its attach snapshot.
+
+    RED-1: the sign-off-snapshot staleness check (``_signed_row_is_stale``) only
+    catches a *re-attach* that advanced ``content_hash_at_attach`` past the signed
+    snapshot. It cannot catch the bound CODE drifting while nobody re-attaches —
+    then ``content_hash_at_attach`` stays frozen at (and equal to) the signed
+    snapshot, so the close was waved through despite the code having moved on.
+
+    Filigree owns this comparison (ADR-029 Decision 3; hub ruling 2026-06-29):
+    resolve each governed binding's CURRENT ``content_hash`` from the Loomweave
+    registry consumer and compare it to ``content_hash_at_attach``.
+
+    Returns a ``STALE`` :class:`GateDecision` if ANY governed binding's current
+    content differs from its attach snapshot; otherwise ``None`` (no current-code
+    drift — the caller proceeds to the existing Legis gate). The check is
+    **enrich-only**: when Loomweave is unreachable / unsupported (no registry,
+    no resolver surface, availability error) or an individual entity is
+    unresolvable (orphaned / not_found / invalid), the binding's freshness is a
+    discriminated UNKNOWN — logged, never silently treated as fresh, and never a
+    hard block. The core close must not become load-bearing on Loomweave.
+    """
+    resolver = getattr(getattr(db, "registry", None), "resolve_entity_content_hashes", None)
+    if resolver is None:
+        # Local-mode / injected fake / pre-surface fallback registry: cannot
+        # determine current-code drift. Flag UNKNOWN, do not block.
+        logger.info(
+            "closure-gate drift check: no Loomweave resolver; entity freshness UNKNOWN (enrich-only, close not blocked on this axis)",
+            extra={"issue_id": issue_id, "freshness": "unknown", "reason": "no_resolver"},
+        )
+        return None
+
+    entity_ids = [_row_entity_id(row) for row in signed_rows if _row_entity_id(row)]
+    try:
+        resolution = resolver(entity_ids)
+    except (RegistryUnavailableError, RegistryVersionMismatchError) as exc:
+        logger.warning(
+            "closure-gate drift check: Loomweave unavailable; entity freshness UNKNOWN (enrich-only, close not blocked on this axis)",
+            extra={"issue_id": issue_id, "freshness": "unknown", "reason": "registry_unavailable", "detail": str(exc)},
+        )
+        return None
+
+    resolved: dict[str, str] = dict(resolution.get("resolved", {}))
+    drifted: list[str] = []
+    unknown: list[str] = []
+    for row in signed_rows:
+        entity_id = _row_entity_id(row)
+        if not entity_id:
+            continue
+        current = resolved.get(entity_id)
+        attach = row.get("content_hash_at_attach")
+        if current is None:
+            unknown.append(entity_id)
+        elif attach is not None and current != attach:
+            drifted.append(entity_id)
+    if unknown:
+        logger.warning(
+            "closure-gate drift check: entity content unresolvable; freshness UNKNOWN (enrich-only, close not blocked on this axis)",
+            extra={"issue_id": issue_id, "freshness": "unknown", "reason": "entity_unresolved", "entity_ids": unknown},
+        )
+    if drifted:
+        return GateDecision(
+            GateOutcome.STALE,
+            "entity content drifted since attach (current code no longer matches content at attach); awaiting re-attest",
+        )
+    return None
+
+
 def evaluate_closure_gate(db: _AssocReader, issue_id: str, *, legis_known_down: bool = False) -> GateDecision:
     """Decide whether *issue_id* may be closed.
 
@@ -140,6 +217,15 @@ def evaluate_closure_gate(db: _AssocReader, issue_id: str, *, legis_known_down:
         # Fail closed locally — do NOT consult Legis (it is asked only issue_id
         # and would answer for the stale snapshot it last saw).
         return GateDecision(GateOutcome.STALE, "entity content drifted since the Legis sign-off; awaiting re-sign")
+    drift = _evaluate_current_drift(db, str(issue_id), signed_rows)
+    if drift is not None:
+        # Current code has moved on since the binding was attached — fail closed
+        # as STALE, like the sign-off-snapshot drift above. This runs BEFORE the
+        # legis_known_down short-circuit (same load-bearing ordering as the
+        # snapshot check): a drifted binding must report STALE, never be masked
+        # as a transient UNAVAILABLE. A Loomweave outage does NOT reach here as a
+        # block — it degrades to UNKNOWN inside the helper (enrich-only).
+        return drift
     if legis_known_down:
         # A governed, non-stale issue needs a Legis round-trip, but a prior issue
         # in this batch already proved Legis unreachable — fail closed without