diff --git a/api/src/shared/database/database.py b/api/src/shared/database/database.py
index 6652d108c..8ddb4e95e 100644
--- a/api/src/shared/database/database.py
+++ b/api/src/shared/database/database.py
@@ -77,6 +77,7 @@ def configure_polymorphic_mappers():
Feed.redirectingids, # redirectingid_source_id_fkey
Feed.redirectingids_, # redirectingid_target_id_fkey
Feed.feed_license_changes,
+ Feed.sealcriteria, # sealcriterion_feed_id_fkey
],
Gtfsfeed: [Gtfsfeed.gtfs_dataset_changelogs],
Gbfsfeed: [
diff --git a/api/tests/integration/cascade_delete/test_cascade_delete.py b/api/tests/integration/cascade_delete/test_cascade_delete.py
index dee5cd89a..8b8532a3d 100644
--- a/api/tests/integration/cascade_delete/test_cascade_delete.py
+++ b/api/tests/integration/cascade_delete/test_cascade_delete.py
@@ -26,6 +26,7 @@
Gtfsfeed,
Geopolygon,
FeedLicenseChange,
+ Sealcriterion,
)
from sqlalchemy import text
@@ -207,6 +208,56 @@ def test_delete_feed_cascadeto_feed_license_changes(test_database):
)
+def test_delete_feed_cascadeto_sealcriterion(test_database):
+ """Deleting a feed removes its per-criterion seal state (issue #1760).
+
+ This exercises the `Feed.sealcriteria` entry in `cascade_entities`: without it,
+ SQLAlchemy would try to NULL `sealcriterion.feed_id`, which is NOT NULL and part of the
+ composite primary key, instead of letting ON DELETE CASCADE do the work.
+ """
+
+ with test_database.start_db_session() as session:
+ feed = Feed(id="f1")
+ seal_criterion = Sealcriterion(feed_id="f1", criterion="official")
+ session.add_all([feed, seal_criterion])
+ session.commit()
+
+ delete_and_assert(
+ session,
+ [
+ "SELECT COUNT(*) FROM feed where id = 'f1'",
+ "SELECT COUNT(*) FROM sealcriterion where feed_id = 'f1'",
+ ],
+ feed,
+ )
+
+
+def test_delete_feed_cascadeto_feedreliabilityseal(test_database):
+ """Deleting a feed removes its overall seal row (issue #1760).
+
+ The row is inserted with raw SQL rather than the ORM: `feedreliabilityseal.feed_id` is
+ both its primary key and a foreign key to `feed.id`, so sqlacodegen maps the table as a
+ joined-table subclass of Feed. Adding a `Feedreliabilityseal()` instance would therefore
+ try to insert a second feed row rather than a child row.
+ """
+
+ with test_database.start_db_session() as session:
+ feed = Feed(id="f1")
+ session.add(feed)
+ session.commit()
+ session.execute(text("INSERT INTO feedreliabilityseal (feed_id) VALUES ('f1')"))
+ session.commit()
+
+ delete_and_assert(
+ session,
+ [
+ "SELECT COUNT(*) FROM feed where id = 'f1'",
+ "SELECT COUNT(*) FROM feedreliabilityseal where feed_id = 'f1'",
+ ],
+ feed,
+ )
+
+
def test_delete_gbfsfeed_cascadeto_gbfsversion_cascadeto_gbfsendpoint_cascadeto_gbfsendpointhttpaccesslog(
test_database,
):
diff --git a/functions-python/tasks_executor/README.md b/functions-python/tasks_executor/README.md
index 420b02178..1ad81f3fa 100644
--- a/functions-python/tasks_executor/README.md
+++ b/functions-python/tasks_executor/README.md
@@ -406,3 +406,74 @@ Only **comparable** datasets are considered: a dataset must have a `downloaded_a
| `pairs_dispatched` | Pairs dispatched (or, in `dry_run`, that would be dispatched) |
| `dispatched` | (dry-run only) list of `{feed_stable_id, base_dataset_stable_id, new_dataset_stable_id}` |
+
+### update_seal_of_reliability
+
+Evaluates the implemented Seal of Reliability criteria (issue #1761) for every eligible GTFS
+feed and updates the `sealcriterion` and `feedreliabilityseal` tables. Reads the source
+tables and never modifies them.
+
+`seal_criterion_name` in the database declares all six criteria, so adding an evaluator
+needs no schema change. The criteria still to be implemented are tracked by #1784 and #1782.
+
+Eligible feeds are GTFS, `operational_status = published`, and `status NOT IN (deprecated,
+development)`. `inactive` and `future` feeds are deliberately included: skipping a feed
+freezes its stored rows rather than making it neutral.
+
+```json
+{
+ "task": "update_seal_of_reliability",
+ "payload": {
+ "dry_run": true,
+ "stable_feed_ids": ["mdb-1210"]
+ }
+}
+```
+
+| Parameter | Type | Default | Description |
+|---|---|---|---|
+| `dry_run` | bool | `true` | Evaluate every feed and return the report without writing anything |
+| `stable_feed_ids` | list[str] \| null | `null` | Evaluate only these feeds. Ids that are unknown or not eligible are skipped with a logged warning naming them; it raises only when *none* of them can be evaluated. When set, `feeds` covers every one of those feeds instead of only the ones that moved |
+| `limit` | int \| null | `null` | Cap the number of feeds evaluated |
+| `criteria` | list[str] \| null | `null` | Evaluate only these criteria. Naming a criterion that has no evaluator yet raises. A subset of the implemented criteria skips the `has_seal` roll-up, since the ones not evaluated cannot be judged |
+| `batch_size` | int | `200` | Feeds loaded per query batch. Every eligible feed is still evaluated — this only sizes the queries |
+| `max_reported_feeds` | int | `50` | Cap on the `feeds` list in the response. Everything is still evaluated and written; `feeds_omitted` says how many entries were left out |
+| `now` | str \| null | `null` | ISO timestamp to evaluate against, for replays and backfills. Defaults to the current UTC time |
+
+**Response fields**:
+
+| Field | Description |
+|---|---|
+| `total_feeds` | Feeds evaluated |
+| `criteria` | The criteria evaluated in this run |
+| `partial_run` | True when `criteria` was a subset, meaning `has_seal` was not recalculated |
+| `criterion_rows_written` | `sealcriterion` rows inserted or updated (`0` on a dry run) |
+| `not_evaluable` | Criterion evaluations skipped because an input was missing |
+| `seals_before_run` | Feeds that held the seal before this run |
+| `seals_after_run` | Feeds holding it afterwards — on a dry run, what *would* be stored |
+| `seals_granted` / `seals_revoked` | Transitions in this run. `before + granted - revoked == after` |
+| `granted_stable_ids` / `revoked_stable_ids` | The feeds behind `seals_granted` / `seals_revoked` — the two transitions written to `feedreliabilityseal` in this run |
+| `first_evaluations` | Criteria evaluated for the first time (no stored row yet) |
+| `feeds` | One entry per reported feed, capped at `max_reported_feeds`: `stable_id`, its `feedreliabilityseal` state (`had_seal`, `has_seal`), and a nested `criteria` list holding every criterion of that feed with `observed_pass`, `confirmed_pass`, `previously_confirmed_pass`, `on_probation` and `reason`. A feed is reported when it was named in `stable_feed_ids`, when one of its criteria moved, or when its seal changed — so a quiet nightly run returns an empty list |
+| `feeds_omitted` | Feeds left out of `feeds` by the cap. `sealcriterion` and `feedreliabilityseal` hold everything regardless |
+
+#### Running it locally
+
+Start Postgres and the function, then post to it:
+
+```shell
+docker compose --env-file ./config/.env.local up -d --force-recreate
+scripts/function-python-run.sh --function_name tasks_executor --no_install_venv
+```
+
+```shell
+curl -s -X POST http://localhost:8080 -H "Content-Type: application/json" \
+ -d '{"task":"update_seal_of_reliability","payload":{"dry_run":true,"stable_feed_ids":["mdb-1210"]}}' \
+ | python3 -m json.tool
+```
+
+`Accept: text/csv` returns a single summary row (the top-level report fields), not one row
+per feed — the converter flattens the returned dict, and `feeds` lands in it as a single
+stringified cell. Use the JSON response for per-feed and per-criterion detail.
+
+Nothing about this task needs GCP credentials — only `FEEDS_DATABASE_URL`.
diff --git a/functions-python/tasks_executor/src/main.py b/functions-python/tasks_executor/src/main.py
index 943590341..00a91913c 100644
--- a/functions-python/tasks_executor/src/main.py
+++ b/functions-python/tasks_executor/src/main.py
@@ -76,6 +76,9 @@
notifications_dispatch_monitor_handler,
)
from tasks.changelog.backfill_changelog import backfill_changelog_handler
+from tasks.seal_of_reliability.update_seal_of_reliability import (
+ update_seal_of_reliability_handler,
+)
init_logger()
LIST_COMMAND: Final[str] = "list"
@@ -255,6 +258,20 @@
),
"handler": backfill_changelog_handler,
},
+ "update_seal_of_reliability": {
+ "description": (
+ "Evaluates the implemented Seal of Reliability criteria for every eligible "
+ "GTFS feed and updates sealcriterion and feedreliabilityseal. "
+ "Reads the source tables and never modifies them. "
+ "Parameters: dry_run (default true), stable_feed_ids (default null; when set, "
+ "`feeds` covers every one of those feeds), limit (default null), "
+ "criteria (default null "
+ "meaning every implemented criterion; a partial set skips the has_seal "
+ "roll-up), batch_size (default 200), now (ISO timestamp, default current "
+ "UTC time)."
+ ),
+ "handler": update_seal_of_reliability_handler,
+ },
}
diff --git a/functions-python/tasks_executor/src/tasks/seal_of_reliability/__init__.py b/functions-python/tasks_executor/src/tasks/seal_of_reliability/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/functions-python/tasks_executor/src/tasks/seal_of_reliability/context.py b/functions-python/tasks_executor/src/tasks/seal_of_reliability/context.py
new file mode 100644
index 000000000..9b909189d
--- /dev/null
+++ b/functions-python/tasks_executor/src/tasks/seal_of_reliability/context.py
@@ -0,0 +1,122 @@
+#
+# MobilityData 2026
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""Feed eligibility query and bulk loading of everything the evaluators need.
+
+The evaluators are pure functions over a `FeedSealContext`, so every DB read for a batch of
+feeds happens here, in a fixed number of queries regardless of batch size.
+
+Only Official is implemented, so the context currently carries just the feed row. Each new
+criterion adds the fields it needs here plus one bulk query to populate them: the latest
+dataset for Compliant and Fresh, the day's availability rows for Available, the full
+dataset coverage history for Fresh continuous coverage.
+"""
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Dict, Optional, Sequence
+
+from sqlalchemy.orm import Session
+
+from shared.database_gen.sqlacodegen_models import Feed, Gtfsfeed
+
+
+@dataclass
+class FeedSealContext:
+ """Everything the evaluators need for one feed.
+
+ Built by `build_contexts`. Evaluators read from this and never query.
+ """
+
+ feed_id: str
+ # The evaluation timestamp. Passed in rather than read from the clock so evaluators
+ # stay pure and a run can be replayed for any point in time.
+ now: datetime
+ stable_id: Optional[str] = None
+
+ # Feed-level flags
+ official: Optional[bool] = None
+
+
+def get_seal_feeds_query(
+ db_session: Session, stable_feed_ids: Optional[Sequence[str]] = None
+):
+ """Return a query for the feeds the seal applies to.
+
+ Eligibility is defined here and nowhere else, so a full run and a one-feed run exercise
+ the same predicate. `inactive` and `future` feeds are deliberately included: skipping a
+ feed does not make it neutral, it freezes its stored rows, and once Fresh (future
+ coverage) exists an inactive feed should fail it rather than keep displaying a seal.
+ """
+ query = db_session.query(Gtfsfeed).filter(
+ Feed.data_type == "gtfs",
+ Feed.status.notin_(["deprecated", "development"]),
+ Feed.operational_status == "published",
+ )
+ if stable_feed_ids is not None:
+ query = query.filter(Feed.stable_id.in_(list(stable_feed_ids)))
+ return query
+
+
+def build_contexts(
+ db_session: Session, feeds: Sequence[Gtfsfeed], now: datetime
+) -> Dict[str, FeedSealContext]:
+ """Load everything the evaluators need for `feeds`, in a fixed number of queries.
+
+ Args:
+ db_session: SQLAlchemy session. Unused while Official is the only criterion, since
+ everything it needs is already on the feed row, but kept in the signature
+ because every further criterion needs it.
+ feeds: The batch of feeds to load, from `get_seal_feeds_query`.
+ now: The evaluation timestamp.
+
+ Returns:
+ feed_id -> FeedSealContext.
+
+ How to add a criterion's data. Two kinds:
+
+ 1. Already on the selected feed row (`official`, `created_at`, `seasonal`,
+ `is_producer_url_unstable`). Add the field to FeedSealContext and read it off `feed`
+ below. No query, no cost.
+
+ 2. Needs its own query. Add the field, then a module-level `_load_*` helper that takes
+ the whole batch and returns a dict keyed by feed_id, and call it once here. Keeping
+ the query per batch rather than per feed is what holds the query count proportional
+ to the number of criteria instead of the number of feeds. For example, Available
+ (issue #1784) would add:
+
+ def _load_availability_today(db_session, feed_ids, day_start) -> Dict[str, bool]:
+ '''feed_id -> whether any availability check succeeded since day_start.
+ Feeds absent from the result had no check at all, which the criterion reads
+ as "not evaluable" rather than "failing".'''
+
+ called once as `availability = _load_availability_today(...)` and consumed per feed
+ as `availability_success_today=availability.get(feed.id, False)`.
+ """
+ return {
+ feed.id: FeedSealContext(
+ feed_id=feed.id,
+ now=now,
+ stable_id=feed.stable_id,
+ official=feed.official,
+ )
+ for feed in feeds
+ }
+
+
+def batched(items: Sequence, size: int):
+ """Yield successive slices of `items` of at most `size` elements."""
+ for start in range(0, len(items), size):
+ yield items[start : start + size]
diff --git a/functions-python/tasks_executor/src/tasks/seal_of_reliability/criteria.py b/functions-python/tasks_executor/src/tasks/seal_of_reliability/criteria.py
new file mode 100644
index 000000000..1ba4393de
--- /dev/null
+++ b/functions-python/tasks_executor/src/tasks/seal_of_reliability/criteria.py
@@ -0,0 +1,54 @@
+#
+# MobilityData 2026
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""Seal of Reliability policy values.
+
+Everything here is the published definition of the seal, so it lives in code rather than
+in DB config: changing any of these values changes which feeds qualify, and that should
+go through code review, tests and a deploy.
+
+Each criterion's grace period belongs with the criterion, as a class attribute on its
+evaluator. Only Official is implemented so far and it has none.
+"""
+
+from datetime import timedelta
+from enum import Enum
+from typing import Final
+
+
+class SealCriterionName(str, Enum):
+ """The six seal criteria. Values match the seal_criterion_name DB enum.
+
+ All six are listed even though only Official has an evaluator (see #1784 and #1782),
+ so that the enum stays a faithful mirror of the database type.
+ """
+
+ OFFICIAL = "official"
+ STABLE = "stable"
+ AVAILABLE = "available"
+ COMPLIANT = "compliant"
+ FRESH_COVERAGE = "fresh_coverage"
+ FRESH_CONTINUOUS = "fresh_continuous"
+
+
+# A criterion that recovers from a confirmed failure is put on probation: it must then go
+# this long with no observed failure before it can contribute to the seal again. It is the
+# default for new evaluators; Official is exempt because it is a point-in-time state check
+# (see OfficialEvaluator).
+#
+# Probation is opened only by a recovery. A first evaluation that passes is not a recovery,
+# so a feed that has never had a confirmed failure never serves probation at all and can
+# hold the seal from its very first evaluation.
+PROBATION_PERIOD: Final[timedelta] = timedelta(days=180)
diff --git a/functions-python/tasks_executor/src/tasks/seal_of_reliability/evaluators/__init__.py b/functions-python/tasks_executor/src/tasks/seal_of_reliability/evaluators/__init__.py
new file mode 100644
index 000000000..3ddabbe94
--- /dev/null
+++ b/functions-python/tasks_executor/src/tasks/seal_of_reliability/evaluators/__init__.py
@@ -0,0 +1,43 @@
+#
+# MobilityData 2026
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""The seal criterion evaluators.
+
+`EVALUATORS` is the registry the job iterates. Only Official is implemented so far
+(issue #1783); the remaining criteria are tracked by #1784 and #1782. Adding one means a
+new subclass, an entry here, and whatever fields it needs on `FeedSealContext`.
+
+`seal_criterion_name` in the database already declares all six values, so a criterion can
+be added without a schema change.
+"""
+
+from typing import Final, List
+
+from tasks.seal_of_reliability.evaluators.base import (
+ CriterionEvaluator,
+ CriterionObservation,
+)
+from tasks.seal_of_reliability.evaluators.official import OfficialEvaluator
+
+EVALUATORS: Final[List[CriterionEvaluator]] = [
+ OfficialEvaluator(),
+]
+
+__all__ = [
+ "EVALUATORS",
+ "CriterionEvaluator",
+ "CriterionObservation",
+ "OfficialEvaluator",
+]
diff --git a/functions-python/tasks_executor/src/tasks/seal_of_reliability/evaluators/base.py b/functions-python/tasks_executor/src/tasks/seal_of_reliability/evaluators/base.py
new file mode 100644
index 000000000..f11aef120
--- /dev/null
+++ b/functions-python/tasks_executor/src/tasks/seal_of_reliability/evaluators/base.py
@@ -0,0 +1,72 @@
+#
+# MobilityData 2026
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""Base class for the per-criterion evaluators."""
+
+from dataclasses import dataclass
+from datetime import timedelta
+from typing import Optional, Tuple
+
+from tasks.seal_of_reliability.context import FeedSealContext
+from tasks.seal_of_reliability.criteria import PROBATION_PERIOD, SealCriterionName
+
+
+@dataclass(frozen=True)
+class CriterionObservation:
+ """One criterion's own check for one feed, with no debouncing.
+
+ `observed_pass` is tri-state: True and False are verdicts, None means the criterion
+ could not be evaluated this run (a missing availability check, a criterion that does not
+ apply to this feed) and the stored state must be left alone.
+
+ Stated positively, so an evaluator returns the same sense as the check it reads
+ (`success = TRUE`, `total_error = 0`) with no inversion in between.
+ """
+
+ criterion: SealCriterionName
+ observed_pass: Optional[bool]
+ reason: str
+
+
+class CriterionEvaluator:
+ """Evaluates one criterion against a pre-loaded feed context.
+
+ Subclasses set `name` and, where they differ from the defaults, `grace_period` and
+ `probation_period`, then implement `_evaluate`. They never touch the database: all the
+ data they need is on the context, loaded in bulk by `context.build_contexts`.
+
+ `grace_period` holds a passing status while an observed failure is still young.
+ `probation_period` is how long the criterion must go with no observed failure after
+ recovering from a confirmed failure. None on either means the criterion does not use it.
+ """
+
+ name: SealCriterionName = None
+ grace_period: Optional[timedelta] = None
+ probation_period: Optional[timedelta] = PROBATION_PERIOD
+
+ def evaluate(self, ctx: FeedSealContext) -> CriterionObservation:
+ """Evaluate the criterion and label the result with this evaluator's name.
+
+ Labelling happens here rather than in the subclasses so an evaluator cannot
+ disagree with its own `name`.
+ """
+ observed_pass, reason = self._evaluate(ctx)
+ return CriterionObservation(
+ criterion=self.name, observed_pass=observed_pass, reason=reason
+ )
+
+ def _evaluate(self, ctx: FeedSealContext) -> Tuple[Optional[bool], str]:
+ """Return (observed_pass, reason) for this feed. Implemented by subclasses."""
+ raise NotImplementedError("Subclasses should implement this method.")
diff --git a/functions-python/tasks_executor/src/tasks/seal_of_reliability/evaluators/official.py b/functions-python/tasks_executor/src/tasks/seal_of_reliability/evaluators/official.py
new file mode 100644
index 000000000..2401c6f36
--- /dev/null
+++ b/functions-python/tasks_executor/src/tasks/seal_of_reliability/evaluators/official.py
@@ -0,0 +1,41 @@
+#
+# MobilityData 2026
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""Official criterion: the feed is flagged official."""
+
+from typing import Optional, Tuple
+
+from tasks.seal_of_reliability.context import FeedSealContext
+from tasks.seal_of_reliability.criteria import SealCriterionName
+from tasks.seal_of_reliability.evaluators.base import CriterionEvaluator
+
+
+class OfficialEvaluator(CriterionEvaluator):
+ """`feed.official IS TRUE`.
+
+ A point-in-time state check: official at the time of reviewing the dataset, with no
+ 6-month check. Both the grace period and the probation period are None, so the criterion
+ fails the same day the flag is lost and clears the same day it comes back, taking the
+ seal with it in both directions.
+ """
+
+ name = SealCriterionName.OFFICIAL
+ grace_period = None
+ probation_period = None
+
+ def _evaluate(self, ctx: FeedSealContext) -> Tuple[Optional[bool], str]:
+ # `is True` rather than a truthiness test: NULL is not an endorsement, and it is a
+ # verdict of "not official" rather than "not evaluable".
+ return ctx.official is True, f"feed.official is {ctx.official!r}"
diff --git a/functions-python/tasks_executor/src/tasks/seal_of_reliability/seal_updater.py b/functions-python/tasks_executor/src/tasks/seal_of_reliability/seal_updater.py
new file mode 100644
index 000000000..10661e47e
--- /dev/null
+++ b/functions-python/tasks_executor/src/tasks/seal_of_reliability/seal_updater.py
@@ -0,0 +1,501 @@
+#
+# MobilityData 2026
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""Nightly Seal of Reliability evaluation (issue #1761).
+
+Reads feed, dataset, validation report and availability data; writes only sealcriterion
+and feedreliabilityseal. The raw source tables are never modified.
+
+Both seal tables are written with Core statements against `__table__` rather than through
+ORM objects. `feedreliabilityseal.feed_id` is both its primary key and a foreign key to
+feed(id), which sqlacodegen maps as joined-table inheritance
+(`class Feedreliabilityseal(Feed)`, a sibling of Gtfsfeed in the polymorphic hierarchy), so
+persisting an ORM instance would try to insert a new feed. Core statements address the
+table directly and are unaffected.
+"""
+
+import logging
+import time
+from datetime import datetime, timezone
+from typing import Dict, List, Optional, Sequence, Set, Tuple
+
+from sqlalchemy import select
+from sqlalchemy.dialects.postgresql import insert
+from sqlalchemy.orm import Session
+
+from shared.database.database import with_db_session
+from shared.database_gen.sqlacodegen_models import (
+ Feed,
+ Feedreliabilityseal,
+ Sealcriterion,
+)
+
+from tasks.seal_of_reliability.context import (
+ batched,
+ build_contexts,
+ get_seal_feeds_query,
+)
+from tasks.seal_of_reliability.criteria import SealCriterionName
+from tasks.seal_of_reliability.evaluators import EVALUATORS
+from tasks.seal_of_reliability.state_machine import SealCriterionState, transition
+
+DEFAULT_BATCH_SIZE: int = 200
+
+# Limit the number of feeds reported so the return does not get gigantic.
+DEFAULT_MAX_REPORTED_FEEDS: int = 50
+
+SEAL_TABLE = Feedreliabilityseal.__table__
+CRITERION_TABLE = Sealcriterion.__table__
+
+
+def _resolve_evaluators(criteria: Optional[Sequence[str]]) -> List:
+ """Return the evaluators to run, optionally filtered to `criteria`."""
+ if criteria is None:
+ return list(EVALUATORS)
+ wanted = {str(name) for name in criteria}
+ known = {evaluator.name.value for evaluator in EVALUATORS}
+ unknown = sorted(wanted - known)
+ if unknown:
+ raise ValueError(
+ f"Unknown criteria: {unknown}. Known criteria: {sorted(known)}"
+ )
+ return [evaluator for evaluator in EVALUATORS if evaluator.name.value in wanted]
+
+
+def _validate_requested_feed_ids(
+ db_session: Session,
+ requested: Sequence[str],
+ evaluated: Set[str],
+) -> None:
+ """Warn about requested feeds that will not be evaluated, or raise if that is all of them."""
+ unusable = sorted(set(requested) - evaluated)
+ if not unusable:
+ return # every requested feed is being evaluated, so there is nothing to report
+
+ # One extra query, and only now that something has already been dropped.
+ existing = {
+ row.stable_id
+ for row in db_session.execute(
+ select(Feed.stable_id).where(Feed.stable_id.in_(unusable))
+ ).all()
+ }
+ unknown = sorted(set(unusable) - existing)
+
+ problems = []
+ if unknown:
+ problems.append(f"not found: {unknown}")
+ if existing:
+ problems.append(
+ f"not eligible for the seal: {sorted(existing)} (must be gtfs, "
+ "operational_status=published, and status not in (deprecated, development))"
+ )
+ summary = "; ".join(problems)
+
+ if not evaluated:
+ # The run did nothing at all, which a log line is too quiet for.
+ raise ValueError(f"no requested feed can be evaluated — {summary}")
+
+ # Only a warning, so one stale id does not cost a run over fifty feeds.
+ logging.warning("Skipping %d of the requested feed(s): %s", len(unusable), summary)
+
+
+def _load_previous_states(
+ db_session: Session, feed_ids: Sequence[str]
+) -> Dict[Tuple[str, str], SealCriterionState]:
+ """Map (feed_id, criterion) -> stored state for a batch of feeds."""
+ if not feed_ids:
+ return {}
+ rows = db_session.execute(
+ select(CRITERION_TABLE).where(CRITERION_TABLE.c.feed_id.in_(list(feed_ids)))
+ ).all()
+ return {
+ (row.feed_id, row.criterion): SealCriterionState(
+ feed_id=row.feed_id,
+ criterion=SealCriterionName(row.criterion),
+ observed_pass=row.observed_pass,
+ confirmed_pass=row.confirmed_pass,
+ evaluated_at=row.evaluated_at,
+ first_observed_failure_at=row.first_observed_failure_at,
+ last_observed_failure_at=row.last_observed_failure_at,
+ last_confirmed_failure_at=row.last_confirmed_failure_at,
+ probation_start=row.probation_start,
+ )
+ for row in rows
+ }
+
+
+def _load_previous_seals(
+ db_session: Session, feed_ids: Sequence[str]
+) -> Dict[str, bool]:
+ """Map feed_id -> stored has_seal, for feeds that already have a seal row."""
+ if not feed_ids:
+ return {}
+ rows = db_session.execute(
+ select(SEAL_TABLE.c.feed_id, SEAL_TABLE.c.has_seal).where(
+ SEAL_TABLE.c.feed_id.in_(list(feed_ids))
+ )
+ ).all()
+ return {row.feed_id: bool(row.has_seal) for row in rows}
+
+
+def _roll_up_has_seal(states: Dict[str, SealCriterionState]) -> bool:
+ """True when every criterion in service is a confirmed pass and not on probation.
+
+ A criterion is *in service* once it has produced a verdict at any point
+ (`observed_pass is not None`). One that never has is skipped rather than counted as a
+ failure, which is what lets the seal be computed before every criterion has a data
+ source: a criterion whose source only starts collecting later sits at NULL until then
+ and simply is not part of the roll-up.
+
+ That waiver is self-limiting because `transition` never writes NULL back, so a
+ criterion can only be skipped before its first verdict ever. Once it has produced one it
+ stays in the roll-up with its last verdict, and an upstream outage freezes it rather
+ than quietly waiving it.
+
+ The non-empty guard stops the roll-up being vacuously true: "every criterion in service
+ qualifies" holds trivially for a feed nothing has ever been measured on.
+
+ `confirmed_pass is True` rather than a bare truthiness test is defensive; within the
+ in-service set it cannot be NULL, since the two booleans are always written together.
+ """
+ in_service = [state for state in states.values() if state.observed_pass is not None]
+ if not in_service:
+ return False
+ return all(
+ state.confirmed_pass is True and state.probation_start is None
+ for state in in_service
+ )
+
+
+def _is_notable(
+ previous: Optional[SealCriterionState], state: Optional[SealCriterionState]
+) -> bool:
+ """True when this evaluation moved a criterion's verdict.
+
+ A first evaluation is not notable on its own: the initial run over the whole catalogue
+ would otherwise report every feed, which is exactly the unbounded payload this is meant
+ to avoid. The `first_evaluations` count covers that case instead. A first evaluation
+ that lands on a failure *is* reported, since that is the actionable half.
+ """
+ if state is None:
+ return False
+ if previous is None:
+ return state.observed_pass is False or state.confirmed_pass is False
+ return (
+ state.observed_pass != previous.observed_pass
+ or state.confirmed_pass != previous.confirmed_pass
+ )
+
+
+def _upsert_criteria(
+ db_session: Session, states: Sequence[SealCriterionState], now: datetime
+) -> None:
+ """Insert or update sealcriterion rows for the given states."""
+ if not states:
+ return
+ payload = [
+ {
+ "feed_id": state.feed_id,
+ "criterion": state.criterion.value,
+ "observed_pass": state.observed_pass,
+ "confirmed_pass": state.confirmed_pass,
+ "evaluated_at": state.evaluated_at,
+ "first_observed_failure_at": state.first_observed_failure_at,
+ "last_observed_failure_at": state.last_observed_failure_at,
+ "last_confirmed_failure_at": state.last_confirmed_failure_at,
+ "probation_start": state.probation_start,
+ "updated_at": now,
+ }
+ for state in states
+ ]
+ statement = insert(CRITERION_TABLE).values(payload)
+ db_session.execute(
+ statement.on_conflict_do_update(
+ index_elements=[CRITERION_TABLE.c.feed_id, CRITERION_TABLE.c.criterion],
+ set_={
+ "observed_pass": statement.excluded.observed_pass,
+ "confirmed_pass": statement.excluded.confirmed_pass,
+ "evaluated_at": statement.excluded.evaluated_at,
+ "first_observed_failure_at": statement.excluded.first_observed_failure_at,
+ "last_observed_failure_at": statement.excluded.last_observed_failure_at,
+ "last_confirmed_failure_at": statement.excluded.last_confirmed_failure_at,
+ "probation_start": statement.excluded.probation_start,
+ "updated_at": statement.excluded.updated_at,
+ },
+ )
+ )
+
+
+def _upsert_seals(db_session: Session, outcomes: Sequence[dict], now: datetime) -> None:
+ """Insert or update feedreliabilityseal rows.
+
+ A row is written for every evaluated feed, whether or not it qualifies. That matters
+ beyond bookkeeping: `created_at` on this row is what a criterion measuring "how long
+ have we been tracking this feed" reads, so a feed that does not qualify still needs one,
+ or such a criterion could never start counting and the feed could never come to qualify.
+
+ seal_earned_at and seal_lost_at are written only when has_seal actually changes, so
+ they record transitions rather than the time of the last evaluation.
+ """
+ for outcome in outcomes:
+ row = {
+ "feed_id": outcome["feed_id"],
+ "has_seal": outcome["has_seal"],
+ "updated_at": now,
+ }
+ if outcome["granted"]:
+ row["seal_earned_at"] = now
+ elif outcome["revoked"]:
+ row["seal_lost_at"] = now
+
+ statement = insert(SEAL_TABLE).values(**row)
+ update_set = {
+ "has_seal": statement.excluded.has_seal,
+ "updated_at": statement.excluded.updated_at,
+ }
+ if "seal_earned_at" in row:
+ update_set["seal_earned_at"] = statement.excluded.seal_earned_at
+ if "seal_lost_at" in row:
+ update_set["seal_lost_at"] = statement.excluded.seal_lost_at
+ db_session.execute(
+ statement.on_conflict_do_update(
+ index_elements=[SEAL_TABLE.c.feed_id], set_=update_set
+ )
+ )
+
+
+@with_db_session
+def update_seals(
+ db_session: Session,
+ dry_run: bool = True,
+ stable_feed_ids: Optional[Sequence[str]] = None,
+ limit: Optional[int] = None,
+ criteria: Optional[Sequence[str]] = None,
+ batch_size: int = DEFAULT_BATCH_SIZE,
+ now: Optional[datetime] = None,
+ max_reported_feeds: int = DEFAULT_MAX_REPORTED_FEEDS,
+) -> dict:
+ """Evaluate the seal criteria for every eligible feed and store the result.
+
+ Every feed is evaluated before anything is written, so a dry run exercises exactly the
+ same code path as a real run.
+
+ Args:
+ db_session: SQLAlchemy session, injected by @with_db_session.
+ dry_run: Evaluate and report without writing. Default True.
+ stable_feed_ids: Evaluate only these feeds. Unknown or ineligible ids are skipped
+ with a logged warning; it raises only if none of them can be evaluated. When
+ given, `evaluations` reports every criterion of those feeds rather than only
+ the ones whose verdict moved.
+ limit: Cap the number of feeds evaluated.
+ criteria: Evaluate only these criteria. A partial set skips the has_seal roll-up,
+ since the criteria that were not evaluated cannot be judged.
+ batch_size: Feeds loaded per query batch.
+ now: Evaluation timestamp. Defaults to the current UTC time.
+ max_reported_feeds: Cap on the `feeds` list in the report. The count dropped is
+ always returned as `feeds_omitted`.
+
+ Returns:
+ A report dict.
+ """
+ started = time.monotonic()
+ now = now or datetime.now(timezone.utc)
+ evaluators = _resolve_evaluators(criteria)
+ partial_run = len(evaluators) < len(EVALUATORS)
+
+ query = get_seal_feeds_query(db_session, stable_feed_ids=stable_feed_ids)
+ if limit is not None:
+ query = query.limit(limit)
+ feeds = query.all()
+
+ if stable_feed_ids is not None:
+ _validate_requested_feed_ids(
+ db_session, stable_feed_ids, {feed.stable_id for feed in feeds}
+ )
+
+ logging.info(
+ "Evaluating %d criterion/criteria for %d feed(s) (dry_run=%s, now=%s).",
+ len(evaluators),
+ len(feeds),
+ dry_run,
+ now.isoformat(),
+ )
+
+ all_states: List[SealCriterionState] = []
+ outcomes: List[dict] = []
+ feed_reports: List[dict] = []
+ not_evaluable = 0
+ first_evaluations = 0
+ is_feed_list_provided = stable_feed_ids is not None
+
+ for batch in batched(feeds, batch_size):
+ batch_ids = [feed.id for feed in batch]
+ contexts = build_contexts(db_session, batch, now)
+ previous_states = _load_previous_states(db_session, batch_ids)
+ previous_seals = _load_previous_seals(db_session, batch_ids)
+
+ for feed in batch:
+ ctx = contexts[feed.id]
+ feed_states: Dict[str, SealCriterionState] = {}
+ criteria_report: List[dict] = []
+ anything_moved = False
+
+ for evaluator in evaluators:
+ observation = evaluator.evaluate(ctx)
+ previous = previous_states.get((feed.id, evaluator.name.value))
+ state = transition(
+ prev=previous,
+ observation=observation,
+ grace_period=evaluator.grace_period,
+ probation_period=evaluator.probation_period,
+ now=now,
+ feed_id=feed.id,
+ )
+ if observation.observed_pass is None:
+ not_evaluable += 1
+ if state is not None:
+ feed_states[evaluator.name.value] = state
+ if state is not previous:
+ all_states.append(state)
+ if previous is None and state is not None:
+ first_evaluations += 1
+
+ if _is_notable(previous, state):
+ anything_moved = True
+
+ criteria_report.append(
+ {
+ "criterion": evaluator.name.value,
+ "observed_pass": observation.observed_pass,
+ "confirmed_pass": (
+ state.confirmed_pass if state is not None else None
+ ),
+ "previously_confirmed_pass": (
+ previous.confirmed_pass if previous is not None else None
+ ),
+ "on_probation": (
+ state.probation_start is not None
+ if state is not None
+ else None
+ ),
+ "reason": observation.reason,
+ }
+ )
+
+ if partial_run:
+ # No roll-up on a partial run, so there is no seal state to report.
+ if is_feed_list_provided or anything_moved:
+ feed_reports.append(
+ {"stable_id": ctx.stable_id, "criteria": criteria_report}
+ )
+ continue
+
+ # Merge in any stored criteria this run did not produce a new state for, so the
+ # roll-up sees the full set even when an evaluator returned "not evaluable".
+ merged = {
+ criterion: state
+ for (owner_id, criterion), state in previous_states.items()
+ if owner_id == feed.id
+ }
+ merged.update(feed_states)
+
+ # A feed with no seal row yet is treated as not holding one, so a first run can
+ # grant the seal but can never withdraw one: nothing was held to lose.
+ had_seal = previous_seals.get(feed.id)
+ has_seal = _roll_up_has_seal(merged)
+ outcome = {
+ "feed_id": feed.id,
+ "stable_id": ctx.stable_id,
+ "had_seal": bool(had_seal),
+ "has_seal": has_seal,
+ # A first evaluation is a grant if it passes, but it is not a loss if
+ # it fails: nothing was held, so nothing was lost. Only these two
+ # flags stamp seal_earned_at / seal_lost_at.
+ "granted": has_seal and not had_seal,
+ "revoked": bool(had_seal) and not has_seal,
+ }
+ outcomes.append(outcome)
+
+ # A feed is reported when the caller asked for it by name, when one of its
+ # criteria moved, or when its seal changed. A run over a quiet catalogue with no
+ # feed list therefore reports nothing, which keeps a nightly response small.
+ if (
+ is_feed_list_provided
+ or anything_moved
+ or outcome["granted"]
+ or outcome["revoked"]
+ ):
+ feed_reports.append(
+ {
+ "stable_id": ctx.stable_id,
+ "had_seal": outcome["had_seal"],
+ "has_seal": has_seal,
+ "criteria": criteria_report,
+ }
+ )
+
+ revoked = [outcome for outcome in outcomes if outcome["revoked"]]
+ granted = [outcome for outcome in outcomes if outcome["granted"]]
+ if not dry_run:
+ for state_batch in batched(all_states, batch_size):
+ _upsert_criteria(db_session, state_batch, now)
+ db_session.commit()
+ for outcome_batch in batched(outcomes, batch_size):
+ _upsert_seals(db_session, outcome_batch, now)
+ db_session.commit()
+
+ report = {
+ "message": (
+ f"{'Dry run: evaluated' if dry_run else 'Updated'} {len(feeds)} feed(s) "
+ f"across {len(evaluators)} criterion/criteria."
+ ),
+ "dry_run": dry_run,
+ "evaluated_at": now.isoformat(),
+ "total_feeds": len(feeds),
+ "criteria": [evaluator.name.value for evaluator in evaluators],
+ "partial_run": partial_run,
+ "criterion_rows_written": 0 if dry_run else len(all_states),
+ "not_evaluable": not_evaluable,
+ "first_evaluations": first_evaluations,
+ # seals_before_run + seals_granted - seals_revoked == seals_after_run.
+ # On a dry run the "after" figure is what would be stored, not what is.
+ "seals_before_run": sum(1 for outcome in outcomes if outcome["had_seal"]),
+ "seals_after_run": sum(1 for outcome in outcomes if outcome["has_seal"]),
+ "seals_granted": len(granted),
+ "seals_revoked": len(revoked),
+ # The two transitions in feedreliabilityseal, by feed. Counts alone cannot say
+ # which feed moved, and that is the first thing anyone asks of a run.
+ "granted_stable_ids": [outcome["stable_id"] for outcome in granted],
+ "revoked_stable_ids": [outcome["stable_id"] for outcome in revoked],
+ "elapsed_seconds": round(time.monotonic() - started, 2),
+ }
+ if partial_run:
+ report["note"] = (
+ "Partial criteria run: has_seal was not recalculated because the criteria "
+ "that were not evaluated cannot be judged."
+ )
+ # A sample, not the record: the two seal tables hold every verdict. The omitted count
+ # is always present so a truncated list can never be mistaken for the whole story.
+ report["feeds"] = feed_reports[:max_reported_feeds]
+ report["feeds_omitted"] = max(0, len(feed_reports) - max_reported_feeds)
+
+ # Log without `feeds`: Cloud Logging drops a LogEntry over 256 KB, so a run naming a
+ # few hundred feeds would lose the whole log entry. The counts belong in logs; `feeds`
+ # is for the caller reading the response.
+ logging.info(
+ "Task completed: %s",
+ {key: value for key, value in report.items() if key != "feeds"},
+ )
+ return report
diff --git a/functions-python/tasks_executor/src/tasks/seal_of_reliability/state_machine.py b/functions-python/tasks_executor/src/tasks/seal_of_reliability/state_machine.py
new file mode 100644
index 000000000..de14d4446
--- /dev/null
+++ b/functions-python/tasks_executor/src/tasks/seal_of_reliability/state_machine.py
@@ -0,0 +1,194 @@
+#
+# MobilityData 2026
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""Generic per-criterion state machine for the Seal of Reliability.
+
+Steps 2 to 4 of the nightly job: observed failure tracking, the confirmed status, and
+probation. This is the only place that knows about grace periods and probation, and it is
+criterion-agnostic — the caller passes both from the evaluator. `now` is a parameter rather
+than a call to `datetime.now()` so runs are replayable and idempotent.
+
+Two pieces of state are tracked per feed per criterion and they are independent:
+
+* `confirmed_pass` — does the criterion pass right now, debounced by its grace period.
+* `probation_start` — the stretch it must serve after recovering from a confirmed failure.
+
+The seal (step 5, in seal_updater) requires every criterion in service to be a confirmed
+pass and not on probation.
+"""
+
+from dataclasses import dataclass, replace
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+
+from tasks.seal_of_reliability.criteria import SealCriterionName
+from tasks.seal_of_reliability.evaluators.base import CriterionObservation
+
+
+@dataclass(frozen=True)
+class SealCriterionState:
+ """One row of the sealcriterion table.
+
+ Both booleans are stored positively: TRUE means the criterion passed. `observed_pass`
+ is the instantaneous check, `confirmed_pass` the debounced status that drives the seal.
+ NULL on either means the criterion has never produced a verdict for this feed, and the
+ job never writes NULL back — see `transition`.
+
+ The failure timestamps stay negative on purpose: they record events that really are
+ failures. Booleans describe state, timestamps record events.
+ """
+
+ feed_id: str
+ criterion: SealCriterionName
+ observed_pass: Optional[bool] = None
+ confirmed_pass: Optional[bool] = None
+ evaluated_at: Optional[datetime] = None
+ first_observed_failure_at: Optional[datetime] = None
+ last_observed_failure_at: Optional[datetime] = None
+ last_confirmed_failure_at: Optional[datetime] = None
+ probation_start: Optional[datetime] = None
+
+
+def _next_day_start(moment: datetime) -> datetime:
+ """The start of the day after `moment`, in UTC.
+
+ Probation is stamped at day granularity for two reasons: two runs on the same day
+ produce the same start, and stamping *tomorrow* on every failing day leaves the start on
+ the day the criterion was repaired once the streak ends.
+ """
+ if moment.tzinfo is not None:
+ moment = moment.astimezone(timezone.utc)
+ return moment.replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=1)
+
+
+def _probation_start(
+ base: SealCriterionState,
+ observed_pass: bool,
+ confirmed_pass: bool,
+ probation_period: Optional[timedelta],
+ now: datetime,
+) -> Optional[datetime]:
+ """Step 4: probation is only ever opened by a recovery.
+
+ Two rules put a criterion on probation, and both are recoveries:
+
+ * recovery from a confirmed failure, in all circumstances;
+ * recovery from an observed failure while already on probation.
+
+ A first evaluation that passes is not a recovery, so it opens nothing. An observed
+ failure absorbed by the grace period costs a criterion that is not on probation nothing,
+ while the same failure during probation restarts the whole count.
+ """
+ if probation_period is None:
+ return None
+
+ if not observed_pass:
+ if base.probation_start is not None or not confirmed_pass:
+ return _next_day_start(now)
+ # Not on probation and this failure is still inside its grace period: the track
+ # record survives the blip.
+ return base.probation_start
+
+ # The check passed today. If the criterion was serving probation and the whole stretch
+ # has now gone by since it started, it has served it: clear it. Otherwise leave the start
+ # exactly where it is — a passing day never moves it, it only brings the end closer.
+ if (
+ base.probation_start is not None
+ and now >= base.probation_start + probation_period
+ ):
+ return None
+ return base.probation_start
+
+
+def transition(
+ prev: Optional[SealCriterionState],
+ observation: CriterionObservation,
+ grace_period: Optional[timedelta],
+ probation_period: Optional[timedelta],
+ now: datetime,
+ feed_id: Optional[str] = None,
+) -> Optional[SealCriterionState]:
+ """Apply one observation to a criterion's stored state.
+
+ Returns the new state, or `prev` unchanged when the criterion was not evaluable
+ (`observation.observed_pass is None`) — a missing input must never be read as a failure,
+ otherwise an upstream outage would withdraw seals across the catalogue. That is also
+ what keeps `observed_pass IS NULL` meaning "never evaluated" rather than "not evaluable
+ last night", which step 5 relies on to decide what is in service.
+
+ Args:
+ prev: The stored state, or None if this criterion has never been evaluated.
+ observation: The evaluator's verdict for this run.
+ grace_period: How long an observed failure streak may last before the status flips.
+ None means the status flips on the first failing day.
+ probation_period: How long the criterion must go with no observed failure after
+ recovering from a confirmed failure. None means the criterion has no probation.
+ now: The evaluation timestamp.
+ feed_id: Required when `prev` is None, to build the first state.
+ """
+ if observation.observed_pass is None:
+ return prev
+
+ resolved_feed_id = prev.feed_id if prev is not None else feed_id
+ if resolved_feed_id is None:
+ raise ValueError("feed_id is required when there is no previous state")
+
+ base = prev or SealCriterionState(
+ feed_id=resolved_feed_id, criterion=observation.criterion
+ )
+
+ # A criterion that has never produced a verdict gets no grace period. The grace period
+ # holds a pass the criterion has actually earned, and one we have never seen pass has
+ # nothing to hold — otherwise a feed that is broken the first time we look at it would
+ # be reported as passing for up to a month on evidence we do not have.
+ first_evaluation = base.observed_pass is None
+
+ if observation.observed_pass:
+ # The streak is over, so the grace period resets. The two `last_*` timestamps are
+ # history and are never cleared.
+ first_observed_failure_at = None
+ last_observed_failure_at = base.last_observed_failure_at
+ last_confirmed_failure_at = base.last_confirmed_failure_at
+ confirmed_pass = True
+ else:
+ # first_observed_failure_at marks the start of the current streak. It is cleared on
+ # recovery, so it is only None here at the start of a new streak.
+ first_observed_failure_at = base.first_observed_failure_at or now
+ last_observed_failure_at = now
+ confirmed_pass = (
+ grace_period is not None
+ and not first_evaluation
+ and now - first_observed_failure_at < grace_period
+ )
+ last_confirmed_failure_at = (
+ base.last_confirmed_failure_at if confirmed_pass else now
+ )
+
+ return replace(
+ base,
+ observed_pass=observation.observed_pass,
+ confirmed_pass=confirmed_pass,
+ evaluated_at=now,
+ first_observed_failure_at=first_observed_failure_at,
+ last_observed_failure_at=last_observed_failure_at,
+ last_confirmed_failure_at=last_confirmed_failure_at,
+ probation_start=_probation_start(
+ base=base,
+ observed_pass=observation.observed_pass,
+ confirmed_pass=confirmed_pass,
+ probation_period=probation_period,
+ now=now,
+ ),
+ )
diff --git a/functions-python/tasks_executor/src/tasks/seal_of_reliability/update_seal_of_reliability.py b/functions-python/tasks_executor/src/tasks/seal_of_reliability/update_seal_of_reliability.py
new file mode 100644
index 000000000..64eeef6c6
--- /dev/null
+++ b/functions-python/tasks_executor/src/tasks/seal_of_reliability/update_seal_of_reliability.py
@@ -0,0 +1,79 @@
+#
+# MobilityData 2026
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""Task entry point for the nightly Seal of Reliability evaluation (issue #1761)."""
+
+from datetime import datetime
+
+from tasks.seal_of_reliability.seal_updater import (
+ DEFAULT_BATCH_SIZE,
+ DEFAULT_MAX_REPORTED_FEEDS,
+ update_seals,
+)
+
+
+def get_parameters(payload: dict):
+ """Read the task parameters from the payload, applying defaults."""
+ now = payload.get("now")
+ return (
+ payload.get("dry_run", True),
+ payload.get("stable_feed_ids", None),
+ payload.get("limit", None),
+ payload.get("criteria", None),
+ payload.get("batch_size", DEFAULT_BATCH_SIZE),
+ datetime.fromisoformat(now) if now else None,
+ payload.get("max_reported_feeds", DEFAULT_MAX_REPORTED_FEEDS),
+ )
+
+
+def update_seal_of_reliability_handler(payload: dict) -> dict:
+ """
+ Handler for the nightly Seal of Reliability evaluation.
+
+ Payload parameters:
+ dry_run (bool): Evaluate every feed and return the report without writing.
+ Default: True.
+ stable_feed_ids (list[str] | None): Evaluate only these feeds. Unknown ids raise.
+ When set, `evaluations` covers every criterion of those feeds.
+ Default: None (all eligible feeds).
+ limit (int | None): Cap the number of feeds evaluated. Default: no limit.
+ criteria (list[str] | None): Evaluate only these criteria. A partial set skips the
+ has_seal roll-up. Default: None (every implemented criterion).
+ batch_size (int): Feeds loaded per query batch. Every eligible feed is still
+ evaluated; this only sizes the queries. Default: 200.
+ now (str | None): ISO timestamp to evaluate against, for replays and backfills.
+ Default: current UTC time.
+ max_reported_feeds (int): Cap on the `feeds` list in the response. Everything is
+ still evaluated and written; `feeds_omitted` reports how many
+ entries were left out. Default: 50.
+ """
+ (
+ dry_run,
+ stable_feed_ids,
+ limit,
+ criteria,
+ batch_size,
+ now,
+ max_reported_feeds,
+ ) = get_parameters(payload)
+ return update_seals(
+ dry_run=dry_run,
+ stable_feed_ids=stable_feed_ids,
+ limit=limit,
+ criteria=criteria,
+ batch_size=batch_size,
+ now=now,
+ max_reported_feeds=max_reported_feeds,
+ )
diff --git a/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_end_to_end_db.py b/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_end_to_end_db.py
new file mode 100644
index 000000000..437ca5d76
--- /dev/null
+++ b/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_end_to_end_db.py
@@ -0,0 +1,308 @@
+#
+# MobilityData 2026
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""End-to-end test of the Seal of Reliability task through the tasks_executor entry point.
+
+Unlike test_seal_updater_db.py, which calls `update_seals` directly, this drives the
+function the way Cloud Scheduler does: a JSON payload posted to `tasks_executor`, dispatched
+by name through the registry in main.py. That covers the layers the direct tests skip —
+payload parsing, defaults, and the task being reachable by its registered name.
+
+The task resolves its own session from FEEDS_DATABASE_URL rather than taking a `db_url`, so
+the environment is pointed at the test database for the duration of each test and the
+Database singleton is reset around it.
+
+These runs are unnamed, so they evaluate every eligible feed in the test database — which
+includes the fixtures seeded by conftest.pytest_sessionstart. Assertions are therefore
+scoped to this module's own PREFIX, and the report counts are checked as invariants and
+deltas rather than absolutes.
+"""
+
+import os
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import MagicMock, patch
+
+import flask
+from main import tasks_executor
+from sqlalchemy import delete, select
+
+from shared.database.database import with_db_session
+from shared.database_gen.sqlacodegen_models import (
+ Feed,
+ Feedreliabilityseal,
+ Gtfsfeed,
+ Sealcriterion,
+)
+from test_shared.test_utils.database_utils import default_db_url, reset_database_class
+
+NOW = datetime(2026, 6, 1, 12, 0, tzinfo=timezone.utc)
+PREFIX = "seal_e2e_"
+
+OFFICIAL = f"{PREFIX}official"
+NOT_OFFICIAL = f"{PREFIX}not_official"
+UNKNOWN_OFFICIAL = f"{PREFIX}unknown_official"
+DEPRECATED = f"{PREFIX}deprecated"
+UNPUBLISHED = f"{PREFIX}unpublished"
+
+
+def _seed(db_session, feed_id, official=True, status="active", operational="published"):
+ db_session.add(
+ Gtfsfeed(
+ id=feed_id,
+ stable_id=feed_id,
+ data_type="gtfs",
+ status=status,
+ operational_status=operational,
+ official=official,
+ created_at=NOW - timedelta(days=400),
+ producer_url=f"https://example.com/{feed_id}.zip",
+ )
+ )
+ db_session.flush()
+
+
+def _cleanup(db_session):
+ """Deleting the parent Feed cascades to gtfsfeed and both seal tables."""
+ db_session.execute(delete(Feed).where(Feed.stable_id.like(f"{PREFIX}%")))
+ db_session.commit()
+
+
+class TestSealTaskEndToEnd(unittest.TestCase):
+ """Seed, run through the entry point, inspect, modify, run again, inspect."""
+
+ @with_db_session(db_url=default_db_url)
+ def setUp(self, db_session):
+ _cleanup(db_session)
+ _seed(db_session, OFFICIAL)
+ _seed(db_session, NOT_OFFICIAL, official=False)
+ _seed(db_session, UNKNOWN_OFFICIAL, official=None)
+ _seed(db_session, DEPRECATED, status="deprecated")
+ _seed(db_session, UNPUBLISHED, operational="unpublished")
+ db_session.commit()
+ self.app = flask.Flask(__name__)
+
+ @with_db_session(db_url=default_db_url)
+ def tearDown(self, db_session):
+ _cleanup(db_session)
+ reset_database_class()
+
+ def run_task(self, payload: dict) -> dict:
+ """Invoke tasks_executor with a hand-built request, as Cloud Scheduler would.
+
+ FEEDS_DATABASE_URL is pointed at the test database because the handler resolves its
+ own session; without this the task would run against the local development DB.
+ """
+ request = MagicMock(spec=flask.Request)
+ request.get_json.return_value = {
+ "task": "update_seal_of_reliability",
+ "payload": payload,
+ }
+ request.headers = {}
+
+ reset_database_class()
+ with patch.dict(os.environ, {"FEEDS_DATABASE_URL": default_db_url}):
+ with self.app.app_context():
+ response = tasks_executor(request)
+ reset_database_class()
+
+ self.assertEqual(response.status_code, 200, response.get_data(as_text=True))
+ return response.get_json()
+
+ @staticmethod
+ def ours(report: dict) -> list:
+ """The report's feed entries for this module's feeds only.
+
+ Unnamed runs also cover the conftest fixtures, so filtering keeps these assertions
+ independent of what else lives in the test database.
+ """
+ return [row for row in report["feeds"] if row["stable_id"].startswith(PREFIX)]
+
+ @staticmethod
+ @with_db_session(db_url=default_db_url)
+ def seal_state(db_session):
+ """stable_id -> (has_seal, earned_at set?, lost_at set?) for the seeded feeds."""
+ seal = Feedreliabilityseal.__table__
+ rows = db_session.execute(
+ select(
+ Feed.stable_id,
+ seal.c.has_seal,
+ seal.c.seal_earned_at,
+ seal.c.seal_lost_at,
+ )
+ .join(seal, seal.c.feed_id == Feed.id)
+ .where(Feed.stable_id.like(f"{PREFIX}%"))
+ ).all()
+ return {
+ row.stable_id: (
+ row.has_seal,
+ row.seal_earned_at is not None,
+ row.seal_lost_at is not None,
+ )
+ for row in rows
+ }
+
+ @staticmethod
+ @with_db_session(db_url=default_db_url)
+ def criterion_state(db_session):
+ """stable_id -> the sealcriterion row, for the seeded feeds."""
+ criterion = Sealcriterion.__table__
+ rows = db_session.execute(
+ select(Feed.stable_id, criterion)
+ .join(criterion, criterion.c.feed_id == Feed.id)
+ .where(Feed.stable_id.like(f"{PREFIX}%"))
+ ).all()
+ return {row.stable_id: row for row in rows}
+
+ @staticmethod
+ @with_db_session(db_url=default_db_url)
+ def set_official(stable_id, official, db_session):
+ db_session.execute(
+ Feed.__table__.update()
+ .where(Feed.__table__.c.stable_id == stable_id)
+ .values(official=official)
+ )
+ db_session.commit()
+
+ def test_dry_run_through_the_entry_point_writes_nothing(self):
+ report = self.run_task({"dry_run": True, "now": NOW.isoformat()})
+ self.assertTrue(report["dry_run"])
+ self.assertGreaterEqual(report["total_feeds"], 3)
+ self.assertEqual(report["criterion_rows_written"], 0)
+ self.assertEqual(self.seal_state(), {}, "nothing written for our feeds")
+ self.assertEqual(self.criterion_state(), {})
+
+ def test_dry_run_is_the_default_when_the_payload_omits_it(self):
+ """The safety default has to survive the payload layer, not just the function."""
+ report = self.run_task({"now": NOW.isoformat()})
+ self.assertTrue(report["dry_run"])
+ self.assertEqual(self.seal_state(), {})
+
+ def test_two_runs_with_a_change_in_between(self):
+ # --- first run: writes the initial state
+ first = self.run_task({"dry_run": False, "now": NOW.isoformat()})
+ self.assertFalse(first["dry_run"])
+ self.assertGreaterEqual(first["criterion_rows_written"], 3)
+ self.assertEqual(first["seals_revoked"], 0, "nothing was held beforehand")
+ self.assertEqual(
+ {row["stable_id"] for row in self.ours(first)},
+ {OFFICIAL, NOT_OFFICIAL, UNKNOWN_OFFICIAL},
+ "the two failures moved a criterion; the official feed gained the seal",
+ )
+
+ # --- inspect the database
+ self.assertEqual(
+ self.seal_state(),
+ {
+ OFFICIAL: (True, True, False),
+ NOT_OFFICIAL: (False, False, False),
+ UNKNOWN_OFFICIAL: (False, False, False),
+ },
+ "only the official feed earned it; the others were never granted or lost",
+ )
+ criteria = self.criterion_state()
+ self.assertNotIn(DEPRECATED, criteria)
+ self.assertNotIn(UNPUBLISHED, criteria)
+ self.assertTrue(criteria[OFFICIAL].confirmed_pass)
+ self.assertIsNone(criteria[OFFICIAL].first_observed_failure_at)
+ self.assertFalse(criteria[NOT_OFFICIAL].confirmed_pass)
+ self.assertEqual(criteria[NOT_OFFICIAL].first_observed_failure_at, NOW)
+
+ # --- modify the database: revoke one, recover another
+ self.set_official(OFFICIAL, False)
+ self.set_official(NOT_OFFICIAL, True)
+
+ # --- second run
+ later = NOW + timedelta(days=1)
+ second = self.run_task({"dry_run": False, "now": later.isoformat()})
+ self.assertEqual(second["first_evaluations"], 0, "every feed already had a row")
+ self.assertEqual(second["seals_granted"], 1)
+ self.assertEqual(second["seals_revoked"], 1)
+ self.assertIn(OFFICIAL, second["revoked_stable_ids"])
+ self.assertEqual(
+ second["seals_before_run"]
+ + second["seals_granted"]
+ - second["seals_revoked"],
+ second["seals_after_run"],
+ "before + granted - revoked == after",
+ )
+
+ moved = {row["stable_id"]: row for row in self.ours(second)}
+ self.assertEqual(set(moved), {OFFICIAL, NOT_OFFICIAL}, "only these two moved")
+
+ self.assertTrue(moved[OFFICIAL]["had_seal"])
+ self.assertFalse(moved[OFFICIAL]["has_seal"])
+ self.assertFalse(moved[OFFICIAL]["criteria"][0]["confirmed_pass"])
+ self.assertTrue(moved[OFFICIAL]["criteria"][0]["previously_confirmed_pass"])
+
+ self.assertFalse(moved[NOT_OFFICIAL]["had_seal"])
+ self.assertTrue(moved[NOT_OFFICIAL]["has_seal"])
+ self.assertTrue(moved[NOT_OFFICIAL]["criteria"][0]["confirmed_pass"])
+ self.assertFalse(
+ moved[NOT_OFFICIAL]["criteria"][0]["previously_confirmed_pass"]
+ )
+
+ # --- inspect again: both transitions are recorded, history is preserved
+ self.assertEqual(
+ self.seal_state(),
+ {
+ OFFICIAL: (False, True, True),
+ NOT_OFFICIAL: (True, True, False),
+ UNKNOWN_OFFICIAL: (False, False, False),
+ },
+ "the revoked feed keeps its earned_at; the recovered one gains earned_at",
+ )
+ criteria = self.criterion_state()
+ self.assertEqual(criteria[OFFICIAL].first_observed_failure_at, later)
+ self.assertIsNone(
+ criteria[NOT_OFFICIAL].first_observed_failure_at, "the streak ended"
+ )
+ self.assertEqual(
+ criteria[NOT_OFFICIAL].last_observed_failure_at,
+ NOW,
+ "history is never cleared, so the old failure time survives recovery",
+ )
+
+ def test_third_run_with_no_change_is_a_no_op(self):
+ self.run_task({"dry_run": False, "now": NOW.isoformat()})
+ before = self.criterion_state()
+
+ later = NOW + timedelta(days=2)
+ report = self.run_task({"dry_run": False, "now": later.isoformat()})
+ self.assertEqual(self.ours(report), [], "none of our feeds moved")
+ self.assertEqual(report["first_evaluations"], 0)
+ self.assertEqual(report["seals_before_run"], report["seals_after_run"])
+
+ after = self.criterion_state()
+ for stable_id, row in after.items():
+ with self.subTest(stable_id=stable_id):
+ self.assertEqual(
+ row.first_observed_failure_at,
+ before[stable_id].first_observed_failure_at,
+ "a re-evaluation must not restart a failure streak",
+ )
+ self.assertEqual(row.evaluated_at, later, "but it is re-evaluated")
+
+ def test_unknown_task_name_is_rejected(self):
+ request = MagicMock(spec=flask.Request)
+ request.get_json.return_value = {"task": "no_such_seal_task", "payload": {}}
+ request.headers = {}
+ with self.app.app_context():
+ response = tasks_executor(request)
+ self.assertEqual(response.status_code, 400)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_evaluators.py b/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_evaluators.py
new file mode 100644
index 000000000..0d897ab53
--- /dev/null
+++ b/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_evaluators.py
@@ -0,0 +1,94 @@
+#
+# MobilityData 2026
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""Unit tests for the seal criterion evaluators. No database."""
+
+import unittest
+from datetime import datetime, timezone
+
+from tasks.seal_of_reliability.context import FeedSealContext
+from tasks.seal_of_reliability.criteria import SealCriterionName
+from tasks.seal_of_reliability.evaluators import (
+ EVALUATORS,
+ CriterionEvaluator,
+ OfficialEvaluator,
+)
+
+NOW = datetime(2026, 6, 1, tzinfo=timezone.utc)
+
+
+def _ctx(**overrides) -> FeedSealContext:
+ defaults = {"feed_id": "feed-1", "now": NOW, "stable_id": "mdb-1"}
+ defaults.update(overrides)
+ return FeedSealContext(**defaults)
+
+
+class TestBaseClass(unittest.TestCase):
+ def test_subclass_must_implement_evaluate(self):
+ class Incomplete(CriterionEvaluator):
+ name = SealCriterionName.OFFICIAL
+
+ with self.assertRaises(NotImplementedError):
+ Incomplete().evaluate(_ctx())
+
+ def test_result_is_labelled_with_the_evaluator_name(self):
+ for evaluator in EVALUATORS:
+ with self.subTest(criterion=evaluator.name):
+ result = evaluator.evaluate(_ctx(official=True))
+ self.assertEqual(result.criterion, evaluator.name)
+
+ def test_every_result_carries_a_reason(self):
+ for evaluator in EVALUATORS:
+ with self.subTest(criterion=evaluator.name):
+ self.assertTrue(evaluator.evaluate(_ctx()).reason)
+
+ def test_registry_has_no_duplicate_criteria(self):
+ names = [evaluator.name for evaluator in EVALUATORS]
+ self.assertEqual(sorted(names), sorted(set(names)))
+
+ def test_registry_only_uses_known_criteria(self):
+ """Every registered evaluator must map onto a value of the DB enum."""
+ for evaluator in EVALUATORS:
+ with self.subTest(criterion=evaluator.name):
+ self.assertIn(evaluator.name, set(SealCriterionName))
+
+
+class TestOfficial(unittest.TestCase):
+ def test_official_feed_passes(self):
+ self.assertTrue(OfficialEvaluator().evaluate(_ctx(official=True)).observed_pass)
+
+ def test_non_official_feed_fails(self):
+ self.assertFalse(
+ OfficialEvaluator().evaluate(_ctx(official=False)).observed_pass
+ )
+
+ def test_unknown_official_flag_fails(self):
+ """NULL is not an endorsement: only an explicit True passes."""
+ self.assertFalse(
+ OfficialEvaluator().evaluate(_ctx(official=None)).observed_pass
+ )
+
+ def test_has_no_grace_or_probation(self):
+ """A point-in-time check: it clears as soon as the feed is official again."""
+ self.assertIsNone(OfficialEvaluator.grace_period)
+ self.assertIsNone(OfficialEvaluator.probation_period)
+
+ def test_reason_names_the_offending_value(self):
+ result = OfficialEvaluator().evaluate(_ctx(official=None))
+ self.assertIn("None", result.reason)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_state_machine.py b/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_state_machine.py
new file mode 100644
index 000000000..1f73350ac
--- /dev/null
+++ b/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_state_machine.py
@@ -0,0 +1,288 @@
+#
+# MobilityData 2026
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""Unit tests for the generic seal criterion state machine. No database."""
+
+import unittest
+from datetime import datetime, timedelta, timezone
+
+from tasks.seal_of_reliability.criteria import PROBATION_PERIOD, SealCriterionName
+from tasks.seal_of_reliability.evaluators.base import CriterionObservation
+from tasks.seal_of_reliability.state_machine import SealCriterionState, transition
+
+DAY_ZERO = datetime(2026, 1, 1, tzinfo=timezone.utc)
+FEED_ID = "feed-1"
+
+# Official, the first criterion implemented, has neither a grace period nor probation, so it
+# exercises almost none of the state machine. These tests drive a synthetic criterion that
+# has both.
+GRACE = timedelta(days=14)
+
+
+def _day(offset: int) -> datetime:
+ return DAY_ZERO + timedelta(days=offset)
+
+
+def _observation(observed_pass, criterion=SealCriterionName.AVAILABLE):
+ return CriterionObservation(
+ criterion=criterion, observed_pass=observed_pass, reason="test"
+ )
+
+
+def _run(days, grace_period=GRACE, probation_period=PROBATION_PERIOD, state=None):
+ """Apply one verdict per entry in `days`: (day offset, observed_pass)."""
+ for offset, observed_pass in days:
+ state = transition(
+ prev=state,
+ observation=_observation(observed_pass),
+ grace_period=grace_period,
+ probation_period=probation_period,
+ now=_day(offset),
+ feed_id=FEED_ID,
+ )
+ return state
+
+
+def _failing(start, end):
+ """Observed failures on every day in [start, end)."""
+ return [(day, False) for day in range(start, end)]
+
+
+class TestNotEvaluable(unittest.TestCase):
+ def test_none_verdict_leaves_state_untouched(self):
+ """A criterion that cannot be evaluated must not change anything."""
+ state = _run([(0, False)])
+ unchanged = transition(
+ prev=state,
+ observation=_observation(None),
+ grace_period=GRACE,
+ probation_period=PROBATION_PERIOD,
+ now=_day(1),
+ )
+ self.assertIs(unchanged, state)
+
+ def test_none_verdict_with_no_previous_state_stays_none(self):
+ """No row is created, so `observed_pass IS NULL` keeps meaning never evaluated."""
+ self.assertIsNone(
+ transition(
+ prev=None,
+ observation=_observation(None),
+ grace_period=GRACE,
+ probation_period=PROBATION_PERIOD,
+ now=DAY_ZERO,
+ feed_id=FEED_ID,
+ )
+ )
+
+ def test_missing_feed_id_without_previous_state_raises(self):
+ with self.assertRaises(ValueError):
+ transition(
+ prev=None,
+ observation=_observation(False),
+ grace_period=GRACE,
+ probation_period=PROBATION_PERIOD,
+ now=DAY_ZERO,
+ )
+
+
+class TestGracePeriod(unittest.TestCase):
+ def test_first_evaluation_passes_immediately(self):
+ """No observation period to serve: a passing first verdict is a confirmed pass."""
+ state = _run([(0, True)])
+ self.assertTrue(state.observed_pass)
+ self.assertTrue(state.confirmed_pass)
+ self.assertIsNone(state.first_observed_failure_at)
+ self.assertIsNone(state.probation_start)
+
+ def test_first_evaluation_gets_no_grace_period(self):
+ """A criterion never seen to pass has no earned pass for the grace period to hold."""
+ state = _run([(0, False)])
+ self.assertFalse(state.observed_pass)
+ self.assertFalse(state.confirmed_pass)
+ self.assertEqual(state.last_confirmed_failure_at, DAY_ZERO)
+
+ def test_failure_within_grace_is_not_confirmed(self):
+ state = _run([(0, True)] + _failing(1, 14))
+ self.assertFalse(state.observed_pass)
+ self.assertTrue(state.confirmed_pass, "still inside the grace period")
+ self.assertIsNone(state.last_confirmed_failure_at)
+
+ def test_failure_at_grace_expiry_is_confirmed(self):
+ state = _run([(0, True)] + _failing(1, 16))
+ self.assertFalse(state.confirmed_pass)
+ self.assertEqual(state.last_confirmed_failure_at, _day(15))
+
+ def test_recovery_is_immediate(self):
+ """The first day with an observed pass clears the status, with no wait."""
+ state = _run([(0, True)] + _failing(1, 20) + [(20, True)])
+ self.assertTrue(state.confirmed_pass)
+ self.assertIsNone(state.first_observed_failure_at, "the streak is cleared")
+ self.assertEqual(state.last_observed_failure_at, _day(19), "history is kept")
+ self.assertEqual(state.last_confirmed_failure_at, _day(19))
+
+ def test_streak_restarts_after_recovery(self):
+ """A new streak gets a full grace period, it does not resume the old one."""
+ days = [(0, True)] + _failing(1, 10) + [(10, True)] + _failing(11, 20)
+ state = _run(days)
+ self.assertFalse(state.observed_pass)
+ self.assertTrue(state.confirmed_pass)
+ self.assertEqual(state.first_observed_failure_at, _day(11))
+
+ def test_no_grace_period_confirms_immediately(self):
+ state = _run([(0, True), (1, False)], grace_period=None)
+ self.assertFalse(state.confirmed_pass)
+ self.assertEqual(state.last_confirmed_failure_at, _day(1))
+
+
+class TestProbation(unittest.TestCase):
+ def test_a_clean_first_evaluation_opens_no_probation(self):
+ """Probation is opened only by a recovery, and this is not one."""
+ self.assertIsNone(_run([(0, True)]).probation_start)
+ self.assertIsNone(_run([(0, True), (1, True), (2, True)]).probation_start)
+
+ def test_a_blip_inside_grace_does_not_open_probation(self):
+ """Off probation, a failure absorbed by the grace period costs nothing."""
+ state = _run([(0, True), (1, False), (2, True)])
+ self.assertTrue(state.confirmed_pass)
+ self.assertIsNone(state.probation_start)
+
+ def test_recovery_from_a_confirmed_failure_opens_probation(self):
+ state = _run([(0, True)] + _failing(1, 16) + [(16, True)])
+ self.assertTrue(state.confirmed_pass, "the status clears the same day")
+ self.assertEqual(
+ state.probation_start, _day(16), "but probation runs from the repair day"
+ )
+
+ def test_probation_starts_the_day_after_the_last_failure(self):
+ """Stamped at tomorrow every failing day, so it lands on the day of repair."""
+ state = _run([(0, True)] + _failing(1, 16))
+ self.assertEqual(state.probation_start, _day(16))
+
+ def test_probation_ends_after_the_full_period(self):
+ days = [(0, True)] + _failing(1, 16) + [(16, True)]
+ served = _run(days + [(16 + PROBATION_PERIOD.days, True)])
+ self.assertIsNone(served.probation_start)
+
+ def test_probation_still_open_one_day_short(self):
+ days = [(0, True)] + _failing(1, 16) + [(16, True)]
+ state = _run(days + [(15 + PROBATION_PERIOD.days, True)])
+ self.assertEqual(state.probation_start, _day(16))
+
+ def test_a_failure_the_day_before_probation_ends_restarts_the_whole_period(self):
+ """Nearly served is not served: the count goes back to zero, not to one day left."""
+ # Probation runs from D16, so it would have ended on D196.
+ served = [(0, True)] + _failing(1, 16) + [(16, True)]
+
+ state = _run(served + [(195, False), (196, True)])
+ self.assertEqual(
+ state.probation_start, _day(196), "restarted on the repair day"
+ )
+
+ self.assertIsNotNone(
+ _run(served + [(195, False), (196, True), (375, True)]).probation_start,
+ "the original D196 end date is gone",
+ )
+ self.assertIsNone(
+ _run(served + [(195, False), (196, True), (376, True)]).probation_start,
+ "it ends 180 days after the restart instead",
+ )
+
+ def test_a_failure_on_the_last_day_of_probation_restarts_it(self):
+ """Expiry is only ever checked on a passing day, so a failure that day wins."""
+ served = [(0, True)] + _failing(1, 16) + [(16, True)]
+ state = _run(served + [(196, False)])
+ self.assertEqual(state.probation_start, _day(197))
+
+ def test_two_runs_on_the_same_failing_day_stamp_the_same_start(self):
+ """The `+1 day` is stamped from the UTC day, so the hour of the run cannot shift it."""
+ state = None
+ for moment in (_day(1) + timedelta(hours=2), _day(1) + timedelta(hours=23)):
+ state = transition(
+ prev=state,
+ observation=_observation(False),
+ grace_period=None,
+ probation_period=PROBATION_PERIOD,
+ now=moment,
+ feed_id=FEED_ID,
+ )
+ self.assertEqual(state.probation_start, _day(2))
+
+ def test_an_observed_failure_during_probation_restarts_it(self):
+ """Confirmed or not: during probation any failure costs the whole count."""
+ days = [(0, True)] + _failing(1, 16) + [(16, True), (100, False), (101, True)]
+ state = _run(days)
+ self.assertTrue(state.confirmed_pass, "the blip stayed inside the grace period")
+ self.assertEqual(state.probation_start, _day(101))
+
+ def test_a_criterion_without_probation_never_gets_one(self):
+ """Official's shape: it clears outright as soon as the feed recovers."""
+ state = _run(
+ [(0, True)] + _failing(1, 30) + [(30, True)],
+ grace_period=None,
+ probation_period=None,
+ )
+ self.assertTrue(state.confirmed_pass)
+ self.assertIsNone(state.probation_start)
+
+ def test_probation_uses_day_granularity(self):
+ """A run at any hour stamps the same start, so the period is not shifted."""
+ midday = _day(1) + timedelta(hours=13, minutes=27)
+ state = transition(
+ prev=None,
+ observation=_observation(False),
+ grace_period=None,
+ probation_period=PROBATION_PERIOD,
+ now=midday,
+ feed_id=FEED_ID,
+ )
+ self.assertEqual(state.probation_start, _day(2))
+
+
+class TestIdempotency(unittest.TestCase):
+ def test_same_timestamp_twice_produces_the_same_state(self):
+ """A re-run for the same instant must not compound a failure streak."""
+ once = _run([(0, True), (1, False), (2, False)])
+ twice = _run([(2, False)], state=once)
+ self.assertEqual(
+ once.first_observed_failure_at, twice.first_observed_failure_at
+ )
+ self.assertEqual(once.last_observed_failure_at, twice.last_observed_failure_at)
+ self.assertEqual(once.confirmed_pass, twice.confirmed_pass)
+ self.assertEqual(once.probation_start, twice.probation_start)
+
+ def test_state_is_not_mutated_in_place(self):
+ state = _run([(0, False)])
+ _run([(1, False)], state=state)
+ self.assertEqual(state.last_observed_failure_at, DAY_ZERO)
+
+
+class TestStateShape(unittest.TestCase):
+ def test_state_carries_feed_id_and_criterion(self):
+ state = _run([(0, False)])
+ self.assertEqual(state.feed_id, FEED_ID)
+ self.assertEqual(state.criterion, SealCriterionName.AVAILABLE)
+
+ def test_state_defaults_are_all_none(self):
+ state = SealCriterionState(
+ feed_id=FEED_ID, criterion=SealCriterionName.OFFICIAL
+ )
+ self.assertIsNone(state.observed_pass)
+ self.assertIsNone(state.confirmed_pass)
+ self.assertIsNone(state.evaluated_at)
+ self.assertIsNone(state.probation_start)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_updater_db.py b/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_updater_db.py
new file mode 100644
index 000000000..ca3b7d8a8
--- /dev/null
+++ b/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_updater_db.py
@@ -0,0 +1,689 @@
+#
+# MobilityData 2026
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""Integration tests for the seal context loader and orchestrator, against the test DB."""
+
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+from tasks.seal_of_reliability.context import build_contexts, get_seal_feeds_query
+from tasks.seal_of_reliability.criteria import PROBATION_PERIOD, SealCriterionName
+from tasks.seal_of_reliability.evaluators import CriterionEvaluator, OfficialEvaluator
+from tasks.seal_of_reliability.seal_updater import update_seals
+from sqlalchemy import delete, select
+
+from shared.database.database import with_db_session
+from shared.database_gen.sqlacodegen_models import (
+ Feed,
+ Feedreliabilityseal,
+ Gtfsfeed,
+ Sealcriterion,
+)
+from test_shared.test_utils.database_utils import default_db_url
+
+NOW = datetime(2026, 6, 1, 12, 0, tzinfo=timezone.utc)
+PREFIX = "seal_test_"
+
+OFFICIAL = f"{PREFIX}official"
+NOT_OFFICIAL = f"{PREFIX}not_official"
+UNKNOWN_OFFICIAL = f"{PREFIX}unknown_official"
+DEPRECATED = f"{PREFIX}deprecated"
+UNPUBLISHED = f"{PREFIX}unpublished"
+INACTIVE = f"{PREFIX}inactive"
+
+# The eligible feeds this module seeds. Runs that assert exact counts must be scoped to
+# these: an unnamed run also covers the fixtures seeded by conftest.pytest_sessionstart and
+# any seal rows another test left behind, so the totals would not be deterministic.
+OURS = [OFFICIAL, NOT_OFFICIAL, UNKNOWN_OFFICIAL, INACTIVE]
+
+
+class _StandInEvaluator(CriterionEvaluator):
+ """A criterion with a grace period and probation, for the probation tests below.
+
+ Official is the only real evaluator and has neither, so without a stand-in nothing
+ exercises `probation_start` against a database: it would never be persisted, reloaded,
+ or seen by the `has_seal` roll-up with a value other than NULL.
+
+ It reads `official` so a test can drive it with the existing `set_official` helper, but
+ unlike Official it debounces failures and serves probation afterwards. It borrows the
+ `available` enum value, which has no evaluator of its own yet (#1784).
+ """
+
+ name = SealCriterionName.AVAILABLE
+ grace_period = timedelta(days=14)
+
+ def _evaluate(self, ctx):
+ return ctx.official is True, f"stand-in, feed.official is {ctx.official!r}"
+
+
+# Patched over the registry so the roll-up sees a criterion that can be on probation.
+WITH_PROBATION = [OfficialEvaluator(), _StandInEvaluator()]
+
+
+DARK_FROM = NOW + timedelta(days=2)
+
+
+class _GoesDarkEvaluator(CriterionEvaluator):
+ """A criterion that loses its upstream input partway through, standing in for #1784.
+
+ It returns no verdict from `DARK_FROM` onwards, keyed on the clock rather than on
+ `official` so that a test can drive it and Official in opposite directions at the same
+ moment. No grace period, so its verdicts land immediately and the tests are about what
+ happens once it goes quiet.
+ """
+
+ name = SealCriterionName.COMPLIANT
+ grace_period = None
+
+ def _evaluate(self, ctx):
+ if ctx.now >= DARK_FROM:
+ return None, "stand-in has no input this run"
+ return ctx.official is True, f"stand-in, feed.official is {ctx.official!r}"
+
+
+# Official is kept in the registry on purpose: it is passing by the time the stand-in goes
+# dark, so if the stand-in's frozen row were dropped from the roll-up the seal would come
+# straight back. Without it, the empty-in_service guard would mask that.
+GOES_DARK = [OfficialEvaluator(), _GoesDarkEvaluator()]
+
+
+def _seed_feed(
+ db_session,
+ feed_id: str,
+ official=True,
+ status="active",
+ operational_status="published",
+):
+ """Insert one GTFS feed."""
+ db_session.add(
+ Gtfsfeed(
+ id=feed_id,
+ stable_id=feed_id,
+ data_type="gtfs",
+ status=status,
+ operational_status=operational_status,
+ official=official,
+ created_at=NOW - timedelta(days=400),
+ producer_url=f"https://example.com/{feed_id}.zip",
+ )
+ )
+ db_session.flush()
+
+
+def _set_official(db_session, feed_id: str, official):
+ db_session.execute(
+ Feed.__table__.update()
+ .where(Feed.__table__.c.id == feed_id)
+ .values(official=official)
+ )
+ db_session.commit()
+
+
+def _cleanup(db_session):
+ """Remove every row this module created.
+
+ Deleting from `feed` rather than `gtfsfeed` is deliberate: Gtfsfeed is a joined-table
+ subclass of Feed, so deleting the subclass leaves the parent row behind and the next
+ insert collides on feed_pkey. Deleting the parent cascades to gtfsfeed and to both seal
+ tables, all of which are ON DELETE CASCADE.
+ """
+ db_session.execute(delete(Feed).where(Feed.stable_id.like(f"{PREFIX}%")))
+ db_session.commit()
+
+
+class SealDbTestCase(unittest.TestCase):
+ """Seeds the feeds below before each test and removes them afterwards."""
+
+ @with_db_session(db_url=default_db_url)
+ def setUp(self, db_session):
+ _cleanup(db_session)
+ _seed_feed(db_session, OFFICIAL)
+ _seed_feed(db_session, NOT_OFFICIAL, official=False)
+ _seed_feed(db_session, UNKNOWN_OFFICIAL, official=None)
+ _seed_feed(db_session, DEPRECATED, status="deprecated")
+ _seed_feed(db_session, UNPUBLISHED, operational_status="unpublished")
+ _seed_feed(db_session, INACTIVE, status="inactive")
+ db_session.commit()
+
+ @with_db_session(db_url=default_db_url)
+ def tearDown(self, db_session):
+ _cleanup(db_session)
+
+ @staticmethod
+ @with_db_session(db_url=default_db_url)
+ def criterion_rows(feed_id, db_session):
+ table = Sealcriterion.__table__
+ return {
+ row.criterion: row
+ for row in db_session.execute(
+ select(table).where(table.c.feed_id == feed_id)
+ ).all()
+ }
+
+ @staticmethod
+ @with_db_session(db_url=default_db_url)
+ def seal_row(feed_id, db_session):
+ table = Feedreliabilityseal.__table__
+ return db_session.execute(
+ select(table).where(table.c.feed_id == feed_id)
+ ).first()
+
+ @staticmethod
+ @with_db_session(db_url=default_db_url)
+ def set_official(feed_id, official, db_session):
+ _set_official(db_session, feed_id, official)
+
+
+class TestEligibilityQuery(SealDbTestCase):
+ @with_db_session(db_url=default_db_url)
+ def test_excludes_deprecated_and_unpublished_but_keeps_inactive(self, db_session):
+ found = {
+ feed.stable_id
+ for feed in get_seal_feeds_query(db_session).all()
+ if feed.stable_id and feed.stable_id.startswith(PREFIX)
+ }
+ self.assertIn(OFFICIAL, found)
+ self.assertIn(INACTIVE, found, "inactive feeds must be evaluated, not frozen")
+ self.assertNotIn(DEPRECATED, found)
+ self.assertNotIn(UNPUBLISHED, found)
+
+ @with_db_session(db_url=default_db_url)
+ def test_stable_feed_ids_narrows_the_same_query(self, db_session):
+ feeds = get_seal_feeds_query(db_session, stable_feed_ids=[OFFICIAL]).all()
+ self.assertEqual([feed.stable_id for feed in feeds], [OFFICIAL])
+
+
+class TestBuildContexts(SealDbTestCase):
+ @with_db_session(db_url=default_db_url)
+ def test_loads_the_fields_the_evaluators_need(self, db_session):
+ feeds = get_seal_feeds_query(db_session, stable_feed_ids=[OFFICIAL]).all()
+ ctx = build_contexts(db_session, feeds, NOW)[feeds[0].id]
+ self.assertEqual(ctx.stable_id, OFFICIAL)
+ self.assertTrue(ctx.official)
+ self.assertEqual(ctx.now, NOW)
+
+ @with_db_session(db_url=default_db_url)
+ def test_builds_one_context_per_feed(self, db_session):
+ feeds = get_seal_feeds_query(
+ db_session, stable_feed_ids=[OFFICIAL, NOT_OFFICIAL]
+ ).all()
+ contexts = build_contexts(db_session, feeds, NOW)
+ self.assertEqual(len(contexts), 2)
+ self.assertEqual({ctx.official for ctx in contexts.values()}, {True, False})
+
+
+class TestUpdateSeals(SealDbTestCase):
+ def test_dry_run_writes_nothing(self):
+ report = update_seals(dry_run=True, stable_feed_ids=[OFFICIAL], now=NOW)
+ self.assertTrue(report["dry_run"])
+ self.assertEqual(report["total_feeds"], 1)
+ self.assertEqual(report["criterion_rows_written"], 0)
+ self.assertEqual(self.criterion_rows(OFFICIAL), {})
+ self.assertIsNone(self.seal_row(OFFICIAL))
+
+ def test_dry_run_counts_are_prospective(self):
+ """Nothing is held yet, so `after` describes what a real run would store."""
+ report = update_seals(dry_run=True, stable_feed_ids=OURS, now=NOW)
+ self.assertEqual(report["seals_before_run"], 0)
+ self.assertEqual(report["seals_after_run"], 2, "the two official feeds")
+ self.assertEqual(report["seals_granted"], 2)
+ self.assertEqual(report["seals_revoked"], 0)
+ self.assertIsNone(self.seal_row(OFFICIAL), "still a dry run")
+
+ def test_both_seal_transitions_are_reported_by_feed(self):
+ """Counts say how many moved; these say which, for each direction."""
+ first = update_seals(dry_run=False, stable_feed_ids=OURS, now=NOW)
+ self.assertEqual(
+ sorted(first["granted_stable_ids"]), sorted([OFFICIAL, INACTIVE])
+ )
+ self.assertEqual(first["revoked_stable_ids"], [])
+
+ self.set_official(OFFICIAL, False)
+ later = NOW + timedelta(days=1)
+ second = update_seals(dry_run=False, stable_feed_ids=OURS, now=later)
+ self.assertEqual(second["granted_stable_ids"], [])
+ self.assertEqual(second["revoked_stable_ids"], [OFFICIAL])
+
+ def test_seal_counts_balance(self):
+ """before + granted - revoked == after, across a grant then a revocation."""
+ first = update_seals(dry_run=False, stable_feed_ids=OURS, now=NOW)
+ self.assertEqual(first["seals_before_run"], 0)
+ self.assertEqual(first["seals_after_run"], 2)
+
+ self.set_official(OFFICIAL, False)
+ second = update_seals(dry_run=False, stable_feed_ids=OURS, now=NOW)
+ self.assertEqual(second["seals_before_run"], 2, "two were held going in")
+ self.assertEqual(second["seals_revoked"], 1)
+ self.assertEqual(second["seals_granted"], 0)
+ self.assertEqual(second["seals_after_run"], 1)
+ self.assertEqual(
+ second["seals_before_run"]
+ + second["seals_granted"]
+ - second["seals_revoked"],
+ second["seals_after_run"],
+ )
+
+ def test_a_reported_feed_carries_its_seal_state_and_its_criteria(self):
+ report = update_seals(dry_run=True, stable_feed_ids=[NOT_OFFICIAL], now=NOW)
+ self.assertEqual(len(report["feeds"]), 1)
+
+ feed = report["feeds"][0]
+ self.assertEqual(feed["stable_id"], NOT_OFFICIAL)
+ self.assertFalse(feed["had_seal"])
+ self.assertFalse(feed["has_seal"])
+ self.assertEqual(
+ [row["criterion"] for row in feed["criteria"]],
+ [SealCriterionName.OFFICIAL.value],
+ )
+ self.assertFalse(feed["criteria"][0]["observed_pass"])
+ self.assertTrue(feed["criteria"][0]["reason"])
+ self.assertIsNone(feed["criteria"][0]["previously_confirmed_pass"])
+
+ def test_named_feed_is_reported_even_when_nothing_moved(self):
+ """An explicit feed list is the debugging path: report it either way."""
+ update_seals(dry_run=False, stable_feed_ids=[OFFICIAL], now=NOW)
+ report = update_seals(dry_run=False, stable_feed_ids=[OFFICIAL], now=NOW)
+
+ feed = report["feeds"][0]
+ self.assertTrue(feed["had_seal"])
+ self.assertTrue(feed["has_seal"])
+ self.assertTrue(feed["criteria"][0]["confirmed_pass"])
+ self.assertTrue(feed["criteria"][0]["previously_confirmed_pass"])
+
+ def test_unnamed_run_reports_only_feeds_that_moved(self):
+ first = update_seals(dry_run=False, now=NOW)
+ reported = {row["stable_id"] for row in first["feeds"]}
+ self.assertIn(NOT_OFFICIAL, reported, "its criterion landed on a failure")
+ self.assertIn(OFFICIAL, reported, "it was granted the seal")
+ self.assertGreaterEqual(first["first_evaluations"], 2)
+
+ steady = update_seals(dry_run=False, now=NOW)
+ self.assertEqual(steady["feeds"], [], "nothing moved, so nothing to report")
+ self.assertEqual(steady["first_evaluations"], 0)
+
+ def test_a_feed_that_flips_is_reported_with_the_previous_verdict(self):
+ update_seals(dry_run=False, now=NOW)
+ self.set_official(OFFICIAL, False)
+ report = update_seals(dry_run=False, now=NOW)
+
+ moved = [row for row in report["feeds"] if row["stable_id"] == OFFICIAL]
+ self.assertEqual(len(moved), 1)
+ self.assertTrue(moved[0]["had_seal"])
+ self.assertFalse(moved[0]["has_seal"])
+ self.assertFalse(moved[0]["criteria"][0]["confirmed_pass"])
+ self.assertTrue(moved[0]["criteria"][0]["previously_confirmed_pass"])
+
+ def test_official_feed_earns_the_seal(self):
+ update_seals(dry_run=False, stable_feed_ids=[OFFICIAL], now=NOW)
+ row = self.criterion_rows(OFFICIAL)[SealCriterionName.OFFICIAL.value]
+ self.assertTrue(row.observed_pass)
+ self.assertTrue(row.confirmed_pass)
+ self.assertEqual(row.evaluated_at, NOW)
+ self.assertIsNone(row.first_observed_failure_at)
+ self.assertIsNone(row.probation_start, "a clean first run opens no probation")
+
+ seal = self.seal_row(OFFICIAL)
+ self.assertTrue(seal.has_seal)
+ self.assertEqual(seal.seal_earned_at, NOW)
+ self.assertIsNone(seal.seal_lost_at)
+
+ def test_failing_official_denies_the_seal_immediately(self):
+ """Official has no grace period, so one failure is confirmed at once."""
+ update_seals(dry_run=False, stable_feed_ids=[NOT_OFFICIAL], now=NOW)
+ row = self.criterion_rows(NOT_OFFICIAL)[SealCriterionName.OFFICIAL.value]
+ self.assertFalse(row.observed_pass)
+ self.assertFalse(row.confirmed_pass)
+ self.assertEqual(row.first_observed_failure_at, NOW)
+ self.assertEqual(row.last_confirmed_failure_at, NOW)
+ self.assertFalse(self.seal_row(NOT_OFFICIAL).has_seal)
+
+ def test_a_feed_that_never_qualifies_still_gets_a_seal_row(self):
+ """`created_at` on that row is what a tracking-age criterion measures against."""
+ update_seals(dry_run=False, stable_feed_ids=[NOT_OFFICIAL], now=NOW)
+ seal = self.seal_row(NOT_OFFICIAL)
+ self.assertIsNotNone(seal, "a row is written whether or not the feed qualifies")
+ self.assertFalse(seal.has_seal)
+ self.assertIsNone(seal.seal_lost_at, "nothing was held, so nothing was lost")
+
+ def test_unknown_official_flag_denies_the_seal(self):
+ update_seals(dry_run=False, stable_feed_ids=[UNKNOWN_OFFICIAL], now=NOW)
+ self.assertFalse(self.seal_row(UNKNOWN_OFFICIAL).has_seal)
+
+ def test_rerun_is_idempotent(self):
+ update_seals(dry_run=False, stable_feed_ids=[NOT_OFFICIAL], now=NOW)
+ first = self.criterion_rows(NOT_OFFICIAL)[SealCriterionName.OFFICIAL.value]
+ update_seals(dry_run=False, stable_feed_ids=[NOT_OFFICIAL], now=NOW)
+ second = self.criterion_rows(NOT_OFFICIAL)[SealCriterionName.OFFICIAL.value]
+ self.assertEqual(
+ first.first_observed_failure_at, second.first_observed_failure_at
+ )
+ self.assertEqual(
+ first.last_confirmed_failure_at, second.last_confirmed_failure_at
+ )
+
+ def test_losing_official_status_revokes_the_seal(self):
+ update_seals(dry_run=False, stable_feed_ids=[OFFICIAL], now=NOW)
+ self.assertTrue(self.seal_row(OFFICIAL).has_seal)
+
+ self.set_official(OFFICIAL, False)
+ later = NOW + timedelta(days=1)
+ update_seals(dry_run=False, stable_feed_ids=[OFFICIAL], now=later)
+ seal = self.seal_row(OFFICIAL)
+ self.assertFalse(seal.has_seal)
+ self.assertEqual(seal.seal_lost_at, later)
+ self.assertEqual(seal.seal_earned_at, NOW, "the earlier grant is preserved")
+
+ def test_regaining_official_status_clears_the_criterion(self):
+ """Official has no probation, so recovery restores the seal the same day."""
+ update_seals(dry_run=False, stable_feed_ids=[NOT_OFFICIAL], now=NOW)
+ self.assertFalse(self.seal_row(NOT_OFFICIAL).has_seal)
+
+ self.set_official(NOT_OFFICIAL, True)
+ later = NOW + timedelta(days=1)
+ update_seals(dry_run=False, stable_feed_ids=[NOT_OFFICIAL], now=later)
+ row = self.criterion_rows(NOT_OFFICIAL)[SealCriterionName.OFFICIAL.value]
+ self.assertTrue(row.confirmed_pass)
+ self.assertIsNone(row.first_observed_failure_at, "the streak is cleared")
+ self.assertIsNone(row.probation_start, "Official serves no probation")
+ self.assertEqual(
+ row.last_confirmed_failure_at, NOW, "the failure is still on record"
+ )
+ seal = self.seal_row(NOT_OFFICIAL)
+ self.assertTrue(seal.has_seal)
+ self.assertEqual(seal.seal_earned_at, later)
+
+ def test_partial_criteria_run_skips_the_roll_up(self):
+ """Named explicitly, `official` is still the whole registry, so not partial."""
+ report = update_seals(
+ dry_run=False,
+ stable_feed_ids=[OFFICIAL],
+ criteria=[SealCriterionName.OFFICIAL.value],
+ now=NOW,
+ )
+ self.assertFalse(report["partial_run"])
+ self.assertTrue(self.seal_row(OFFICIAL).has_seal)
+
+ def test_unknown_criterion_raises(self):
+ with self.assertRaises(ValueError):
+ update_seals(criteria=["not_a_criterion"], now=NOW)
+
+ def test_criterion_without_an_evaluator_raises(self):
+ """`stable` is a valid DB enum value but has no evaluator yet (#1784)."""
+ with self.assertRaises(ValueError):
+ update_seals(criteria=[SealCriterionName.STABLE.value], now=NOW)
+
+ def test_a_run_with_no_usable_feed_raises(self):
+ """Nothing was evaluated, so a report saying so would be too quiet."""
+ with self.assertRaises(ValueError) as caught:
+ update_seals(stable_feed_ids=[f"{PREFIX}does_not_exist"], now=NOW)
+ self.assertIn("not found", str(caught.exception))
+
+ def test_an_ineligible_feed_says_so_rather_than_not_found(self):
+ """A filtered-out feed is in the database, so "not found" would send you hunting."""
+ with self.assertRaises(ValueError) as caught:
+ update_seals(stable_feed_ids=[DEPRECATED], now=NOW)
+ message = str(caught.exception)
+ self.assertIn("not eligible", message)
+ self.assertIn(DEPRECATED, message)
+ self.assertNotIn("not found", message)
+
+ def test_unusable_feeds_are_dropped_rather_than_costing_the_run(self):
+ """One stale id must not cost a run over the feeds that are fine."""
+ with self.assertLogs(level="WARNING") as logs:
+ report = update_seals(
+ dry_run=False,
+ stable_feed_ids=[OFFICIAL, DEPRECATED, f"{PREFIX}does_not_exist"],
+ now=NOW,
+ )
+ self.assertEqual(report["total_feeds"], 1, "the eligible feed still ran")
+ self.assertTrue(self.seal_row(OFFICIAL).has_seal)
+
+ warning = "\n".join(logs.output)
+ self.assertIn(DEPRECATED, warning)
+ self.assertIn(f"{PREFIX}does_not_exist", warning)
+
+ def test_limit_caps_the_feeds_evaluated(self):
+ report = update_seals(dry_run=True, limit=2, now=NOW)
+ self.assertLessEqual(report["total_feeds"], 2)
+
+ def test_the_feed_list_is_capped_without_capping_the_run(self):
+ """The list is a sample; the two seal tables are the record."""
+ report = update_seals(
+ dry_run=False, stable_feed_ids=OURS, now=NOW, max_reported_feeds=2
+ )
+ self.assertEqual(len(report["feeds"]), 2)
+ self.assertEqual(report["feeds_omitted"], 2, "4 feeds requested")
+ self.assertEqual(report["total_feeds"], 4, "every feed was still evaluated")
+ self.assertEqual(len(self.criterion_rows(OFFICIAL)), 1, "and still written")
+
+ def test_feeds_omitted_is_zero_when_nothing_was_dropped(self):
+ report = update_seals(dry_run=True, stable_feed_ids=[OFFICIAL], now=NOW)
+ self.assertEqual(len(report["feeds"]), 1)
+ self.assertEqual(report["feeds_omitted"], 0)
+
+
+@patch("tasks.seal_of_reliability.seal_updater.EVALUATORS", WITH_PROBATION)
+class TestProbation(SealDbTestCase):
+ """Probation persisted and rolled up, driven by `_StandInEvaluator`.
+
+ Timestamps are written out rather than derived so the assertions do not restate the
+ implementation they are checking.
+ """
+
+ STREAK_STARTS = datetime(2026, 6, 2, 12, 0, tzinfo=timezone.utc)
+ CONFIRMED_AT = datetime(2026, 6, 16, 12, 0, tzinfo=timezone.utc) # 14 days later
+ REPAIRED_AT = datetime(2026, 6, 17, 12, 0, tzinfo=timezone.utc)
+ PROBATION_FROM = datetime(
+ 2026, 6, 17, 0, 0, tzinfo=timezone.utc
+ ) # the day of repair
+
+ def stand_in(self):
+ return self.criterion_rows(OFFICIAL)[SealCriterionName.AVAILABLE.value]
+
+ def run_at(self, moment):
+ update_seals(dry_run=False, stable_feed_ids=[OFFICIAL], now=moment)
+
+ def test_a_clean_run_puts_nothing_on_probation(self):
+ self.run_at(NOW)
+ self.assertIsNone(self.stand_in().probation_start)
+ self.assertTrue(self.seal_row(OFFICIAL).has_seal)
+
+ def test_a_failure_inside_the_grace_period_starts_no_probation(self):
+ self.run_at(NOW)
+ self.set_official(OFFICIAL, False)
+ self.run_at(self.STREAK_STARTS)
+
+ row = self.stand_in()
+ self.assertFalse(row.observed_pass)
+ self.assertTrue(row.confirmed_pass, "still inside its grace period")
+ self.assertIsNone(row.probation_start)
+
+ def test_probation_starts_on_the_day_of_repair(self):
+ self.run_at(NOW)
+ self.set_official(OFFICIAL, False)
+ self.run_at(self.STREAK_STARTS)
+ self.run_at(self.CONFIRMED_AT)
+
+ row = self.stand_in()
+ self.assertFalse(row.confirmed_pass, "the streak outlasted the grace period")
+ self.assertEqual(row.probation_start, self.PROBATION_FROM)
+
+ def test_probation_withholds_the_seal_while_every_criterion_passes(self):
+ self.run_at(NOW)
+ self.set_official(OFFICIAL, False)
+ self.run_at(self.STREAK_STARTS)
+ self.run_at(self.CONFIRMED_AT)
+ self.set_official(OFFICIAL, True)
+ self.run_at(self.REPAIRED_AT)
+
+ rows = self.criterion_rows(OFFICIAL)
+ self.assertTrue(rows[SealCriterionName.OFFICIAL.value].confirmed_pass)
+ self.assertTrue(rows[SealCriterionName.AVAILABLE.value].confirmed_pass)
+ self.assertEqual(
+ rows[SealCriterionName.AVAILABLE.value].probation_start,
+ self.PROBATION_FROM,
+ "recovery clears the status but starts probation",
+ )
+ self.assertFalse(
+ self.seal_row(OFFICIAL).has_seal,
+ "both criteria pass, so only the open probation can be withholding it",
+ )
+
+ def test_the_seal_returns_once_probation_is_served(self):
+ self.run_at(NOW)
+ self.set_official(OFFICIAL, False)
+ self.run_at(self.STREAK_STARTS)
+ self.run_at(self.CONFIRMED_AT)
+ self.set_official(OFFICIAL, True)
+ self.run_at(self.REPAIRED_AT)
+
+ served_at = self.PROBATION_FROM + PROBATION_PERIOD
+ self.run_at(served_at)
+
+ self.assertIsNone(self.stand_in().probation_start)
+ seal = self.seal_row(OFFICIAL)
+ self.assertTrue(seal.has_seal)
+ self.assertEqual(seal.seal_earned_at, served_at)
+
+
+class TestCriterionBroughtIntoServiceLate(SealDbTestCase):
+ """A criterion whose evaluator arrives after the feed already has rows.
+
+ This is the `available` case from #1761: its source only starts collecting on some date,
+ so until then it has no row at all and the roll-up has to carry on without it. Each test
+ runs once with Official alone, then again with the stand-in registered.
+ """
+
+ LATER = NOW + timedelta(days=1)
+
+ def run_at(self, moment, evaluators):
+ with patch("tasks.seal_of_reliability.seal_updater.EVALUATORS", evaluators):
+ update_seals(dry_run=False, stable_feed_ids=[OFFICIAL], now=moment)
+
+ def test_a_criterion_with_no_row_does_not_deny_the_seal(self):
+ """The waiver itself: an unevaluated criterion is skipped, not counted as failing."""
+ self.run_at(NOW, [OfficialEvaluator()])
+
+ self.assertEqual(
+ set(self.criterion_rows(OFFICIAL)),
+ {SealCriterionName.OFFICIAL.value},
+ "the stand-in has no row yet",
+ )
+ self.assertTrue(self.seal_row(OFFICIAL).has_seal)
+
+ def test_a_passing_newcomer_joins_without_disturbing_the_seal(self):
+ self.run_at(NOW, [OfficialEvaluator()])
+ earned_at = self.seal_row(OFFICIAL).seal_earned_at
+
+ self.run_at(self.LATER, WITH_PROBATION)
+
+ row = self.criterion_rows(OFFICIAL)[SealCriterionName.AVAILABLE.value]
+ self.assertTrue(row.confirmed_pass)
+ self.assertIsNone(row.probation_start, "a first verdict is not a recovery")
+ seal = self.seal_row(OFFICIAL)
+ self.assertTrue(seal.has_seal)
+ self.assertEqual(
+ seal.seal_earned_at,
+ earned_at,
+ "the seal was never withdrawn and re-granted",
+ )
+ self.assertIsNone(seal.seal_lost_at)
+
+ def test_a_newcomer_whose_first_verdict_fails_gets_no_grace_period(self):
+ """`first_evaluation` denies the grace period: nothing has been earned to hold.
+
+ The stand-in does have a 14-day grace period, and it is deliberately not applied
+ here — otherwise this feed would keep its seal for a fortnight on a criterion we
+ have never once seen pass. So the failure is a confirmed failure on day one.
+ """
+ self.run_at(NOW, [OfficialEvaluator()])
+ self.assertTrue(self.seal_row(OFFICIAL).has_seal)
+
+ self.set_official(OFFICIAL, False)
+ self.run_at(self.LATER, WITH_PROBATION)
+
+ row = self.criterion_rows(OFFICIAL)[SealCriterionName.AVAILABLE.value]
+ self.assertFalse(row.observed_pass)
+ self.assertFalse(
+ row.confirmed_pass, "a confirmed failure on its very first verdict"
+ )
+ self.assertEqual(row.first_observed_failure_at, self.LATER)
+ self.assertEqual(row.last_confirmed_failure_at, self.LATER)
+
+ seal = self.seal_row(OFFICIAL)
+ self.assertFalse(seal.has_seal)
+ self.assertEqual(seal.seal_lost_at, self.LATER)
+
+
+@patch("tasks.seal_of_reliability.seal_updater.EVALUATORS", GOES_DARK)
+class TestCriterionThatStopsBeingEvaluable(SealDbTestCase):
+ """A criterion that produced a verdict once and then loses its input.
+
+ The roll-up skips criteria that have *never* produced a verdict, so the safety of that
+ rule rests entirely on a criterion never falling back into that state once it has one.
+ """
+
+ FAILED_AT = NOW + timedelta(days=1)
+
+ def stand_in(self):
+ return self.criterion_rows(OFFICIAL).get(SealCriterionName.COMPLIANT.value)
+
+ def run_at(self, moment):
+ update_seals(dry_run=False, stable_feed_ids=[OFFICIAL], now=moment)
+
+ def test_going_dark_freezes_the_verdict_and_keeps_the_seal_withheld(self):
+ # Both criteria pass, then both fail and the seal goes.
+ self.run_at(NOW)
+ self.assertTrue(self.seal_row(OFFICIAL).has_seal)
+
+ self.set_official(OFFICIAL, False)
+ self.run_at(self.FAILED_AT)
+ self.assertFalse(self.seal_row(OFFICIAL).has_seal)
+
+ # Official recovers, but the stand-in has lost its input. Its stored failure must
+ # stand and keep withholding the seal.
+ self.set_official(OFFICIAL, True)
+ self.run_at(DARK_FROM)
+
+ rows = self.criterion_rows(OFFICIAL)
+ self.assertTrue(
+ rows[SealCriterionName.OFFICIAL.value].confirmed_pass,
+ "the only other criterion is passing again",
+ )
+ frozen = rows[SealCriterionName.COMPLIANT.value]
+ self.assertFalse(frozen.observed_pass, "the last verdict stands")
+ self.assertFalse(frozen.confirmed_pass)
+ self.assertEqual(
+ frozen.evaluated_at, self.FAILED_AT, "not re-stamped without a verdict"
+ )
+ self.assertFalse(
+ self.seal_row(OFFICIAL).has_seal,
+ "going dark must not hand the seal back",
+ )
+
+ def test_going_dark_before_any_verdict_writes_no_row(self):
+ """The other half: with no verdict ever, there is nothing to hold in service."""
+ self.run_at(DARK_FROM)
+
+ self.assertIsNone(self.stand_in(), "no row is written without a verdict")
+ self.assertTrue(
+ self.seal_row(OFFICIAL).has_seal,
+ "and the criterion is skipped rather than denying the seal",
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/liquibase/changelog.xml b/liquibase/changelog.xml
index 1408c7220..f04ba7ae4 100644
--- a/liquibase/changelog.xml
+++ b/liquibase/changelog.xml
@@ -127,6 +127,8 @@
+
+
diff --git a/liquibase/changes/feat_1783.sql b/liquibase/changes/feat_1783.sql
new file mode 100644
index 000000000..3b51a0a05
--- /dev/null
+++ b/liquibase/changes/feat_1783.sql
@@ -0,0 +1,24 @@
+-- Issue #1783: Seal of Reliability algorithm (spec in #1761). Adjusts the tables added by
+-- feat_1760.sql. Nothing reads them yet, so this renames in place.
+
+-- The two status booleans become positive: TRUE now means the criterion passed.
+ALTER TABLE SealCriterion RENAME COLUMN raw_failing TO observed_pass;
+ALTER TABLE SealCriterion RENAME COLUMN grace_failing TO confirmed_pass;
+UPDATE SealCriterion SET observed_pass = NOT observed_pass WHERE observed_pass IS NOT NULL;
+UPDATE SealCriterion SET confirmed_pass = NOT confirmed_pass WHERE confirmed_pass IS NOT NULL;
+
+-- The timestamps stay negative: they record events that really are failures.
+ALTER TABLE SealCriterion RENAME COLUMN first_raw_failure_at TO first_observed_failure_at;
+ALTER TABLE SealCriterion RENAME COLUMN last_raw_failure_at TO last_observed_failure_at;
+ALTER TABLE SealCriterion RENAME COLUMN last_grace_failure_at TO last_confirmed_failure_at;
+
+ALTER TABLE SealCriterion ADD COLUMN IF NOT EXISTS probation_start TIMESTAMPTZ;
+
+COMMENT ON COLUMN SealCriterion.observed_pass IS
+ 'The criterion''s own check on this day, with no debouncing. NULL = never evaluated.';
+COMMENT ON COLUMN SealCriterion.confirmed_pass IS
+ 'The debounced status, FALSE once an observed failure outlasts the grace period. '
+ 'NULL = never evaluated.';
+COMMENT ON COLUMN SealCriterion.probation_start IS
+ 'Start of the 180 days with no observed failure a criterion serves after recovering '
+ 'from a confirmed failure. NULL = not on probation.';