From 1711f3b0d100d288f12ae545360041504a92d4a7 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Mon, 14 Sep 2026 11:46:19 -0300 Subject: [PATCH 1/6] SIENTIAPDE-2011: Compute query-time levels union / ignore_models_id intersection Push level and model-exclusion filtering down to the Mongo query instead of fetching the full notification_queue unfiltered on every run. - Fix await ordering in load_notification_package so sending_configs (active receiver_groups) resolves before load_latest_data is called, letting the query filter be built from group data. - Add _compute_query_time_filter: unions active groups' `levels` (falls back to the workflow's default filter if any group has no restriction, since that group wants every level) and intersects their `ignore_models_id` (a model is only safe to drop from the fetch if every active group ignores it). - Add unit tests covering mixed groups, a model ignored by only one group (must not be excluded from the fetch), and the empty active-groups case. Co-Authored-By: Claude Sonnet 5 --- .../subworkflows/load_notification_package.py | 48 ++++++++++++- .../test_load_notification_package.py | 70 ++++++++++++++++++- 2 files changed, 113 insertions(+), 5 deletions(-) diff --git a/orchestrator/workflows/subworkflows/load_notification_package.py b/orchestrator/workflows/subworkflows/load_notification_package.py index 8b603f2..0f28ec2 100644 --- a/orchestrator/workflows/subworkflows/load_notification_package.py +++ b/orchestrator/workflows/subworkflows/load_notification_package.py @@ -68,18 +68,23 @@ class LoadNotificationPackage: # Load notification package from collection "notification_queue", using a # static filter + sending_configs = await sending_configs_handler + + base_data_filter = self._compute_query_time_filter( + sending_configs, input_data['base_data_filter'] + ) + notification_package = await workflow.start_local_activity_method( Activities.load_latest_data, { **metadata, 'collection_name': 'notification_queue', 'last_data_timestamp': last_timestamp, - 'base_data_filter': input_data['base_data_filter'], + 'base_data_filter': base_data_filter, }, start_to_close_timeout=timedelta(seconds=60), retry_policy=retry_policy, ) - sending_configs = await sending_configs_handler if not sending_configs or not notification_package: return { @@ -106,3 +111,42 @@ class LoadNotificationPackage: 'notification_package': notification_package, 'sending_configs': sending_configs, } + + def _compute_query_time_filter( + self, sending_configs: list[dict[str, Any]], default_filter: dict[str, Any] + ) -> dict[str, Any]: + """ + Compute the Mongo query-time filter from all active receiver groups. + + - levels: union of every group's `levels`. If ANY active group has no + restriction (missing/empty `levels`), that group wants every level, so + the query can't restrict levels at all -> falls back to whatever level + restriction `default_filter` itself carries (e.g. ERROR-only for Alerts). + - ignore_models_id: intersection of every group's `ignore_models_id`. A model + is only safe to drop from the fetch if EVERY active group ignores it. + A group with no `ignore_models_id` is treated as "ignores nothing" (empty + set), which naturally zeroes out the intersection via set math. + """ + if not sending_configs: + return default_filter + + sending_configs_levels = [group.get('levels') for group in sending_configs] + + if all(sending_configs_levels): + levels_union: set[str] = set() + for levels in sending_configs_levels: + if levels: + levels_union.update(levels) + query_filter = {**default_filter, 'level': {'$in': sorted(levels_union)}} + else: + query_filter = dict(default_filter) + + sending_configs_ignore_models_id = [ + set(group.get('ignore_models_id', [])) for group in sending_configs + ] + models_intersection = set.intersection(*sending_configs_ignore_models_id) + + if models_intersection: + query_filter['model_id'] = {'$nin': sorted(models_intersection)} + + return query_filter diff --git a/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py b/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py index faeac7f..9f7d01b 100644 --- a/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py +++ b/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py @@ -32,16 +32,19 @@ async def test_run(workflow_mock, load_notification_package): 'base_data_filter': {'level': 'ERROR'}, } + # Await order is now: last_timestamp -> sending_configs -> notification_package + # (fixed ordering bug: sending_configs must resolve before load_latest_data is + # called, so its levels/ignore_models_id can feed the query-time filter). workflow_mock.start_local_activity_method.side_effect = [ '2023-01-01 12:00:00', [ { - 'id': '1', + 'id_r': '1', } ], [ { - 'id_r': '1', + 'id': '1', } ], ] @@ -158,7 +161,9 @@ async def test_run_has_data(workflow_mock, load_notification_package): 'mail_type': 'test_mail_type', } - workflow_mock.start_local_activity_method.side_effect = ['2023-01-01 12:00:00', ['data'], []] + # sending_configs (awaited 2nd) is empty here on purpose, to exercise the + # early-return branch below regardless of notification_package having data. + workflow_mock.start_local_activity_method.side_effect = ['2023-01-01 12:00:00', [], ['data']] output = await load_notification_package.run(input_data) @@ -169,3 +174,62 @@ async def test_run_has_data(workflow_mock, load_notification_package): } workflow_mock.start_activity_method.assert_not_called() + + +def test_compute_query_time_filter_empty_active_groups(load_notification_package): + default_filter = {'level': 'ERROR'} + + result = load_notification_package._compute_query_time_filter([], default_filter) + + assert result == default_filter + + +def test_compute_query_time_filter_mixed_groups_no_levels_restriction(load_notification_package): + # One active group restricts levels/ignore_models_id, the other has neither set + # (the common case today) -> that group wants everything, so no levels + # restriction can be applied at query time; falls back to default_filter. + sending_configs = [ + {'group_name': 'group_a', 'levels': ['ERROR'], 'ignore_models_id': ['model_a']}, + {'group_name': 'group_b'}, + ] + default_filter = {'level': 'ERROR'} + + result = load_notification_package._compute_query_time_filter(sending_configs, default_filter) + + assert result == {'level': 'ERROR'} + + +def test_compute_query_time_filter_model_ignored_by_only_one_group_not_excluded( + load_notification_package, +): + # model_a is ignored by group_a but group_b still wants it -> must stay in + # the fetch (the per-group filter is what drops it for group_a specifically). + sending_configs = [ + {'group_name': 'group_a', 'levels': ['ERROR'], 'ignore_models_id': ['model_a']}, + {'group_name': 'group_b', 'levels': ['ERROR'], 'ignore_models_id': []}, + ] + default_filter = {} + + result = load_notification_package._compute_query_time_filter(sending_configs, default_filter) + + assert result == {'level': {'$in': ['ERROR']}} + assert 'model_id' not in result + + +def test_compute_query_time_filter_union_levels_and_intersection_models( + load_notification_package, +): + # Every active group restricts levels -> union of levels applies. + # model_a is ignored by every active group -> safe to exclude at query time. + sending_configs = [ + {'group_name': 'group_a', 'levels': ['ERROR'], 'ignore_models_id': ['model_a', 'model_b']}, + {'group_name': 'group_b', 'levels': ['WARNING', 'ERROR'], 'ignore_models_id': ['model_a']}, + ] + default_filter = {} + + result = load_notification_package._compute_query_time_filter(sending_configs, default_filter) + + assert result == { + 'level': {'$in': ['ERROR', 'WARNING']}, + 'model_id': {'$nin': ['model_a']}, + } From 14b854148aa83c17929efcb6ca2174b65e3a2a97 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Mon, 14 Sep 2026 12:15:23 -0300 Subject: [PATCH 2/6] SIENTIAPDE-2011: add e2e coverage for query-time filter, fix stale e2e fixtures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit e2e fixtures (default_notification, seed_last_timestamp, default_receiver_group) still assumed the pre-SIENTIAPDE-2007 string-timestamp representation and the old ignore_models field name, so the load_notification_package e2e suite was failing independently of this ticket's changes: - default_notification now seeds `timestamp` as a native datetime (matches production since SIENTIAPDE-2007's Notification.timestamp = now()). - seed_last_timestamp now uses DATETIME_FORMAT_MS_WITH_TZ (matches what put_last_data_timestamp actually writes to Redis today). - default_receiver_group's ignore_models renamed to ignore_models_id, per this ticket's decision. - test_reports_main_workflow.py's explicit timestamp= call sites updated to match. Also fixes a real production bug in put_last_data_timestamp (slot_manager.py) surfaced by running the e2e suite against a real Temporal environment: Temporal's default data converter serializes datetime values to ISO-8601 strings across the activity boundary, and pymongo returns naive datetimes for BSON Date fields, so `data['timestamp'].max().strftime(...)` crashed on a plain str. Unit tests never caught this because they mock the activity boundary and never cross real (de)serialization. Fixed by normalizing through `pandas.to_datetime(..., utc=True)`, which handles native datetime/Timestamp and offset-aware/naive ISO strings uniformly. Adds test_load_package_query_time_filter_union_and_intersection, a new e2e test exercising the actual SIENTIAPDE-2011 behavior end-to-end: union of active groups' levels, intersection of their ignore_models_id, and the requirement that a model ignored by only one of several active groups must not be excluded from the fetch. Note: test_reports_main_workflow.py / test_alerts_main_workflow.py still have a separate, pre-existing failure in process_notifications.py (same family of bug: a notification's timestamp field parsed with a fixed strptime format after crossing the same Temporal serialization boundary) — confirmed present on the unmodified baseline too, unrelated to this ticket, tracked separately. Co-Authored-By: Claude Sonnet 5 --- e2e/helpers.py | 44 +++---- e2e/scenarios.md | 8 +- e2e/test_reports_main_workflow.py | 5 +- ...t_subworkflow_load_notification_package.py | 110 +++++++++++++++--- orchestrator/activities/slot_manager.py | 23 +++- 5 files changed, 143 insertions(+), 47 deletions(-) diff --git a/e2e/helpers.py b/e2e/helpers.py index 3af6f87..cafe1ce 100644 --- a/e2e/helpers.py +++ b/e2e/helpers.py @@ -29,7 +29,6 @@ MONGO_COLLECTIONS = ( ) DATETIME_FORMAT_MS_WITH_TZ = '%Y-%m-%d %H:%M:%S.%f%z' -DATETIME_FORMAT_WITH_TZ = '%Y-%m-%d %H:%M:%S%z' _TIMESTAMP_MARKER_PATTERN = re.compile(r'^@now(?:([+-])(\d+)([smhd]))?$') _TIMESTAMP_UNIT_TO_KWARG = {'s': 'seconds', 'm': 'minutes', 'h': 'hours', 'd': 'days'} @@ -222,17 +221,19 @@ def seed_last_timestamp(redis_client: Redis, mail_type: str, value: str) -> None Set notification_last_timestamp for a mail type, JSON-encoded. Values must be JSON-encoded so ``redis_repository.get`` (which calls - ``json.loads`` on the raw payload) can deserialize them. The value - must follow the exact format ``sientia_do.notifications.models.Notification`` - writes into ``notification_queue.timestamp``: ``DATETIME_FORMAT_WITH_TZ`` - (e.g. ``"2026-05-22 16:47:02+0000"``) — no microseconds and no colon in - the timezone offset. + ``json.loads`` on the raw payload) can deserialize them. The value must + follow the exact format ``put_last_data_timestamp`` + (``orchestrator/activities/slot_manager.py``) writes into Redis today: + ``DATETIME_FORMAT_MS_WITH_TZ`` (e.g. ``"2026-05-22 16:47:02.000000+0000"``) + — microseconds included, no colon in the timezone offset. ``load_latest_data`` + parses this same format back into a ``datetime`` for the ``$gt`` comparison, + so the two sides must agree exactly. Args: redis_client: Redis client connected to the test instance. mail_type: Mail type identifier (e.g. ``"Alerts"``, ``"Reports"``). - value: Timestamp string in ``DATETIME_FORMAT_WITH_TZ`` - (e.g. ``"2024-06-01 10:30:00+0000"``). + value: Timestamp string in ``DATETIME_FORMAT_MS_WITH_TZ`` + (e.g. ``"2024-06-01 10:30:00.000000+0000"``). """ redis_client.set(f'notification_last_timestamp:{mail_type}', json.dumps(value)) @@ -295,34 +296,35 @@ def default_notification( *, notification_id: str, level: str = 'ERROR', - timestamp: str | None = None, + timestamp: datetime | None = None, model_name: str = 'model-a', model_id: str = '1', ) -> dict[str, Any]: """ Build a minimal notification_queue document mirroring production layout. - The ``timestamp`` field is stored as a string in ``DATETIME_FORMAT_WITH_TZ`` - because that is exactly what ``sientia_do.notifications.models.Notification`` - writes into ``notification_queue`` in production (``now().strftime( - DATETIME_FORMAT_WITH_TZ)``). Tests intentionally use this same format so we + The ``timestamp`` field is stored as a native, timezone-aware ``datetime`` + (BSON Date) because that is exactly what + ``sientia_do.notifications.models.Notification`` writes into + ``notification_queue`` in production since SIENTIAPDE-2007 + (``self.timestamp = now()`` — a real ``datetime``, no more string + formatting). Tests intentionally use this same representation so we surface, rather than hide, real production behavior in downstream - activities. + activities (e.g. ``put_last_data_timestamp``'s ``DataFrame(...)['timestamp'] + .max().strftime(...)``, which requires a real ``datetime``, not a string). Args: notification_id: Unique identifier for the notification. level: Notification level (e.g. ``"ERROR"``, ``"WARNING"``). - timestamp: Optional production-format timestamp string. ``None`` falls - back to a fixed sample value. + timestamp: Optional timezone-aware datetime. ``None`` falls back to a + fixed sample value. model_name: Model name attached to the notification. model_id: Model id attached to the notification. Return: dict[str, Any]: A notification document ready for insertion. """ - ts = timestamp if timestamp is not None else datetime( - 2024, 6, 1, 12, 0, 0, tzinfo=UTC - ).strftime(DATETIME_FORMAT_WITH_TZ) + ts = timestamp if timestamp is not None else datetime(2024, 6, 1, 12, 0, 0, tzinfo=UTC) return { 'notification_id': notification_id, 'level': level, @@ -343,7 +345,7 @@ def default_receiver_group( members: list[str] | None = None, levels: list[str] | None = None, contents: list[str] | None = None, - ignore_models: list[str] | None = None, + ignore_models_id: list[str] | None = None, ) -> dict[str, Any]: """Minimal active receiver_groups document.""" return { @@ -352,5 +354,5 @@ def default_receiver_group( 'members': members or ['admin@example.com'], 'levels': levels or ['ERROR', 'WARNING', 'INFO'], 'contents': contents or ['core_alerts', 'persistent_alerts', 'reports'], - 'ignore_models': ignore_models or [], + 'ignore_models_id': ignore_models_id or [], } diff --git a/e2e/scenarios.md b/e2e/scenarios.md index 169abc2..1bf1eb5 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -71,7 +71,12 @@ Alert past `notification_ttl` with cache cleared → new email sent. ### A.2.1 Group filtering -Receiver group `levels` / `ignore_models` honored. +Receiver group `levels` / `ignore_models_id` honored. The query-time half +(union of `levels`, intersection of `ignore_models_id` across active groups) +is covered at the subworkflow level — see `LoadNotificationPackage` below. +The per-group half (each group only receiving its own precise slice) is not +yet implemented in production code (SIENTIAPDE-2012) and so is not yet +exercised here. ### A.3.1 Empty queue @@ -103,6 +108,7 @@ No notifications → no SMTP, no Postgres row. - No prior Redis timestamp → all matching notifications returned, max timestamp stored. - Prior timestamp → only newer notifications returned. +- Query-time filter (SIENTIAPDE-2011) → union of active groups' `levels`, intersection of their `ignore_models_id`; a model ignored by only one of several active groups is not excluded from the fetch. - Empty Mongo → no Redis timestamp write. ### ProcessNotifications — `e2e/test_subworkflow_process_notifications.py` diff --git a/e2e/test_reports_main_workflow.py b/e2e/test_reports_main_workflow.py index 64ea6f7..84d31d8 100644 --- a/e2e/test_reports_main_workflow.py +++ b/e2e/test_reports_main_workflow.py @@ -1,5 +1,6 @@ """E2E tests for the Reports main workflow.""" +from datetime import UTC, datetime from email.message import EmailMessage, Message import pytest @@ -79,13 +80,13 @@ async def test_scenario_r_1_1_happy_path( notification_id='r-warn', level='WARNING', model_name='m2', - timestamp='2024-06-01 12:01:00+0000', + timestamp=datetime(2024, 6, 1, 12, 1, 0, tzinfo=UTC), ), default_notification( notification_id='r-info', level='INFO', model_name='m3', - timestamp='2024-06-01 12:02:00+0000', + timestamp=datetime(2024, 6, 1, 12, 2, 0, tzinfo=UTC), ), ], ) diff --git a/e2e/test_subworkflow_load_notification_package.py b/e2e/test_subworkflow_load_notification_package.py index 2bb556f..9ba3f61 100644 --- a/e2e/test_subworkflow_load_notification_package.py +++ b/e2e/test_subworkflow_load_notification_package.py @@ -1,5 +1,7 @@ """E2E tests for LoadNotificationPackage subworkflow.""" +from datetime import UTC, datetime + import pytest from redis import Redis from temporalio.testing import WorkflowEnvironment @@ -42,10 +44,12 @@ async def test_load_package_without_prior_timestamp( """ No Redis timestamp → returns notifications and stores max timestamp. - Notifications are seeded with the exact production format produced by - ``sientia_do.notifications.models.Notification`` (string in - ``DATETIME_FORMAT_WITH_TZ``, e.g. ``"2024-06-01 11:00:00+0000"``). The - cached "last timestamp" must mirror that representation. + Notifications are seeded with the exact production representation written + by ``sientia_do.notifications.models.Notification`` since SIENTIAPDE-2007: + a native ``datetime`` (BSON Date), not a formatted string. The cached + "last timestamp" written back to Redis by ``put_last_data_timestamp`` is a + string formatted per ``DATETIME_FORMAT_MS_WITH_TZ`` (microseconds + included) — that is the format this test asserts against. """ seed_receiver_groups(mongo_uri, E2E_DATABASE, [default_receiver_group()]) seed_notifications( @@ -54,11 +58,11 @@ async def test_load_package_without_prior_timestamp( [ default_notification( notification_id='n1', - timestamp='2024-06-01 10:00:00+0000', + timestamp=datetime(2024, 6, 1, 10, 0, 0, tzinfo=UTC), ), default_notification( notification_id='n2', - timestamp='2024-06-01 11:00:00+0000', + timestamp=datetime(2024, 6, 1, 11, 0, 0, tzinfo=UTC), ), ], ) @@ -75,7 +79,7 @@ async def test_load_package_without_prior_timestamp( assert len(result['notification_package']) == 2 stored = redis_client.get('notification_last_timestamp:Alerts') or '' - assert stored.strip('"') == '2024-06-01 11:00:00+0000' + assert stored.strip('"') == '2024-06-01 11:00:00.000000+0000' @pytest.mark.e2e @@ -90,27 +94,27 @@ async def test_load_package_with_prior_timestamp( """ Existing timestamp → only newer notifications returned. - Both the seeded "last timestamp" (Redis) and the notification timestamps - (Mongo) follow the production format used by - ``sientia_do.notifications.models.Notification`` (string in - ``DATETIME_FORMAT_WITH_TZ``). ``load_latest_data`` will translate the - Redis value into a Python ``datetime`` and apply ``{$gt: }`` - against the Mongo string timestamps; this exercise surfaces the real - BSON comparison semantics rather than a synthetic ideal. + The seeded "last timestamp" (Redis) follows the format + ``put_last_data_timestamp`` actually writes (``DATETIME_FORMAT_MS_WITH_TZ``, + microseconds included); the notification timestamps (Mongo) are native + ``datetime`` values, matching current production behavior since + SIENTIAPDE-2007. ``load_latest_data`` parses the Redis string back into a + ``datetime`` and applies ``{$gt: }`` against the Mongo BSON dates — + this exercises the real comparison semantics rather than a synthetic ideal. """ seed_receiver_groups(mongo_uri, E2E_DATABASE, [default_receiver_group()]) - seed_last_timestamp(redis_client, 'Alerts', '2024-06-01 10:30:00+0000') + seed_last_timestamp(redis_client, 'Alerts', '2024-06-01 10:30:00.000000+0000') seed_notifications( mongo_uri, E2E_DATABASE, [ default_notification( notification_id='old', - timestamp='2024-06-01 10:00:00+0000', + timestamp=datetime(2024, 6, 1, 10, 0, 0, tzinfo=UTC), ), default_notification( notification_id='new', - timestamp='2024-06-01 11:00:00+0000', + timestamp=datetime(2024, 6, 1, 11, 0, 0, tzinfo=UTC), ), ], ) @@ -129,6 +133,78 @@ async def test_load_package_with_prior_timestamp( assert ids == {'new'} +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_load_package_query_time_filter_union_and_intersection( + temporal_env: WorkflowEnvironment, + orchestrator_worker, + stub_workers, + mongo_uri: str, + redis_client: Redis, +): + """ + SIENTIAPDE-2011: query-time filter unions active groups' levels and + intersects their ignore_models_id, instead of fetching everything. + + Two active receiver groups narrow the fetch differently: + - group_a: levels=['ERROR'], ignore_models_id=['model-x'] + - group_b: levels=['WARNING'], ignore_models_id=['model-x', 'model-y'] + + The Mongo query must return exactly what the UNION of levels allows + (ERROR or WARNING, never INFO) and must exclude only models in the + INTERSECTION of every active group's ignore list (model-x, ignored by + both). A model ignored by only one group (model-y) must still be + fetched, since the other group still wants it — per-group correctness + for that case is SIENTIAPDE-2012's job, not this query-time filter's. + """ + seed_receiver_groups( + mongo_uri, + E2E_DATABASE, + [ + default_receiver_group( + group_name='group_a', levels=['ERROR'], ignore_models_id=['model-x'] + ), + default_receiver_group( + group_name='group_b', + levels=['WARNING'], + ignore_models_id=['model-x', 'model-y'], + ), + ], + ) + seed_notifications( + mongo_uri, + E2E_DATABASE, + [ + default_notification( + notification_id='error-included', level='ERROR', model_id='model-a' + ), + default_notification( + notification_id='warning-included', level='WARNING', model_id='model-a' + ), + default_notification(notification_id='info-excluded', level='INFO', model_id='model-a'), + default_notification( + notification_id='modelx-excluded', level='ERROR', model_id='model-x' + ), + default_notification( + notification_id='modely-included', level='WARNING', model_id='model-y' + ), + ], + ) + + input_data = load_scenario_input('subworkflow_load_notification_package.json') + _build_metadata(input_data) + + result = await start_and_await_workflow( + temporal_env.client, + LoadNotificationPackage.run, + input_data, + make_workflow_id('load-pkg-filter'), + ) + + ids = {n['notification_id'] for n in result['notification_package']} + assert ids == {'error-included', 'warning-included', 'modely-included'} + + @pytest.mark.e2e @pytest.mark.asyncio async def test_load_package_empty_mongo_no_redis_write( diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index e0f778a..38113ea 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -7,7 +7,7 @@ with workflow.unsafe.imports_passed_through(): from logging import Logger from typing import Any - from pandas import DataFrame + from pandas import DataFrame, to_datetime from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring @@ -331,14 +331,25 @@ class SlotManager(SientiaMonitoring): self.warning('No data to insert', metadata=metadata) return None - last_data_timestamp = data['timestamp'].max() + # ``timestamp`` started out as a native ``datetime``/BSON Date in Mongo, + # but by the time it gets here it has crossed a Temporal activity + # boundary (this activity's ``input_data`` is whatever + # ``load_latest_data`` returned): Temporal's default data converter + # serializes ``datetime`` values to plain ISO-8601 strings and does not + # restore them on decode, so ``data['timestamp']`` is a ``str`` in a + # real workflow run even though unit tests mocking the activity in + # isolation never see that conversion. pymongo also returns naive + # datetimes for BSON Date fields (no tzinfo), so the ISO string may + # carry no UTC offset at all. ``utc=True`` normalizes every case + # (native datetime/Timestamp, offset-aware or naive ISO string) into a + # UTC-aware ``Timestamp`` — naive values are BSON Dates, always UTC. + last_data_timestamp = to_datetime(data['timestamp'], utc=True).max() self.debug(f'Last collected timestamp to insert: {last_data_timestamp}', metadata=metadata) - # ``timestamp`` is a native ``datetime``/``Timestamp`` (BSON Date from - # Mongo), which the Redis repository's plain ``json.dumps`` cannot - # serialize. Store it as a string formatted per - # ``DATETIME_FORMAT_MS_WITH_TZ`` instead. + # The Redis repository's plain ``json.dumps`` cannot serialize a + # ``Timestamp``/``datetime`` directly, so store it as a string + # formatted per ``DATETIME_FORMAT_MS_WITH_TZ`` instead. last_data_timestamp = last_data_timestamp.strftime(DATETIME_FORMAT_MS_WITH_TZ) try: From ed666c260d79e9858cd3206d87f28ab8abc42dc3 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 15 Sep 2026 08:18:17 -0300 Subject: [PATCH 3/6] fix: satisfy mypy in put_last_data_timestamp after utc=True normalization CI mypy caught what my local environment couldn't (Python 3.14 has no compatible numpy stub for the project's python_version=3.11 mypy target, so mypy couldn't run locally at all): slot_manager.py:353: error: Incompatible types in assignment (expression has type "str", variable has type "Timestamp") slot_manager.py:368: error: Incompatible return value type (got "Timestamp", expected "str | None") Reassigning `last_data_timestamp` from a Timestamp to its strftime() string trips mypy's inference (it fixes an unannotated local's type from its first assignment). Split into a separate `last_data_timestamp_str` for the string form instead of reusing the name. Co-Authored-By: Claude Sonnet 5 --- orchestrator/activities/slot_manager.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index 38113ea..bbb9ea6 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -349,11 +349,15 @@ class SlotManager(SientiaMonitoring): # The Redis repository's plain ``json.dumps`` cannot serialize a # ``Timestamp``/``datetime`` directly, so store it as a string - # formatted per ``DATETIME_FORMAT_MS_WITH_TZ`` instead. - last_data_timestamp = last_data_timestamp.strftime(DATETIME_FORMAT_MS_WITH_TZ) + # formatted per ``DATETIME_FORMAT_MS_WITH_TZ`` instead. Kept as a + # separate variable (rather than reassigning ``last_data_timestamp``) + # since it's a different type — mypy infers a local's type from its + # first assignment and flags a same-name reassignment to an + # incompatible type as an error. + last_data_timestamp_str = last_data_timestamp.strftime(DATETIME_FORMAT_MS_WITH_TZ) try: - self.redis_repository.set(key, last_data_timestamp, ttl=60 * 60 * 5) + self.redis_repository.set(key, last_data_timestamp_str, ttl=60 * 60 * 5) except Exception as e: self.send_notification( metadata=metadata, @@ -365,7 +369,7 @@ class SlotManager(SientiaMonitoring): ) raise e - return last_data_timestamp + return last_data_timestamp_str @activity.defn(name='filter_notification_alerts') def filter_notification_alerts(self, input_data: dict[str, Any]) -> dict[str, Any]: From 58c0bd4f2da1363fb6361f99758a91e49faec7a7 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 15 Sep 2026 09:35:25 -0300 Subject: [PATCH 4/6] SIENTIAPDE-2011: normalize notification timestamp in format_log_report notification['timestamp'] crosses at least one Temporal activity boundary (load_latest_data -> filter_notification_reports/alerts) before reaching format_log_report. Temporal's default data converter serializes datetime values to plain ISO-8601 strings without restoring them on decode, and pymongo returns naive datetimes for BSON Date fields, so the string can carry no UTC offset at all. That broke export_data_to_postgres's strict to_datetime(..., format= DATETIME_FORMAT_WITH_TZ) parsing downstream with: ValueError: time data "2024-06-01T12:00:00" doesn't match format "%Y-%m-%d %H:%M:%S%z" Same underlying issue as SIENTIAPDE-2007/2009, fixed here using the same pattern as SlotManager.put_last_data_timestamp: normalize with pandas.to_datetime(..., utc=True) and re-format to DATETIME_FORMAT_WITH_TZ before the value crosses the next boundary. Co-Authored-By: Claude Sonnet 5 --- orchestrator/activities/formatters.py | 25 ++++++++++++++++--- .../activities/test_formatters.py | 2 +- 2 files changed, 23 insertions(+), 4 deletions(-) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 67efdf5..79a2701 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -7,11 +7,15 @@ with workflow.unsafe.imports_passed_through(): from math import ceil from typing import Any, TypedDict, cast - from pandas import DataFrame + from pandas import DataFrame, to_datetime from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring - from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now + from sientia_do.temporal.constants import ( + DATETIME_FORMAT_MS_WITH_TZ, + DATETIME_FORMAT_WITH_TZ, + now, + ) from orchestrator.utils.orchestrator_functions import ( build_tag_config, @@ -755,9 +759,24 @@ class Formatters(SientiaMonitoring): key = f'{notification_id}:{trigger}' if key not in data: + # ``notification['timestamp']`` started out as a native + # ``datetime``/BSON Date in Mongo, but by the time it gets here + # it has crossed at least one Temporal activity boundary: + # Temporal's default data converter serializes ``datetime`` + # values to plain ISO-8601 strings and does not restore them on + # decode, and pymongo returns naive datetimes for BSON Date + # fields (no tzinfo), so the ISO string may carry no UTC offset + # at all. Normalizing with ``utc=True`` and re-formatting to + # ``DATETIME_FORMAT_WITH_TZ`` (same pattern as + # ``SlotManager.put_last_data_timestamp``) keeps the value a + # fixed-format string that ``export_data_to_postgres``'s + # ``timestamp_conversion`` can parse after the next boundary + # crossing, regardless of what shape it arrived in here. data[key] = { 'status': group_config['status'], - 'timestamp': notification['timestamp'], + 'timestamp': to_datetime(notification['timestamp'], utc=True).strftime( + DATETIME_FORMAT_WITH_TZ + ), 'groups': [group_name], 'message': notification['message'], 'level': notification['level'], diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index 70b74ab..c85c745 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -716,7 +716,7 @@ def test_format_log_report(formatters): [ { 'status': 'sent', - 'timestamp': '2021-01-01', + 'timestamp': '2021-01-01 00:00:00+0000', 'groups': ['test_receiver_group', 'test_receiver_group2'], 'message': 'test_message', 'level': 'test_level', From 6479737dfb2130da788cfe1156a2442829110e06 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 15 Sep 2026 11:13:17 -0300 Subject: [PATCH 5/6] fix: harden ignore_models_id against explicit null, document levels-union design Code review findings on the SIENTIAPDE-2011 PR: - [Major] _compute_query_time_filter used group.get('ignore_models_id', []) for the intersection, whose default only applies when the key is missing. receiver_groups has no schema validation, so a document with ignore_models_id explicitly set to null (key present, value None) made .get() return None, and set(None) raised TypeError, crashing the whole activity for every mail type. Fixed with `group.get(...) or []`. Added a regression test. - [Minor] CLAUDE.md:48 described Alerts' base data filter as always `level: ERROR`, which this ticket's own query-time union can now widen. Updated the line to reflect that. - [Blocker, not applied] A finding argued the levels union should AND with default_filter's own level restriction instead of replacing it, to avoid widening Alerts (ERROR-only real-time channel) when a receiver_group sets broader levels. Both SIENTIAPDE-2011's ticket text and the parent design doc explicitly call for the union to replace the hard-coded ERROR-only default ("whatever levels the active groups actually care about"), so applying that fix would revert the subtask's actual purpose. The interim gap the finding correctly identifies (no per-group level filter exists yet until SIENTIAPDE-2012 lands) is now documented in _compute_query_time_filter's docstring instead of "fixed" by narrowing query-time behavior the ticket asked for. Co-Authored-By: Claude Sonnet 5 --- CLAUDE.md | 2 +- .../subworkflows/load_notification_package.py | 16 ++++++++++++++- .../test_load_notification_package.py | 20 +++++++++++++++++++ 3 files changed, 36 insertions(+), 2 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 66e7092..5ab46c1 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -45,7 +45,7 @@ CI quality gate (`.github/workflows/quality-gate.yml`) calls the shared reusable ### Layering - **`orchestrator/worker/worker.py`** — entrypoint (`python -m orchestrator.worker.worker`). Builds a single `Activities` instance, connects to Temporal, starts Prometheus (`HTTP_METRICS_PORT`) and Temporal SDK metrics (`HTTP_SDK_METRICS_PORT`) servers, then runs three `Worker`s concurrently via `asyncio.gather`, each on its own task queue: `orchestrator-queue`, `alerts-queue`, `reports-queue`. `prepare_worker.py` derives the queue name from the workflow class name (`camel_to_snake`) and reads per-queue concurrency/poller tuning from env vars prefixed with the workflow name (e.g. `ORCHESTRATOR_MAX_CONCURRENT_ACTIVITIES`). -- **`orchestrator/workflows/`** — Temporal workflow definitions: `orchestrator.py` (pipeline/slot/schedule orchestration), `alerts.py` (real-time ERROR notifications), `reports.py` (scheduled all-level reports). `alerts.py` and `reports.py` both delegate to the same two subworkflows in `workflows/subworkflows/`: `load_notification_package.py` (load notifications + receiver configs from Mongo, incremental via Redis timestamp) and `process_notifications.py` (build HTML email, send, log to Postgres). The difference between alerts and reports is which filter activity they call (`filter_notification_alerts` vs `filter_notification_reports`) and their base data filter (`level: ERROR` vs all levels). +- **`orchestrator/workflows/`** — Temporal workflow definitions: `orchestrator.py` (pipeline/slot/schedule orchestration), `alerts.py` (real-time ERROR notifications), `reports.py` (scheduled all-level reports). `alerts.py` and `reports.py` both delegate to the same two subworkflows in `workflows/subworkflows/`: `load_notification_package.py` (load notifications + receiver configs from Mongo, incremental via Redis timestamp) and `process_notifications.py` (build HTML email, send, log to Postgres). The difference between alerts and reports is which filter activity they call (`filter_notification_alerts` vs `filter_notification_reports`) and their base data filter (`level: ERROR` by default vs all levels) — `load_notification_package.py`'s `_compute_query_time_filter` can widen either beyond that default to the union of active receiver groups' own `levels` (SIENTIAPDE-2011); the ERROR-only default only holds when no active group narrows further. - **`orchestrator/activities/activities.py`** — `Activities` is a single class combining every activity mixin via multiple inheritance: `TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres` (Postgres comes from `sientia_do.temporal.activities.postgres`, the external `sientia-dataops-library` dependency). One instance is constructed in `worker.py` and its bound methods are handed to each `Worker` as the activity list — activities are plain methods on this shared object, not separate classes. Each mixin's `__init__` takes its own config dict plus shared `logger`, `notification_handler`, `metrics_controller`. `Activities.shutdown()` closes every mixin's connections. - **`orchestrator/utils/`** — `connectors_config.py` builds each config dict from env vars (Temporal/Redis/MongoDB/Postgres/Email); `email_builder.py` renders Jinja2 templates (`utils/templates/email_template.html`, `general_template.html`); `orchestrator_functions.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`); `converters.py` handles Temporal schedule frequency parsing. diff --git a/orchestrator/workflows/subworkflows/load_notification_package.py b/orchestrator/workflows/subworkflows/load_notification_package.py index 0f28ec2..614899c 100644 --- a/orchestrator/workflows/subworkflows/load_notification_package.py +++ b/orchestrator/workflows/subworkflows/load_notification_package.py @@ -122,6 +122,17 @@ class LoadNotificationPackage: restriction (missing/empty `levels`), that group wants every level, so the query can't restrict levels at all -> falls back to whatever level restriction `default_filter` itself carries (e.g. ERROR-only for Alerts). + + When every active group DOES restrict `levels`, the union REPLACES + `default_filter`'s own level restriction rather than ANDing with it — + this is deliberate, not an oversight: SIENTIAPDE-2011's design doc and + ticket text both call for Alerts' hard-coded `level: ERROR` to become + "whatever levels the active groups actually care about", falling back + to ERROR-only only when no group narrows further. Until SIENTIAPDE-2012 + (per-group `levels` filtering) lands, this means a single active group + configuring `levels` to include WARNING/INFO widens the Alerts fetch for + every active group, not just that one — an accepted interim gap, not a + bug in this function. - ignore_models_id: intersection of every group's `ignore_models_id`. A model is only safe to drop from the fetch if EVERY active group ignores it. A group with no `ignore_models_id` is treated as "ignores nothing" (empty @@ -141,8 +152,11 @@ class LoadNotificationPackage: else: query_filter = dict(default_filter) + # `or []`, not a `.get(..., [])` default: a document with `ignore_models_id` + # explicitly set to `null` (no schema validation on this collection) makes + # `.get()` return None, and `set(None)` would raise TypeError. sending_configs_ignore_models_id = [ - set(group.get('ignore_models_id', [])) for group in sending_configs + set(group.get('ignore_models_id') or []) for group in sending_configs ] models_intersection = set.intersection(*sending_configs_ignore_models_id) diff --git a/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py b/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py index 9f7d01b..4f049be 100644 --- a/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py +++ b/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py @@ -233,3 +233,23 @@ def test_compute_query_time_filter_union_levels_and_intersection_models( 'level': {'$in': ['ERROR', 'WARNING']}, 'model_id': {'$nin': ['model_a']}, } + + +def test_compute_query_time_filter_ignore_models_id_explicit_none_does_not_raise( + load_notification_package, +): + # receiver_groups has no schema validation, so a document can have + # ignore_models_id explicitly set to null (key present, value None) rather + # than omitted. `group.get('ignore_models_id', [])` would still return None + # in that case (the default only applies when the key is missing), and + # `set(None)` raises TypeError -- must not crash the whole activity. + sending_configs = [ + {'group_name': 'group_a', 'levels': ['ERROR'], 'ignore_models_id': None}, + {'group_name': 'group_b', 'levels': ['ERROR'], 'ignore_models_id': ['model_a']}, + ] + default_filter = {} + + result = load_notification_package._compute_query_time_filter(sending_configs, default_filter) + + assert result == {'level': {'$in': ['ERROR']}} + assert 'model_id' not in result From 3d2a5ecc43c8af965f69ece910d5bd10c1b21f09 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 15 Sep 2026 11:26:06 -0300 Subject: [PATCH 6/6] test: lock in that levels union replaces default_filter's level, not AND Manual code-review pass (code-review-python checklist) flagged that no test exercised default_filter already restricting level (e.g. Alerts' ERROR-only default) combined with every active group specifying its own levels. The replace-not-AND behavior is deliberate per SIENTIAPDE-2011's design doc and ticket text, but was previously only covered by a docstring comment. Adds test_compute_query_time_filter_union_replaces_default_level_restriction so a future change reintroducing AND semantics fails a test instead of passing silently. Co-Authored-By: Claude Sonnet 5 --- .../test_load_notification_package.py | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py b/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py index 4f049be..3266f70 100644 --- a/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py +++ b/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py @@ -199,6 +199,27 @@ def test_compute_query_time_filter_mixed_groups_no_levels_restriction(load_notif assert result == {'level': 'ERROR'} +def test_compute_query_time_filter_union_replaces_default_level_restriction( + load_notification_package, +): + # default_filter already restricts level (e.g. Alerts' ERROR-only default), + # but every active group specifies levels of its own -> the union REPLACES + # default_filter's level restriction, it does not AND with it. This is + # deliberate per SIENTIAPDE-2011's design doc and ticket text ("Alerts' + # hard-coded level: ERROR should become whatever levels the active groups + # actually care about"), not a bug -- this test locks the behavior in so a + # future "fix" reintroducing AND semantics fails loudly instead of passing + # silently. + sending_configs = [ + {'group_name': 'group_a', 'levels': ['WARNING'], 'ignore_models_id': []}, + ] + default_filter = {'level': 'ERROR'} + + result = load_notification_package._compute_query_time_filter(sending_configs, default_filter) + + assert result == {'level': {'$in': ['WARNING']}} + + def test_compute_query_time_filter_model_ignored_by_only_one_group_not_excluded( load_notification_package, ):