Address PR #2050 review: test constants, mark() guard, can_handle contract

Respond to the automated review on PR #2050:

- Replace the bare discovery-state literals in the three new regression tests
  with the C.DISCOVERY_STATE_* constants, so the new tests follow the same
  single-source-of-truth the PR establishes (no within-method "ingested"
  constant vs "failed"/"queued" literal mix).
- mark(): add a guard rejecting error= alongside a terminal SUCCESS state,
  mirroring the existing exclusivity guards. This makes the "an ingested row
  carries no error" invariant explicit rather than relying on no caller ever
  passing the contradiction.
- Document on BaseAuthoritySourceProvider.can_handle that overrides MUST be
  stateless — the orchestrator now probes one instance with multiple candidate
  keys and reuses it for locate/fetch, so the purity the optimisation relies on
  is codified for future provider authors.

Declined the list->tuple nit for DISCOVERY_STATE_CHOICES: the field's choices
match migration 0100's recorded list, and no code mutates it, so keeping the
list avoids any migration-state churn for a marginal immutability gain.

Author: claude

opencontractserver/enrichment/services/authority_frontier_service.py

@@ -277,6 +277,14 @@ def mark(
             raise ValueError("mark(): error and clear_error are exclusive.")
         if set_provider is not None and clear_provider:
             raise ValueError("mark(): set_provider and clear_provider are exclusive.")
+        if error is not None and state in C.DISCOVERY_SUCCESS_STATES:
+            # A terminal SUCCESS row must carry no error (the invariant the
+            # last_error auto-clear below establishes); refuse the contradiction
+            # loudly rather than persist an "ingested but errored" row.
+            raise ValueError(
+                "mark(): cannot set error on a terminal success transition "
+                f"({state!r}); omit error= (success states clear last_error)."
+            )
 
         row.discovery_state = state
         row.last_attempt = timezone.now()

opencontractserver/pipeline/base/base_authority_source_provider.py

@@ -55,6 +55,15 @@ class BaseAuthoritySourceProvider(PipelineComponentBase, ABC):
 
     # ---- public API (registry/orchestrator calls these) ---------------------
     def can_handle(self, canonical_key: str) -> bool:
+        """Return True if this provider serves *canonical_key*.
+
+        MUST be stateless. The discovery orchestrator instantiates each provider
+        once per ``_provider_for`` call and probes the SAME instance with multiple
+        candidate keys (then reuses it for ``locate``/``fetch``), so an override
+        that stashed match state on ``self`` here would leak stale state across
+        keys. Derive everything ``locate``/``fetch`` need from the key passed to
+        them instead.
+        """
         prefix = canonical_key.split(":", 1)[0]
         return prefix in self.supported_prefixes
 

opencontractserver/tests/test_authority_frontier.py

@@ -339,7 +339,7 @@ def test_dequeue_for_provider_excludes_unprovidered_rows(self):
         AuthorityFrontier.objects.create(
             canonical_key="usc-15:78j-noprov",
             authority="usc-15",
-            discovery_state="queued",
+            discovery_state=C.DISCOVERY_STATE_QUEUED,
             provider=None,
             mention_count=10,
         )
@@ -453,15 +453,19 @@ def test_mark_ingested_clears_stale_error_from_prior_failure(self):
         caller does NOT pass clear_error=True — ``mark()`` clears it implicitly on
         the SUCCESS transition (C.DISCOVERY_SUCCESS_STATES).
         """
-        row = self._make_row("usc-15:78j-retry", provider="TestProv", state="failed")
-        AuthorityFrontierService.mark(row, "failed", error="timeout fetching USLM")
+        row = self._make_row(
+            "usc-15:78j-retry", provider="TestProv", state=C.DISCOVERY_STATE_FAILED
+        )
+        AuthorityFrontierService.mark(
+            row, C.DISCOVERY_STATE_FAILED, error="timeout fetching USLM"
+        )
         row.refresh_from_db()
         self.assertEqual(row.last_error, "timeout fetching USLM")
 
         # Successful retry, no explicit clear_error.
         AuthorityFrontierService.mark(row, C.DISCOVERY_STATE_INGESTED)
         row.refresh_from_db()
-        self.assertEqual(row.discovery_state, "ingested")
+        self.assertEqual(row.discovery_state, C.DISCOVERY_STATE_INGESTED)
         self.assertIsNone(row.last_error)
 
     def test_mark_non_success_state_preserves_error(self):
@@ -471,9 +475,13 @@ def test_mark_non_success_state_preserves_error(self):
         error message keeps that message so operators can see why it failed.
         """
         row = self._make_row(
-            "usc-15:78j-stillbad", provider="TestProv", state="in_progress"
+            "usc-15:78j-stillbad",
+            provider="TestProv",
+            state=C.DISCOVERY_STATE_IN_PROGRESS,
+        )
+        AuthorityFrontierService.mark(
+            row, C.DISCOVERY_STATE_FAILED, error="still broken"
         )
-        AuthorityFrontierService.mark(row, "failed", error="still broken")
         row.refresh_from_db()
         self.assertEqual(row.last_error, "still broken")