feat: evaluate fresh + stable criteria

api/src/shared/common/seal_criteria.py

@@ -30,6 +30,54 @@ class SealCriterionName(str, Enum):
     FRESH_CONTINUOUS = "fresh_continuous"
 
 
+class CriterionStatus(str, Enum):
+    """A criterion's status. Values match the `seal_criterion_status` DB enum.
+
+    Only PASS and FAIL are verdicts. The other three say why there is no verdict, and they
+    are not interchangeable - the job's roll-up sends them in three different directions:
+
+    * UNKNOWN - we could not look; the inputs the check needs were not there. A property of
+      the run, not of the feed. The criterion keeps its last confirmed verdict and stays in
+      the roll-up, so an upstream outage freezes a criterion rather than waiving it.
+    * NOT_APPLICABLE - there is no question to ask; the criterion is deliberately excluded
+      for this feed (Fresh / future coverage on a seasonal feed). A property of the feed.
+      The criterion leaves the roll-up entirely.
+    * NEVER_EVALUATED - never had a verdict, since the feed first appeared. The initial
+      value, and the only one the job never writes back once a criterion has left it.
+
+    UNKNOWN is never written to `confirmed_status`: a run that could not look does not
+    change the answer, it leaves the previous one standing.
+    """
+
+    PASS = "pass"
+    FAIL = "fail"
+    UNKNOWN = "unknown"
+    NEVER_EVALUATED = "never_evaluated"
+    NOT_APPLICABLE = "not_applicable"
+
+    @property
+    def is_verdict(self) -> bool:
+        """True for PASS and FAIL - the two values that mean the check actually answered."""
+        return self in (CriterionStatus.PASS, CriterionStatus.FAIL)
+
+
+class CriterionPhase(str, Enum):
+    """Which of the two debouncing mechanisms is currently acting on a criterion.
+
+    Derived from the stored row rather than stored itself (see the job's
+    `state_machine.phase`): it is a pure function of `probation_start`, `confirmed_status`
+    and `first_observed_failure_at`, all of which are already on the row, so a stored copy
+    would be a second thing to keep in step for no gain.
+
+    The three values are mutually exclusive: probation suspends the grace period, so a
+    criterion can never be serving a penalty and holding a failure under grace at once.
+    """
+
+    STEADY = "steady"
+    IN_GRACE_PERIOD = "in_grace_period"
+    ON_PROBATION = "on_probation"
+
+
 # How long a criterion may keep failing its own check before the failure is confirmed and
 # the seal is withdrawn. None means the status flips on the first failing day.
 GRACE_PERIODS: Final[Dict[SealCriterionName, Optional[timedelta]]] = {
@@ -63,6 +111,33 @@ class SealCriterionName(str, Enum):
 )
 
 
+# Stable: how long we must have been tracking a feed - measured from its
+# `feed_reliability_seal.created_at` - before it can be called stable.
+TRACKING_PERIOD: Final[timedelta] = timedelta(days=180)
+
+# Fresh / future coverage: how far ahead the latest dataset's service coverage must reach
+FUTURE_COVERAGE_HORIZON: Final[timedelta] = timedelta(days=7)
+
+
+def grace_period_for(criterion: str | SealCriterionName) -> Optional[timedelta]:
+    """How long an observed failure of `criterion` may run before it is confirmed.
+
+    None means the criterion has no grace period and its status flips on the first failing
+    day. Callers ask for the window rather than declaring their own, so `GRACE_PERIODS`
+    stays the only place a value can change.
+    """
+    return GRACE_PERIODS[resolve_criterion(criterion)]
+
+
+def probation_period_for(criterion: str | SealCriterionName) -> Optional[timedelta]:
+    """How long `criterion` must go with no observed failure after a confirmed failure.
+
+    None means the criterion never serves probation (`official` and `stable`, which are
+    point-in-time state checks).
+    """
+    return PROBATION_PERIODS[resolve_criterion(criterion)]
+
+
 def resolve_criterion(criterion: str | SealCriterionName) -> SealCriterionName:
     """Coerce a stored criterion value to a `SealCriterionName`.
 

api/src/shared/db_models/reliability_criterion_impl.py

@@ -2,21 +2,18 @@
 
 from feeds_gen.models.reliability_criterion import ReliabilityCriterion
 from shared.common.seal_criteria import (
-    GRACE_PERIODS,
-    PROBATION_PERIODS,
+    CriterionStatus,
     SealCriterionName,
+    grace_period_for,
+    probation_period_for,
     resolve_criterion,
     window_end,
 )
 from shared.database_gen.sqlacodegen_models import SealCriterion as SealCriterionOrm
 
 # The API `status` values are the `seal_criterion_status` DB enum verbatim, so a stored status is
-# served as-is with no translation.
-STATUS_PASS = "pass"
-STATUS_FAIL = "fail"
-STATUS_UNKNOWN = "unknown"
-STATUS_NEVER_EVALUATED = "never_evaluated"
-STATUS_NOT_APPLICABLE = "not_applicable"
+# served as-is with no translation - `CriterionStatus` is a `str` enum over exactly those values,
+# shared with the nightly job so the two cannot drift apart.
 
 
 class ReliabilityCriterionImpl(ReliabilityCriterion):
@@ -42,7 +39,7 @@ def never_evaluated(cls, criterion: SealCriterionName) -> ReliabilityCriterion:
         """
         return cls(
             criterion=criterion.value,
-            status=STATUS_NEVER_EVALUATED,
+            status=CriterionStatus.NEVER_EVALUATED.value,
             in_grace_period=False,
             on_probation=False,
         )
@@ -65,7 +62,7 @@ def from_orm(cls, criterion_row: SealCriterionOrm | None) -> ReliabilityCriterio
         # `not_applicable` (withdrawn for this feed) and `never_evaluated` (no verdict ever) do not
         # participate in the seal, so they carry no grace period, no probation and no windows - just
         # the flat status, mirroring the row-less `never_evaluated` entry.
-        if status in (STATUS_NEVER_EVALUATED, STATUS_NOT_APPLICABLE):
+        if status in (CriterionStatus.NEVER_EVALUATED, CriterionStatus.NOT_APPLICABLE):
             return cls(
                 criterion=criterion.value,
                 status=status,
@@ -76,8 +73,8 @@ def from_orm(cls, criterion_row: SealCriterionOrm | None) -> ReliabilityCriterio
         # Criteria exempt from a window (`official` and `stable` from both, `fresh_continuous` from
         # grace) have their stored values ignored rather than trusted - the policy maps are the
         # authority on which criteria serve them.
-        grace_period = GRACE_PERIODS.get(criterion)
-        probation_period = PROBATION_PERIODS.get(criterion)
+        grace_period = grace_period_for(criterion)
+        probation_period = probation_period_for(criterion)
         probation_start = criterion_row.probation_start if probation_period else None
         on_probation = probation_start is not None
 
@@ -87,8 +84,8 @@ def from_orm(cls, criterion_row: SealCriterionOrm | None) -> ReliabilityCriterio
         # nothing left for grace to protect.
         in_grace_period = (
             grace_period is not None
-            and status == STATUS_FAIL
-            and criterion_row.confirmed_status == STATUS_PASS
+            and status == CriterionStatus.FAIL
+            and criterion_row.confirmed_status == CriterionStatus.PASS
             and not on_probation
         )
 

api/tests/unittest/models/test_feed_reliability_report_impl.py

@@ -3,10 +3,9 @@
 from types import SimpleNamespace
 
 from shared.common.error_handling import InternalHTTPException
-from shared.common.seal_criteria import PROBATION_PERIOD, SealCriterionName
+from shared.common.seal_criteria import PROBATION_PERIOD, CriterionStatus, SealCriterionName
 from shared.database_gen.sqlacodegen_models import FeedReliabilitySeal, SealCriterion
 from shared.db_models.feed_reliability_report_impl import FeedReliabilityReportImpl
-from shared.db_models.reliability_criterion_impl import STATUS_FAIL, STATUS_NEVER_EVALUATED, STATUS_PASS
 
 # Anchored to the real clock because the countdowns are derived against `datetime.now`. Every window
 # below is at least a day clear of its boundary, so the assertions do not race the wall clock.
@@ -68,7 +67,7 @@ def test_never_evaluated_feed(self):
         assert report.evaluated_at is None
         assert report.on_probation is False
         assert len(report.criteria) == 6
-        assert all(criterion.status == STATUS_NEVER_EVALUATED for criterion in report.criteria)
+        assert all(criterion.status == CriterionStatus.NEVER_EVALUATED.value for criterion in report.criteria)
 
     def test_all_six_criteria_always_returned_in_order(self):
         """Criteria with no row are filled in, so a client can render six cards unconditionally."""
@@ -77,8 +76,8 @@ def test_all_six_criteria_always_returned_in_order(self):
 
         assert [criterion.criterion for criterion in report.criteria] == [name.value for name in SealCriterionName]
         criteria = by_criterion(report)
-        assert criteria["official"].status == STATUS_PASS
-        assert criteria["compliant"].status == STATUS_NEVER_EVALUATED
+        assert criteria["official"].status == CriterionStatus.PASS.value
+        assert criteria["compliant"].status == CriterionStatus.NEVER_EVALUATED.value
 
     def test_mixed_criteria(self):
         """The report carries each criterion's own verdict alongside the stored seal outcome."""
@@ -106,10 +105,10 @@ def test_mixed_criteria(self):
 
         criteria = by_criterion(report)
         assert report.has_seal is False
-        assert criteria["official"].status == STATUS_PASS
-        assert criteria["available"].status == STATUS_FAIL
+        assert criteria["official"].status == CriterionStatus.PASS.value
+        assert criteria["available"].status == CriterionStatus.FAIL.value
         assert criteria["available"].in_grace_period is False
-        assert criteria["compliant"].status == STATUS_FAIL
+        assert criteria["compliant"].status == CriterionStatus.FAIL.value
         assert criteria["compliant"].in_grace_period is True
 
     def test_evaluated_at_is_latest_across_criteria(self):

api/tests/unittest/models/test_reliability_criterion_impl.py

@@ -2,16 +2,14 @@
 from datetime import datetime, timedelta, timezone
 
 from shared.common.error_handling import InternalHTTPException
-from shared.common.seal_criteria import GRACE_PERIODS, PROBATION_PERIOD, SealCriterionName
-from shared.database_gen.sqlacodegen_models import SealCriterion
-from shared.db_models.reliability_criterion_impl import (
-    STATUS_FAIL,
-    STATUS_NEVER_EVALUATED,
-    STATUS_NOT_APPLICABLE,
-    STATUS_PASS,
-    STATUS_UNKNOWN,
-    ReliabilityCriterionImpl,
+from shared.common.seal_criteria import (
+    GRACE_PERIODS,
+    PROBATION_PERIOD,
+    CriterionStatus,
+    SealCriterionName,
 )
+from shared.database_gen.sqlacodegen_models import SealCriterion
+from shared.db_models.reliability_criterion_impl import ReliabilityCriterionImpl
 
 # Anchored to the real clock because the countdowns are derived against `datetime.now`. Every window
 # below is at least a day clear of its boundary, so the assertions do not race the wall clock.
@@ -43,7 +41,7 @@ def test_passing_criterion(self):
         result = ReliabilityCriterionImpl.from_orm(make_row())
 
         assert result.criterion == "compliant"
-        assert result.status == STATUS_PASS
+        assert result.status == CriterionStatus.PASS.value
         assert result.in_grace_period is False
         assert result.grace_period_ends_at is None
         assert result.on_probation is False
@@ -73,7 +71,7 @@ def test_never_evaluated_factory(self):
         result = ReliabilityCriterionImpl.never_evaluated(SealCriterionName.AVAILABLE)
 
         assert result.criterion == "available"
-        assert result.status == STATUS_NEVER_EVALUATED
+        assert result.status == CriterionStatus.NEVER_EVALUATED.value
         assert result.in_grace_period is False
         assert result.on_probation is False
 
@@ -82,7 +80,7 @@ def test_never_evaluated_status_passes_through(self):
         row = make_row(observed_status="never_evaluated", confirmed_status="never_evaluated")
         result = ReliabilityCriterionImpl.from_orm(row)
 
-        assert result.status == STATUS_NEVER_EVALUATED
+        assert result.status == CriterionStatus.NEVER_EVALUATED.value
         assert result.in_grace_period is False
         assert result.on_probation is False
 
@@ -91,7 +89,7 @@ def test_unknown_status_passes_through(self):
         row = make_row(criterion=SealCriterionName.AVAILABLE, observed_status="unknown", confirmed_status="pass")
         result = ReliabilityCriterionImpl.from_orm(row)
 
-        assert result.status == STATUS_UNKNOWN
+        assert result.status == CriterionStatus.UNKNOWN.value
         assert result.in_grace_period is False
 
     def test_not_applicable_status_is_withdrawn(self):
@@ -107,7 +105,7 @@ def test_not_applicable_status_is_withdrawn(self):
         )
         result = ReliabilityCriterionImpl.from_orm(row)
 
-        assert result.status == STATUS_NOT_APPLICABLE
+        assert result.status == CriterionStatus.NOT_APPLICABLE.value
         assert result.in_grace_period is False
         assert result.on_probation is False
         assert result.probation_ends_at is None
@@ -126,7 +124,7 @@ def test_failing_inside_grace_period(self):
         )
         result = ReliabilityCriterionImpl.from_orm(row)
 
-        assert result.status == STATUS_FAIL
+        assert result.status == CriterionStatus.FAIL.value
         assert result.in_grace_period is True
         assert result.grace_period_ends_at == first_failure + GRACE_PERIODS[SealCriterionName.COMPLIANT]
         assert result.first_failure_at == first_failure
@@ -147,7 +145,7 @@ def test_grace_exempt_criterion_is_never_in_grace(self):
         )
         result = ReliabilityCriterionImpl.from_orm(row)
 
-        assert result.status == STATUS_FAIL
+        assert result.status == CriterionStatus.FAIL.value
         assert result.in_grace_period is False
         assert result.grace_period_ends_at is None
 
@@ -162,7 +160,7 @@ def test_failing_beyond_grace_period(self):
         )
         result = ReliabilityCriterionImpl.from_orm(row)
 
-        assert result.status == STATUS_FAIL
+        assert result.status == CriterionStatus.FAIL.value
         assert result.in_grace_period is False
         assert result.grace_period_ends_at is None
 
@@ -175,7 +173,7 @@ def test_passing_while_on_probation(self):
         row = make_row(criterion=SealCriterionName.AVAILABLE, probation_start=probation_start)
         result = ReliabilityCriterionImpl.from_orm(row)
 
-        assert result.status == STATUS_PASS
+        assert result.status == CriterionStatus.PASS.value
         assert result.on_probation is True
         assert result.probation_ends_at == probation_start + PROBATION_PERIOD
 
@@ -191,7 +189,7 @@ def test_grace_does_not_apply_during_probation(self):
         )
         result = ReliabilityCriterionImpl.from_orm(row)
 
-        assert result.status == STATUS_FAIL
+        assert result.status == CriterionStatus.FAIL.value
         assert result.on_probation is True
         assert result.in_grace_period is False
         assert result.grace_period_ends_at is None
@@ -218,7 +216,7 @@ def test_official_and_stable_have_no_grace_period(self):
                 )
                 result = ReliabilityCriterionImpl.from_orm(row)
 
-                assert result.status == STATUS_FAIL
+                assert result.status == CriterionStatus.FAIL.value
                 assert result.in_grace_period is False
                 assert result.grace_period_ends_at is None
 

api/tests/unittest/models/test_seal_enum_contract.py

@@ -11,8 +11,9 @@
 mirror of these tests over its own enums, in
 `functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_enum_contract.py`.
 
-Adding a criterion or status means updating, in lockstep: the Liquibase enum, `SealCriterionName`
-and the status constants here, `docs/DatabaseCatalogAPI.yaml` (plus a stub regen), and the job enums.
+Adding a criterion or status means updating, in lockstep: the Liquibase enum, the enums in
+`shared.common.seal_criteria` (which the API and the nightly job both read), and
+`docs/DatabaseCatalogAPI.yaml` plus a stub regen.
 """
 
 import unittest
@@ -21,24 +22,12 @@
 from shared.common.seal_criteria import (
     GRACE_PERIODS,
     PROBATION_PERIODS,
+    CriterionStatus,
     SealCriterionName,
 )
 from shared.database_gen.sqlacodegen_models import SealCriterion
-from shared.db_models.reliability_criterion_impl import (
-    STATUS_FAIL,
-    STATUS_NEVER_EVALUATED,
-    STATUS_NOT_APPLICABLE,
-    STATUS_PASS,
-    STATUS_UNKNOWN,
-)
 
-API_STATUSES = {
-    STATUS_PASS,
-    STATUS_FAIL,
-    STATUS_UNKNOWN,
-    STATUS_NEVER_EVALUATED,
-    STATUS_NOT_APPLICABLE,
-}
+API_STATUSES = {status.value for status in CriterionStatus}
 
 
 def db_enum_values(column_name: str) -> set:
@@ -53,13 +42,19 @@ def test_criterion_names_match_db_enum(self):
         """`SealCriterionName` is the full `seal_criterion_name` type, no more and no less."""
         assert {criterion.value for criterion in SealCriterionName} == db_enum_values("criterion")
 
-    def test_status_constants_match_db_enum(self):
-        """The API `status` values are the `seal_criterion_status` type verbatim.
+    def test_status_values_match_db_enum(self):
+        """`CriterionStatus` is the `seal_criterion_status` type verbatim.
 
         There is no DB-to-API translation left, so any divergence would be served raw to clients.
+        The nightly job writes these same values from the same enum, so one assertion pins both
+        sides to the schema.
         """
         assert API_STATUSES == db_enum_values("observed_status")
 
+    def test_confirmed_status_shares_the_same_type(self):
+        """Both status columns are the one enum; a check that only covered one could drift."""
+        assert db_enum_values("confirmed_status") == db_enum_values("observed_status")
+
     def test_every_db_status_passes_response_validation(self):
         """Every status the job can store must be accepted by the generated response model.
 
@@ -84,8 +79,9 @@ def test_every_criterion_passes_response_validation(self):
     def test_policy_maps_cover_every_criterion(self):
         """Every criterion needs a grace and probation entry, even if the window is None.
 
-        `GRACE_PERIODS.get()` would silently return None for a criterion missing from the map,
-        turning it into a no-grace criterion by accident rather than by decision.
+        The maps are the single source for both the API's countdowns and the job's debouncing
+        (`grace_period_for` / `probation_period_for`), so a criterion missing from one would raise
+        on lookup rather than quietly becoming a no-grace criterion.
         """
         assert set(GRACE_PERIODS) == set(SealCriterionName)
         assert set(PROBATION_PERIODS) == set(SealCriterionName)