Merge pull request #2050 from Open-Source-Legal/claude/upbeat-euler-wvsuth

Address issue #2020 authority-frontier code review findings

Author: JSv4

changelog.d/2020-authority-frontier-review.changed.md

@@ -0,0 +1,4 @@
+- Moved the `AuthorityFrontier` discovery-state vocabulary into `opencontractserver/enrichment/constants.py` as a single source of truth (`DISCOVERY_STATE_*` constants, `DISCOVERY_STATE_CHOICES`, and `DISCOVERY_SUCCESS_STATES`). The model field (`opencontractserver/annotations/models.py`), the frontier service transition verbs, the discovery orchestrator, the crawl driver, and the verify/license gate's `GATE_*` verdicts now reference these constants instead of bare string literals, so a rename is a one-line edit and the parallel definitions cannot drift (CLAUDE.md item 4; issue #2020 finding 7). The model's `DISCOVERY_STATE_CHOICES` is re-exported from the constants with identical values, so no migration is generated.
+- Renamed `enrichment.constants._USC_PREFIX_RE` / `_CFR_PREFIX_RE` to `USC_PREFIX_RE` / `CFR_PREFIX_RE` (no leading underscore). They are imported cross-module by the USC/CFR authority source providers' `can_handle` overrides, so the underscore (signalling module-private) was misleading; updated both providers' imports (issue #2020 finding 3).
+- Optimised `AuthorityDiscoveryService._provider_for` (`opencontractserver/enrichment/services/authority_discovery_service.py`) to instantiate each enabled provider once per call and reuse the instances across every candidate key, instead of re-instantiating every provider for each candidate. `can_handle` is pure given the key, so the prior per-candidate instantiation was wasted work that grew with the candidate-key fan-out (issue #2020 finding 6).
+- Expanded the `dequeue_for_provider` docstring to document its crash-recovery purpose (freshly-seeded rows carry `provider=None` and are intentionally skipped; the primary dispatch path is the provider-agnostic `dequeue_queued`) and clarified the `AuthorityKeyEquivalence` model docstring that its direction convention is documentation-only because callers query both columns (issue #2020 findings 9, 10).

changelog.d/2020-authority-frontier-review.fixed.md

@@ -0,0 +1,4 @@
+- Fixed a TOCTOU race in `AuthorityFrontierService.seed_from_wanted_authorities` (`opencontractserver/enrichment/services/authority_frontier_service.py`): the per-key create-or-refresh now runs inside a single `transaction.atomic()` + `select_for_update()` critical section instead of `get_or_create` followed by an unconditional `save`. Two concurrent seed passes could previously both clear `get_or_create` and have one silently clobber the other's count update, and a freshly created row was briefly visible at `mention_count=0` (the field default) between the insert and its follow-up save. The lock serialises the second writer and seeds the real counts atomically. The "prefer existing non-null" semantics for `jurisdiction`/`authority_type` are preserved (issue #2020 finding 1).
+- Fixed stale `last_error` on a successful retry in `AuthorityFrontierService.mark()` (same file): a row transitioning `failed (error=...) -> ingested` previously retained the old error string, so downstream health checks reading `last_error` misread a healthy row as broken. `mark()` now clears `last_error` implicitly when transitioning into a terminal success state (`enrichment.constants.DISCOVERY_SUCCESS_STATES`, i.e. `ingested`); the failure history is still preserved in the append-only `candidate_sources` audit trail (issue #2020 finding 2).
+- Fixed a double XML traversal in `CFRAuthoritySourceProvider._fetch_impl` (`opencontractserver/pipeline/authority_source_providers/cfr_provider.py`): the `<P>` flatten list comprehension called `_flatten_element_text(p_el)` twice per element (once for the value, once for the truthiness filter). A walrus binds the result once so each `<P>` is traversed a single time (issue #2020 finding 5).
+- Added regression tests in `opencontractserver/tests/test_authority_frontier.py`: the seed upsert takes a `SELECT ... FOR UPDATE` row lock, a `failed -> ingested` transition clears the stale error (while a non-success transition preserves it), and `dequeue_for_provider` excludes freshly-seeded `provider=None` rows (issue #2020 findings 11, 12).

opencontractserver/annotations/models.py

@@ -2187,28 +2187,18 @@ class AuthorityFrontier(django.db.models.Model):
     # ("top_detection_tier" is one of enrichment_constants.ALL_DETECTION_TIERS).
     candidate_sources = django.db.models.JSONField(default=list, blank=True)
 
-    # Discovery state machine. NOTE: the historical ``discovered`` and
-    # ``resolved`` states were retired (Authority Console Phase 4): no production
-    # code path ever assigned them (discovery jumps in_progress -> ingested, and
-    # the resolution outcome lives on the relink result / Analysis, not the
-    # frontier row), so carrying them as choices was a dead-vocabulary trap.
-    DISCOVERY_STATE_CHOICES = [
-        ("queued", "Queued"),
-        ("in_progress", "In progress"),
-        ("ingested", "Document imported"),
-        ("failed", "No source found"),
-        ("unsupported", "No provider can_handle"),
-        # Phase 4: visible, non-silent gate outcomes
-        ("blocked_license", "Provider license is not public-domain"),
-        ("blocked_domain", "Source domain not on the public-domain allowlist"),
-        ("unlocated", "Located text did not verify against the requested key"),
-        ("pending_approval", "Found, awaiting human approval before ingest"),
-        # Phase 5: per-jurisdiction cap reached; row parked so dequeue can skip it
-        ("deferred_cap", "Deferred: per-jurisdiction cap reached"),
-    ]
+    # Discovery state machine. The state vocabulary + labels are the single
+    # source of truth in ``enrichment.constants`` (CLAUDE.md item 4: no magic
+    # strings) so the model, the frontier service transition verbs, the
+    # discovery orchestrator, and the gate cannot drift. Re-exported here as a
+    # class attribute so existing references (e.g.
+    # config/graphql/annotation_types.py) keep resolving
+    # ``AuthorityFrontier.DISCOVERY_STATE_CHOICES``. See the constants module for
+    # the note on the retired ``discovered``/``resolved`` states.
+    DISCOVERY_STATE_CHOICES = enrichment_constants.DISCOVERY_STATE_CHOICES
     discovery_state = django.db.models.CharField(
         max_length=32,
-        default="queued",
+        default=enrichment_constants.DISCOVERY_STATE_QUEUED,
         db_index=True,
         choices=DISCOVERY_STATE_CHOICES,
     )
@@ -2261,6 +2251,11 @@ class AuthorityKeyEquivalence(django.db.models.Model):
     that they denote the same authority so find_authority_target can hop across
     namespaces. Plain Model: derived reference data (populated from USLM
     @identifier + <sourceCredit> <ref>s), never user content.
+
+    Direction convention (``from_key`` = popular-name act citation, ``to_key`` =
+    USC codification) is for documentation only: every caller queries both
+    columns, so equivalence is treated as symmetric and a developer reading the
+    model in isolation should not assume lookups are one-directional.
     """
 
     # Single-column indexes are declared once in Meta.indexes below (a bare

opencontractserver/enrichment/constants.py

@@ -215,8 +215,11 @@ def doc_max_concurrency() -> int:
 
 
 # --- Phase 3: prefix classifier ---------------------------------------- #
-_USC_PREFIX_RE = _re.compile(r"^usc-\d+$")
-_CFR_PREFIX_RE = _re.compile(r"^cfr-\d+$")
+# Public (no leading underscore): these are imported cross-module by the USC /
+# CFR authority source providers' ``can_handle`` overrides, so they are part of
+# the package's intentional surface, not module-private helpers.
+USC_PREFIX_RE = _re.compile(r"^usc-\d+$")
+CFR_PREFIX_RE = _re.compile(r"^cfr-\d+$")
 # Municipal grammar keys (issue #1995): the ``muni`` catch-all (bare "Municipal
 # Code § N") and per-city ``muni-<city-slug>`` keys (both the table-keyed codes
 # and open-vocab city captures). Matched by shape so a city added to the table
@@ -237,6 +240,59 @@ def doc_max_concurrency() -> int:
 GRAMMAR_STATUTE_META_PREFIXES = frozenset({"act", "publ", "stat"})
 
 
+# --- Phase 3/5: AuthorityFrontier discovery-state machine ------------------ #
+# Single source of truth for the frontier state vocabulary (CLAUDE.md item 4:
+# no magic strings). The model field choices (annotations.models.AuthorityFrontier
+# .DISCOVERY_STATE_CHOICES), the transition primitive (AuthorityFrontierService
+# .mark), the discovery orchestrator, and the crawl driver all reference these
+# names so a rename is a one-line edit. The verify+license gate
+# (AuthorityGateService) reuses the overlapping subset for its GATE_* verdicts.
+DISCOVERY_STATE_QUEUED = "queued"
+DISCOVERY_STATE_IN_PROGRESS = "in_progress"
+DISCOVERY_STATE_INGESTED = "ingested"
+DISCOVERY_STATE_FAILED = "failed"
+DISCOVERY_STATE_UNSUPPORTED = "unsupported"
+# Phase 4: visible, non-silent gate outcomes.
+DISCOVERY_STATE_BLOCKED_LICENSE = "blocked_license"
+DISCOVERY_STATE_BLOCKED_DOMAIN = "blocked_domain"
+DISCOVERY_STATE_UNLOCATED = "unlocated"
+DISCOVERY_STATE_PENDING_APPROVAL = "pending_approval"
+# Phase 5: per-jurisdiction cap reached; row parked so dequeue can skip it.
+DISCOVERY_STATE_DEFERRED_CAP = "deferred_cap"
+
+# (value, human label) pairs for the model field. The labels live with the
+# vocabulary so the model, admin, and any serializer share one definition.
+# NOTE: the historical ``discovered`` and ``resolved`` states were retired
+# (Authority Console Phase 4): no production code path ever assigned them
+# (discovery jumps in_progress -> ingested, and the resolution outcome lives on
+# the relink result / Analysis, not the frontier row), so carrying them as
+# choices was a dead-vocabulary trap.
+DISCOVERY_STATE_CHOICES = [
+    (DISCOVERY_STATE_QUEUED, "Queued"),
+    (DISCOVERY_STATE_IN_PROGRESS, "In progress"),
+    (DISCOVERY_STATE_INGESTED, "Document imported"),
+    (DISCOVERY_STATE_FAILED, "No source found"),
+    (DISCOVERY_STATE_UNSUPPORTED, "No provider can_handle"),
+    (DISCOVERY_STATE_BLOCKED_LICENSE, "Provider license is not public-domain"),
+    (
+        DISCOVERY_STATE_BLOCKED_DOMAIN,
+        "Source domain not on the public-domain allowlist",
+    ),
+    (
+        DISCOVERY_STATE_UNLOCATED,
+        "Located text did not verify against the requested key",
+    ),
+    (DISCOVERY_STATE_PENDING_APPROVAL, "Found, awaiting human approval before ingest"),
+    (DISCOVERY_STATE_DEFERRED_CAP, "Deferred: per-jurisdiction cap reached"),
+]
+
+# States that represent a successful terminal ingest. ``mark()`` clears
+# ``last_error`` when transitioning into one of these, so a healthy row never
+# retains a stale error string from an earlier failed attempt. Only "ingested"
+# qualifies today ("resolved" was retired — see the note above).
+DISCOVERY_SUCCESS_STATES = frozenset({DISCOVERY_STATE_INGESTED})
+
+
 def classify_prefix(prefix: str) -> tuple:
     """(jurisdiction, authority_type) for a canonical_key prefix.
 
@@ -249,9 +305,9 @@ def classify_prefix(prefix: str) -> tuple:
     Falls back to ``PREFIX_CLASSIFICATION`` for named registry bodies (dgcl,
     exchange-act, irc, …) and returns ``(None, None)`` for unknown prefixes.
     """
-    if _USC_PREFIX_RE.match(prefix):
+    if USC_PREFIX_RE.match(prefix):
         return (JURISDICTION_US_FEDERAL, AUTHORITY_TYPE_STATUTE)
-    if _CFR_PREFIX_RE.match(prefix):
+    if CFR_PREFIX_RE.match(prefix):
         return (JURISDICTION_US_FEDERAL, AUTHORITY_TYPE_REGULATION)
     if prefix == "fedreg":
         return (JURISDICTION_US_FEDERAL, AUTHORITY_TYPE_ADMIN_RULE)

opencontractserver/enrichment/services/authority_discovery_service.py

@@ -25,6 +25,7 @@
 from django.utils import timezone
 
 from opencontractserver.annotations.models import AuthorityFrontier
+from opencontractserver.enrichment import constants as C
 from opencontractserver.shared.services.base import BaseService
 
 logger = logging.getLogger(__name__)
@@ -83,15 +84,21 @@ def _provider_for(cls, canonical_key: str) -> tuple[str | None, Any, str | None]
             key=lambda d: getattr(d.component_class, "priority", 100),
         )
 
+        # Instantiate each enabled provider ONCE per call and reuse the instances
+        # across every candidate key below. ``can_handle`` is pure given the key,
+        # so re-instantiating a provider for each candidate (the prior behaviour)
+        # was wasted work that grew with the candidate-key fan-out.
+        providers = [
+            (defn.name, defn.component_class())
+            for defn in defns
+            if defn.component_class is not None
+            and getattr(defn.component_class, "enabled", True)
+        ]
+
         def _match(key: str):
-            for defn in defns:
-                if defn.component_class is None:
-                    continue
-                if not getattr(defn.component_class, "enabled", True):
-                    continue
-                provider = defn.component_class()
+            for name, provider in providers:
                 if provider.can_handle(key):
-                    return defn.name, provider
+                    return name, provider
             return None
 
         from opencontractserver.enrichment.data import mappings as _mappings
@@ -221,8 +228,11 @@ def discover_and_bootstrap(
         name, provider, fetch_key = cls._provider_for(canonical_key)
 
         if provider is None:
-            AuthorityFrontierService.mark(frontier_row, "unsupported")
-            return {"status": "unsupported", "canonical_key": canonical_key}
+            AuthorityFrontierService.mark(frontier_row, C.DISCOVERY_STATE_UNSUPPORTED)
+            return {
+                "status": C.DISCOVERY_STATE_UNSUPPORTED,
+                "canonical_key": canonical_key,
+            }
 
         # ``_provider_for`` only returns a non-None provider alongside a non-None
         # ``fetch_key`` (the matched candidate key); the ``(None, None, None)``
@@ -234,7 +244,7 @@ def discover_and_bootstrap(
         # Record which provider was selected and mark in-flight.
         frontier_row.provider = name
         frontier_row.save(update_fields=["provider", "modified"])
-        AuthorityFrontierService.mark(frontier_row, "in_progress")
+        AuthorityFrontierService.mark(frontier_row, C.DISCOVERY_STATE_IN_PROGRESS)
 
         # --- fetch -----------------------------------------------------------
         try:
@@ -257,17 +267,17 @@ def discover_and_bootstrap(
             candidate_record = cls._audit_record(
                 provider_name=name,
                 provider_license=provider.license,
-                outcome="failed",
+                outcome=C.DISCOVERY_STATE_FAILED,
                 error=str(exc),
             )
             AuthorityFrontierService.mark(
                 frontier_row,
-                "failed",
+                C.DISCOVERY_STATE_FAILED,
                 error=str(exc),
                 candidate_record=candidate_record,
             )
             return {
-                "status": "failed",
+                "status": C.DISCOVERY_STATE_FAILED,
                 "error": str(exc),
                 "canonical_key": canonical_key,
             }
@@ -298,7 +308,11 @@ def discover_and_bootstrap(
             provider_license=provider.license,
             source_domain=decision.source_domain,
             verify=decision.verify,
-            outcome=decision.verdict if decision.verdict != GATE_OK else "ingested",
+            outcome=(
+                decision.verdict
+                if decision.verdict != GATE_OK
+                else C.DISCOVERY_STATE_INGESTED
+            ),
             error=None if decision.verdict == GATE_OK else decision.reason,
         )
         if decision.verdict != GATE_OK:
@@ -378,9 +392,11 @@ def discover_and_bootstrap(
             result["equivalence_relink"] = relink_result
 
             # --- mark ingested -----------------------------------------------
+            # mark() clears any stale last_error from a prior failed attempt on
+            # this SUCCESS transition (see C.DISCOVERY_SUCCESS_STATES).
             AuthorityFrontierService.mark(
                 frontier_row,
-                "ingested",
+                C.DISCOVERY_STATE_INGESTED,
                 document_id=ingested_doc.id if ingested_doc else None,
                 candidate_record=candidate_record,
             )
@@ -394,25 +410,25 @@ def discover_and_bootstrap(
             )
             AuthorityFrontierService.mark(
                 frontier_row,
-                "failed",
+                C.DISCOVERY_STATE_FAILED,
                 error=str(exc),
                 candidate_record=cls._audit_record(
                     provider_name=name,
                     provider_license=provider.license,
                     source_domain=decision.source_domain,
                     verify=decision.verify,
-                    outcome="failed",
+                    outcome=C.DISCOVERY_STATE_FAILED,
                     error=str(exc),
                 ),
             )
             return {
-                "status": "failed",
+                "status": C.DISCOVERY_STATE_FAILED,
                 "error": str(exc),
                 "canonical_key": canonical_key,
             }
 
         return {
-            "status": "ingested",
+            "status": C.DISCOVERY_STATE_INGESTED,
             **result,
             "relinked_count": relinked_count,
             "canonical_key": canonical_key,