diff --git a/src/filigree/core.py b/src/filigree/core.py index ecfd2626..e6828e59 100644 --- a/src/filigree/core.py +++ b/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() diff --git a/src/filigree/db_base.py b/src/filigree/db_base.py index 7f782f02..72bef1e1 100644 --- a/src/filigree/db_base.py +++ b/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 --------------------------------------------------- diff --git a/src/filigree/db_entity_associations.py b/src/filigree/db_entity_associations.py index f33ae437..f890e73b 100644 --- a/src/filigree/db_entity_associations.py +++ b/src/filigree/db_entity_associations.py @@ -399,7 +399,12 @@ 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 @@ -407,6 +412,16 @@ def list_entity_associations(self, issue_id: IssueId) -> list[EntityAssociationR ``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 diff --git a/src/filigree/governance.py b/src/filigree/governance.py index a8e6b3a2..96e7b02d 100644 --- a/src/filigree/governance.py +++ b/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 diff --git a/src/filigree/registry.py b/src/filigree/registry.py index dc16937d..5127e329 100644 --- a/src/filigree/registry.py +++ b/src/filigree/registry.py @@ -41,7 +41,7 @@ from dataclasses import KW_ONLY, dataclass from dataclasses import field as dataclass_field from typing import Any, Literal, Protocol, TypeAlias, TypedDict -from urllib.parse import urlencode, urlparse +from urllib.parse import quote, urlencode, urlparse import httpx @@ -60,6 +60,13 @@ LOOMWEAVE_RESOLVE_FILE_MAX_ATTEMPTS = 3 LOOMWEAVE_RESOLVE_FILE_RETRY_BACKOFF_SECONDS = 0.05 +# The one sanctioned peek at Loomweave's otherwise-opaque entity-id grammar: the +# frozen Stable Entity Identity prefix (ADR-038). An id carrying it is an SEI and +# resolves through the by-SEI endpoint; anything else is a locator and resolves +# through the locator batch. Mirrors ``sei_backfill.SEI_PREFIX`` (kept here too so +# the registry layer does not import the operator-command module). +LOOMWEAVE_SEI_PREFIX = "loomweave:eid:" + # Loomweave's `_capabilities` response declares an `api_version: u8`. Filigree # rejects startup under `loomweave` mode if Loomweave advertises a version this # build was not written against — a mismatch means the wire contract changed @@ -188,6 +195,27 @@ class SeiResolution(TypedDict): LOOMWEAVE_BATCH_MAX_QUERIES = 256 +class EntityHashResolution(TypedDict): + """Outcome of resolving opaque entity ids to their CURRENT content_hash. + + The drift-comparison input for the closure gate (ADR-029 Decision 3: drift + detection is the consumer's job; Filigree owns the current-code-vs-attach + comparison per the 2026-06-29 hub ruling). + + - ``resolved`` — keyed by the *submitted* entity id, value is the entity's + current Loomweave ``content_hash`` (only ALIVE entities appear here). + - ``unresolved`` — entity ids Loomweave could not vouch for *while reachable*: + orphaned (``alive:false``), not_found, or rejected as invalid. The caller + treats these as freshness UNKNOWN — never silently fresh, never a hard + block (the enrich-only invariant). A whole-backend availability failure is + NOT reported here; it raises :class:`RegistryUnavailableError` so the + caller can degrade with a distinct, discriminated reason. + """ + + resolved: dict[str, str] + unresolved: list[str] + + def resolve_files_batch_via_loop( registry: object, queries: list[BatchQuery], @@ -344,6 +372,16 @@ def loomweave_identity_resolve_batch_url(base_url: str) -> str: return f"{base_url.rstrip('/')}/api/v1/identity/resolve:batch" +def loomweave_identity_sei_url(base_url: str, sei: str) -> str: + """Build the Loomweave by-SEI resolve URL (``GET /api/v1/identity/sei/{sei}``). + + The SEI carries reserved ``:`` separators, so the segment is percent-encoded + (Loomweave's axum router decodes it back). The endpoint returns the entity's + current ``content_hash`` for an alive SEI — the by-SEI counterpart to the + locator-keyed ``resolve:batch`` surface (ADR-038).""" + return f"{base_url.rstrip('/')}/api/v1/identity/sei/{quote(sei, safe='')}" + + def _is_loopback_origin(url: str) -> bool: host = urlparse(url).hostname if host is None: @@ -1052,6 +1090,161 @@ def record_outcome(locator: str, channel: str) -> None: return SeiResolution(resolved=resolved, orphaned=orphaned, already_migrated=already_migrated) + def resolve_entity_content_hashes(self, entity_ids: list[str]) -> EntityHashResolution: + """Resolve opaque entity ids to their CURRENT Loomweave ``content_hash``. + + The closure gate's drift-comparison input. Filigree treats the entity id + as opaque except for the one sanctioned peek — the frozen SEI prefix — + which selects the resolution endpoint (mirrors ``sei_backfill``): + + - ``loomweave:eid:`` SEIs → ``GET /api/v1/identity/sei/{sei}`` + (``resolve_sei``), which returns ``content_hash`` for an alive SEI. + - everything else (locators) → ``POST /api/v1/identity/resolve:batch``, + which returns ``content_hash`` per alive locator and routes SEI-shaped + / malformed inputs to its own ``invalid`` channel. + + Both endpoints carry ``content_hash`` only for an ALIVE entity; an + orphaned / not_found / invalid id is omitted from ``resolved`` and listed + in ``unresolved`` (UNKNOWN to the caller, never silently fresh). + Whole-backend availability failures (network / timeout / 5xx / auth / + malformed body) raise :class:`RegistryUnavailableError`; the caller + degrades to UNKNOWN without hard-blocking (enrich-only). + """ + resolved: dict[str, str] = {} + unresolved: list[str] = [] + # Preserve first-seen order, drop duplicates, partition by id scheme. + seen: set[str] = set() + locators: list[str] = [] + seis: list[str] = [] + for eid in entity_ids: + if not eid or eid in seen: + continue + seen.add(eid) + (seis if eid.startswith(LOOMWEAVE_SEI_PREFIX) else locators).append(eid) + + if locators: + locator_hashes = self._resolve_locator_content_hashes(locators) + for loc in locators: + current = locator_hashes.get(loc) + if current is not None: + resolved[loc] = current + else: + unresolved.append(loc) + for sei in seis: + current = self._resolve_sei_content_hash(sei) + if current is not None: + resolved[sei] = current + else: + unresolved.append(sei) + return EntityHashResolution(resolved=resolved, unresolved=unresolved) + + def _resolve_locator_content_hashes(self, locators: list[str]) -> dict[str, str]: + """Resolve locators → current ``content_hash`` via ``resolve:batch``. + + Returns a map of submitted locator → current content_hash for ALIVE + locators only; a locator Loomweave reports in ``not_found`` / ``invalid`` + (or as a non-alive resolved record) is simply absent from the map. Chunks + at the 256 per-batch cap like the sibling resolve paths. + """ + out: dict[str, str] = {} + url = loomweave_identity_resolve_batch_url(self.base_url) + for start in range(0, len(locators), LOOMWEAVE_BATCH_MAX_QUERIES): + chunk = locators[start : start + LOOMWEAVE_BATCH_MAX_QUERIES] + payload = self._request_identity_json("POST", url, body={"locators": chunk}) + resolved = payload.get("resolved") + if not isinstance(resolved, dict): + msg = f"Loomweave identity resolve at {url}: 'resolved' must be an object" + raise RegistryUnavailableError(msg, url=url, path="", cause_kind="invalid_response") + for locator, record in resolved.items(): + if not isinstance(record, dict) or record.get("alive") is not True: + continue + content_hash = record.get("content_hash") + if isinstance(content_hash, str) and content_hash: + out[locator] = content_hash + return out + + def _resolve_sei_content_hash(self, sei: str) -> str | None: + """Resolve a single SEI → current ``content_hash`` via ``GET .../sei/{sei}``. + + Returns the content_hash for an alive SEI, or ``None`` when Loomweave + reports it ``alive:false`` (orphaned) — there is no batch by-SEI surface, + so this is a per-SEI read (governed bindings per issue are few). + + FEDERATION-CONTRACT GAP (flag for the hub): Loomweave's *implementation* + (``http_read/identity.rs::get_identity_sei``) returns ``content_hash`` on + the by-SEI response today, but the blessed ``weft-sei-conformance-oracle`` + only documents it on the by-*locator* shape (``resolve_locator``), NOT on + ``resolve_sei``. So the locator path rests on a blessed field while the + SEI path rests on an un-blessed implementation detail. A future + "conform Loomweave to the oracle" pass could legitimately trim + content_hash from this response, at which point SEI-form drift detection + degrades to UNKNOWN (enrich-only — fails safe, never silently fresh). The + durable fix is to bless content_hash on ``resolve_sei`` in the oracle; + until then this dependency is intentional and self-degrading.""" + url = loomweave_identity_sei_url(self.base_url, sei) + payload = self._request_identity_json("GET", url) + if payload.get("alive") is True: + content_hash = payload.get("content_hash") + if isinstance(content_hash, str) and content_hash: + return content_hash + return None + + def _request_identity_json(self, method: str, url: str, *, body: dict[str, Any] | None = None) -> dict[str, Any]: + """Issue an identity-resolve request and return its parsed JSON object. + + Shares the deadline / retry / backoff budget of the other read paths + (transient 5xx and network failures retry; auth and other non-2xx are + deterministic and raise immediately). Every failure mode — connectivity, + non-2xx, malformed body — raises :class:`RegistryUnavailableError` so the + single drift-resolution consumer has one exception to degrade on. + """ + has_body = body is not None + deadline = time.monotonic() + self.timeout_seconds + attempt = 1 + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + msg = f"Loomweave identity request unreachable at {url}: retry budget exhausted" + raise RegistryUnavailableError(msg, url=url, path="", cause_kind="timeout") + try: + response = self._http_client.request( + method, + url, + json=body if has_body else None, + headers=self._headers_for_request(url, has_body=has_body), + timeout=remaining, + ) + raw = response.text + if response.status_code >= 400: + if response.status_code >= 500 and self._should_retry_read(attempt, deadline): + self._log_retry(url=url, attempt=attempt, cause_kind="http_error") + self._sleep_before_retry(deadline) + attempt += 1 + continue + reason = response.reason_phrase + cause_kind = "auth" if response.status_code == 401 else "http_error" + msg = f"Loomweave identity request rejected at {url}: HTTP {response.status_code} {reason}" + raise RegistryUnavailableError(msg, url=url, path="", cause_kind=cause_kind) + break + except (httpx.TimeoutException, httpx.TransportError) as exc: + if self._should_retry_read(attempt, deadline): + self._log_retry(url=url, attempt=attempt, cause_kind="network") + self._sleep_before_retry(deadline) + attempt += 1 + continue + msg = f"Loomweave identity request unreachable at {url}: {exc}" + raise RegistryUnavailableError(msg, url=url, path="", cause_kind="network") from exc + + try: + payload = json.loads(raw) + except json.JSONDecodeError as exc: + msg = f"Loomweave identity request returned invalid JSON from {url}: {exc}" + raise RegistryUnavailableError(msg, url=url, path="", cause_kind="invalid_response") from exc + if not isinstance(payload, dict): + msg = f"Loomweave identity request returned non-object response from {url}: {type(payload).__name__}" + raise RegistryUnavailableError(msg, url=url, path="", cause_kind="invalid_response") + return payload + def is_displaced(self) -> bool: return True diff --git a/tests/core/test_entity_associations.py b/tests/core/test_entity_associations.py index eab7f9b5..91deda49 100644 --- a/tests/core/test_entity_associations.py +++ b/tests/core/test_entity_associations.py @@ -429,6 +429,31 @@ def test_forward_list_omits_lifecycle_facts(self, db: FiligreeDB) -> None: assert "claim_commit" not in fwd assert "close_commit" not in fwd + def test_forward_list_freshness_defaults_unknown(self, db: FiligreeDB) -> None: + """Without supplied current hashes the forward list keeps the ``unknown`` + sentinel — the data layer never resolves drift itself (no network).""" + issue = db.create_issue("freshness default", priority=2) + db.add_entity_association(issue.id, "py:func:fresh-target", content_hash="h1", actor="alice") + (fwd,) = db.list_entity_associations(issue.id) + assert fwd["freshness_status"] == "unknown" + + def test_forward_list_freshness_from_supplied_hashes(self, db: FiligreeDB) -> None: + """Bonus (RED-1): when a caller supplies resolved current hashes, the + forward list populates ``freshness_status`` per row (fresh / stale), + mirroring the reverse-lookup consumer-supplies-hashes convention.""" + issue = db.create_issue("freshness supplied", priority=2) + db.add_entity_association(issue.id, "py:func:fresh", content_hash="h1", actor="alice") + db.add_entity_association(issue.id, "py:func:drifted", content_hash="h1", actor="alice") + rows = { + r["loomweave_entity_id"]: r + for r in db.list_entity_associations( + issue.id, + current_content_hashes={"py:func:fresh": "h1", "py:func:drifted": "h2"}, + ) + } + assert rows["py:func:fresh"]["freshness_status"] == "fresh" + assert rows["py:func:drifted"]["freshness_status"] == "stale" + def test_closed_with_commit_row_exposes_close_commit(self, db: FiligreeDB) -> None: """The reverse row carries the close commit anchor (warpline correlates 'changed since closed' on the COMMIT, not the clock).""" diff --git a/tests/core/test_governance_gate.py b/tests/core/test_governance_gate.py index b7f09cb5..423b8287 100644 --- a/tests/core/test_governance_gate.py +++ b/tests/core/test_governance_gate.py @@ -386,3 +386,151 @@ def _record(issue_id: str) -> LegisGateResult: db.add_entity_association(issue.id, "sei:x", content_hash="h2", actor="legis", signature="sig2", signoff_seq=2) assert governance.evaluate_closure_gate(db, issue.id).outcome is GateOutcome.PROCEED assert spy == [issue.id] + + +# --- RED-1: current-code-vs-attach drift (Filigree owns the comparison) -------- +# The snapshot-STALE check above 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) signed_content_hash, _signed_row_is_stale is False, and the +# close was waved through. The gate now resolves each governed binding's CURRENT +# content_hash via the Loomweave registry consumer and fails closed as STALE on a +# mismatch. The resolution is enrich-only: a Loomweave outage degrades to a +# discriminated freshness UNKNOWN and never hard-blocks the close. + + +class _FakeRegistry: + """Mirrors ``registry.resolve_entity_content_hashes``: returns the current + content_hash for ids it knows, lists the rest as ``unresolved`` (the orphan / + not_found / invalid degrade), or raises ``RegistryUnavailableError`` to + simulate a whole-backend Loomweave outage.""" + + def __init__(self, hashes: dict[str, str], *, raise_unavailable: bool = False) -> None: + self._hashes = hashes + self._raise_unavailable = raise_unavailable + self.calls: list[list[str]] = [] + + def resolve_entity_content_hashes(self, entity_ids: list[str]) -> dict[str, object]: + self.calls.append(list(entity_ids)) + if self._raise_unavailable: + from filigree.registry import RegistryUnavailableError + + raise RegistryUnavailableError("loomweave down", url="http://legis.test", cause_kind="network") + resolved = {eid: self._hashes[eid] for eid in entity_ids if eid in self._hashes} + unresolved = [eid for eid in entity_ids if eid not in self._hashes] + return {"resolved": resolved, "unresolved": unresolved} + + +class _FakeDBWithRegistry(_FakeDB): + """``_FakeDB`` plus a ``.registry`` exposing the entity-hash resolver.""" + + def __init__(self, rows: list[dict[str, object]], registry: object) -> None: + super().__init__(rows) + self.registry = registry + + +def _governed_rows_attached_at(entity_id: str, attach_hash: str) -> list[dict[str, object]]: + # Sign-off snapshot is FRESH (signed == attach): the v27 snapshot check does + # NOT fire, so any STALE verdict here is the new current-code drift check. + return [ + { + "loomweave_entity_id": entity_id, + "signature": "deadbeef", + "signoff_seq": 1, + "content_hash_at_attach": attach_hash, + "signed_content_hash": attach_hash, + } + ] + + +def test_current_code_drift_fails_closed_as_stale_without_legis(monkeypatch: pytest.MonkeyPatch) -> None: + """(a) Current code moved on (h1 at attach, registry reports h2) -> STALE, + no Legis call. Uses an SEI-form entity id to prove SEI bindings ARE checked + (not silently degraded).""" + monkeypatch.setenv(legis_client.LEGIS_URL_ENV, "http://legis.test") + entity_id = "loomweave:eid:00000000000000000000000000000001" + registry = _FakeRegistry({entity_id: "h2"}) + db = _FakeDBWithRegistry(_governed_rows_attached_at(entity_id, "h1"), registry) + spy: list[str] = [] + monkeypatch.setattr(governance, "check_closure_gate", lambda iid: spy.append(iid)) + decision = governance.evaluate_closure_gate(db, "test-1") + assert decision.outcome is GateOutcome.STALE + assert "drifted since attach" in decision.reason + assert registry.calls == [[entity_id]] # current hash was resolved + assert spy == [] # fail closed locally, no Legis consultation + + +def test_current_code_match_proceeds_to_legis(monkeypatch: pytest.MonkeyPatch) -> None: + """(b) Current content still matches the attach snapshot -> no drift block; + the close proceeds through the normal Legis gate (ALLOWED -> PROCEED).""" + _patch_gate(monkeypatch, LegisGateResult(LegisGateStatus.ALLOWED)) + entity_id = "py:func:mod::f" + registry = _FakeRegistry({entity_id: "h1"}) + db = _FakeDBWithRegistry(_governed_rows_attached_at(entity_id, "h1"), registry) + decision = governance.evaluate_closure_gate(db, "test-1") + assert decision.outcome is GateOutcome.PROCEED + assert registry.calls == [[entity_id]] + + +def test_loomweave_unavailable_degrades_to_unknown_not_blocked(monkeypatch: pytest.MonkeyPatch) -> None: + """(c) Loomweave unreachable -> discriminated UNKNOWN, the drift check does + NOT hard-block: the close still proceeds through the Legis gate (enrich-only, + core close not load-bearing on Loomweave).""" + _patch_gate(monkeypatch, LegisGateResult(LegisGateStatus.ALLOWED)) + entity_id = "py:func:mod::f" + registry = _FakeRegistry({entity_id: "h2"}, raise_unavailable=True) + db = _FakeDBWithRegistry(_governed_rows_attached_at(entity_id, "h1"), registry) + decision = governance.evaluate_closure_gate(db, "test-1") + assert decision.outcome is GateOutcome.PROCEED # NOT STALE, NOT blocked + assert registry.calls == [[entity_id]] # drift resolution was attempted + + +def test_entity_unresolved_degrades_to_unknown_not_blocked(monkeypatch: pytest.MonkeyPatch) -> None: + """Loomweave reachable but the entity is orphaned/not_found (absent from + ``resolved``) -> UNKNOWN, not a block: proceeds to the Legis gate. Distinct + from a drift (which we DO know about) and from an outage.""" + _patch_gate(monkeypatch, LegisGateResult(LegisGateStatus.ALLOWED)) + entity_id = "py:func:mod::gone" + registry = _FakeRegistry({}) # entity not in resolved -> unresolved + db = _FakeDBWithRegistry(_governed_rows_attached_at(entity_id, "h1"), registry) + decision = governance.evaluate_closure_gate(db, "test-1") + assert decision.outcome is GateOutcome.PROCEED + assert registry.calls == [[entity_id]] + + +def test_ungoverned_close_never_resolves_drift(monkeypatch: pytest.MonkeyPatch) -> None: + """(d) Ungoverned close is unchanged: no signature -> PROCEED before any + drift resolution; the registry is never consulted.""" + monkeypatch.setenv(legis_client.LEGIS_URL_ENV, "http://legis.test") + registry = _FakeRegistry({"py:func:mod::f": "h2"}) + db = _FakeDBWithRegistry(_ungoverned_rows(), registry) + spy: list[str] = [] + monkeypatch.setattr(governance, "check_closure_gate", lambda iid: spy.append(iid)) + decision = governance.evaluate_closure_gate(db, "test-1") + assert decision.outcome is GateOutcome.PROCEED + assert registry.calls == [] # ungoverned short-circuit precedes drift resolution + assert spy == [] + + +def test_drift_wins_over_unknown_when_mixed(monkeypatch: pytest.MonkeyPatch) -> None: + """One binding drifted + one unresolvable -> STALE: a known drift on any + governed binding fails the close closed regardless of an UNKNOWN sibling.""" + monkeypatch.setenv(legis_client.LEGIS_URL_ENV, "http://legis.test") + rows = _governed_rows_attached_at("py:func:mod::f", "h1") + _governed_rows_attached_at("py:func:mod::g", "h1") + registry = _FakeRegistry({"py:func:mod::f": "h2"}) # f drifted, g unresolved + db = _FakeDBWithRegistry(rows, registry) + spy: list[str] = [] + monkeypatch.setattr(governance, "check_closure_gate", lambda iid: spy.append(iid)) + decision = governance.evaluate_closure_gate(db, "test-1") + assert decision.outcome is GateOutcome.STALE + assert spy == [] + + +def test_no_registry_attribute_degrades_to_unknown(monkeypatch: pytest.MonkeyPatch) -> None: + """A db with no ``.registry`` (local mode / bare fake) cannot resolve drift -> + UNKNOWN, proceeds to Legis. Pins that the new check is a no-op for the + registry-less _FakeDB the rest of this module relies on.""" + _patch_gate(monkeypatch, LegisGateResult(LegisGateStatus.ALLOWED)) + db = _FakeDB(_governed_rows_attached_at("py:func:mod::f", "h1")) + decision = governance.evaluate_closure_gate(db, "test-1") + assert decision.outcome is GateOutcome.PROCEED diff --git a/tests/unit/test_entity_content_hash_resolve.py b/tests/unit/test_entity_content_hash_resolve.py new file mode 100644 index 00000000..aa4dfb86 --- /dev/null +++ b/tests/unit/test_entity_content_hash_resolve.py @@ -0,0 +1,200 @@ +"""Wire-level tests for ``LoomweaveRegistry.resolve_entity_content_hashes``. + +RED-1: the closure gate resolves each governed binding's CURRENT content_hash +through this surface and compares it to the attach snapshot. These tests run a +throwaway HTTP server that mirrors Loomweave's REAL identity-resolve wire shapes +(``crates/loomweave-cli/src/http_read/identity.rs``) so the form-dispatch and the +``content_hash`` extraction are exercised against the production response shape — +NOT a convenient fake that would return a hash for any id and hide a false-green: + +- locators -> ``POST /api/v1/identity/resolve:batch`` -> ``resolved[locator] = + {sei, current_locator, content_hash, alive:true}``; SEI-shaped inputs land in + ``invalid``, unknown valid locators in ``not_found``. +- SEIs -> ``GET /api/v1/identity/sei/{sei}`` -> ``{sei, current_locator, + content_hash, alive:true}`` for an alive SEI, or ``{alive:false, lineage:[]}``. +""" + +from __future__ import annotations + +import json +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import unquote, urlparse + +import pytest + +from filigree.registry import LoomweaveRegistry, RegistryUnavailableError + +# Loomweave's identity DB, keyed by the form the consumer submits. +_LOCATOR_HASHES = { + "py:func:mod::f": "sha256:current-f", + "core:file:abc@src/x.py": "sha256:current-x", +} +_SEI_HASHES = { + "loomweave:eid:00000000000000000000000000000001": "sha256:current-sei", +} +# An SEI that resolves alive:false (orphaned / renamed away). +_ORPHANED_SEI = "loomweave:eid:0000000000000000000000000000dead" + + +class _IdentityHandler(BaseHTTPRequestHandler): + """Stub of Loomweave's identity-resolve endpoints (alive records only).""" + + def _send(self, payload: dict[str, object]) -> None: + body = json.dumps(payload).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_POST(self) -> None: + length = int(self.headers.get("Content-Length", "0")) + request = json.loads(self.rfile.read(length) or b"{}") + resolved: dict[str, object] = {} + invalid: list[str] = [] + not_found: list[str] = [] + for locator in request.get("locators", []): + if locator.startswith("loomweave:eid:"): + invalid.append(locator) # resolve:batch rejects SEIs + elif locator in _LOCATOR_HASHES: + resolved[locator] = { + "sei": "loomweave:eid:resolved", + "current_locator": locator, + "content_hash": _LOCATOR_HASHES[locator], + "alive": True, + } + else: + not_found.append(locator) + self._send({"resolved": resolved, "invalid": invalid, "not_found": not_found}) + + def do_GET(self) -> None: + # Path: /api/v1/identity/sei/ + sei = unquote(urlparse(self.path).path.rsplit("/", 1)[-1]) + if sei in _SEI_HASHES: + self._send( + { + "sei": sei, + "current_locator": "py:func:mod::renamed", + "content_hash": _SEI_HASHES[sei], + "alive": True, + } + ) + else: + self._send({"sei": sei, "alive": False, "lineage": []}) + + def log_message(self, format: str, *args: object) -> None: + return + + +@pytest.fixture +def identity_registry() -> object: + server = ThreadingHTTPServer(("127.0.0.1", 0), _IdentityHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + registry = LoomweaveRegistry(f"http://127.0.0.1:{server.server_port}", timeout_seconds=2) + try: + yield registry + finally: + registry.close() + server.shutdown() + server.server_close() + thread.join(timeout=2) + + +def test_resolves_locator_content_hash_via_batch(identity_registry: LoomweaveRegistry) -> None: + result = identity_registry.resolve_entity_content_hashes(["py:func:mod::f"]) + assert result["resolved"] == {"py:func:mod::f": "sha256:current-f"} + assert result["unresolved"] == [] + + +def test_resolves_sei_content_hash_via_by_sei_get(identity_registry: LoomweaveRegistry) -> None: + sei = "loomweave:eid:00000000000000000000000000000001" + result = identity_registry.resolve_entity_content_hashes([sei]) + assert result["resolved"] == {sei: "sha256:current-sei"} + assert result["unresolved"] == [] + + +def test_resolves_mixed_locator_and_sei_forms(identity_registry: LoomweaveRegistry) -> None: + sei = "loomweave:eid:00000000000000000000000000000001" + result = identity_registry.resolve_entity_content_hashes(["py:func:mod::f", sei, "core:file:abc@src/x.py"]) + assert result["resolved"] == { + "py:func:mod::f": "sha256:current-f", + "core:file:abc@src/x.py": "sha256:current-x", + sei: "sha256:current-sei", + } + assert result["unresolved"] == [] + + +def test_orphaned_sei_is_unresolved_not_fresh(identity_registry: LoomweaveRegistry) -> None: + result = identity_registry.resolve_entity_content_hashes([_ORPHANED_SEI]) + assert result["resolved"] == {} + assert result["unresolved"] == [_ORPHANED_SEI] + + +def test_unknown_locator_is_unresolved(identity_registry: LoomweaveRegistry) -> None: + result = identity_registry.resolve_entity_content_hashes(["py:func:mod::missing"]) + assert result["resolved"] == {} + assert result["unresolved"] == ["py:func:mod::missing"] + + +def test_backend_unreachable_raises_registry_unavailable() -> None: + # Bind a port, close it -> connection refused -> whole-backend availability + # failure surfaces as RegistryUnavailableError (the gate degrades to UNKNOWN). + registry = LoomweaveRegistry("http://127.0.0.1:1", timeout_seconds=1) + try: + with pytest.raises(RegistryUnavailableError): + registry.resolve_entity_content_hashes(["py:func:mod::f"]) + finally: + registry.close() + + +# --- fallback wrapper delegation --------------------------------------------- +# In loomweave mode with allow_local_fallback, ``db.registry`` is the +# _LoomweaveLocalFallbackRegistry wrapper. If it did NOT expose +# resolve_entity_content_hashes, the gate's getattr would miss it and degrade to +# UNKNOWN even when Loomweave is UP — a false-green that disables the drift gate +# whenever fallback is enabled. These pin the delegation. + + +class _PrimaryWithResolver: + def __init__(self, resolution: dict[str, object]) -> None: + self._resolution = resolution + + def resolve_entity_content_hashes(self, entity_ids: list[str]) -> dict[str, object]: + return self._resolution + + def is_displaced(self) -> bool: + return True + + +class _LegacyPrimaryNoResolver: + """A pre-surface injected primary (e.g. an older fake) lacking the method.""" + + def is_displaced(self) -> bool: + return True + + +def test_fallback_wrapper_delegates_to_primary() -> None: + from filigree.core import LocalRegistry, _LoomweaveLocalFallbackRegistry + + resolution = {"resolved": {"py:func:mod::f": "sha256:x"}, "unresolved": []} + wrapper = _LoomweaveLocalFallbackRegistry( + _PrimaryWithResolver(resolution), + LocalRegistry(lambda: "f-local"), + base_url="http://loomweave.test", + ) + assert wrapper.resolve_entity_content_hashes(["py:func:mod::f"]) == resolution + + +def test_fallback_wrapper_without_primary_surface_degrades_to_unresolved() -> None: + from filigree.core import LocalRegistry, _LoomweaveLocalFallbackRegistry + + wrapper = _LoomweaveLocalFallbackRegistry( + _LegacyPrimaryNoResolver(), + LocalRegistry(lambda: "f-local"), + base_url="http://loomweave.test", + ) + result = wrapper.resolve_entity_content_hashes(["a", "b"]) + assert result["resolved"] == {} + assert result["unresolved"] == ["a", "b"]