SIENTIAPDE-1974

SIENTIAPDE-1974-connectors-pi-web-api: Introduce config_hash for robust schedule redeployment. Pipelines now redeploy on any content change (e.g., tag database updates) even without an updated_at bump. orchestrated_schedules tracks the hash; updated documentation, E2E tests, and the SIENTIAPDE-1974 backfill script.

CLAUDE.md

@@ -47,7 +47,8 @@ CI quality gate (`.github/workflows/quality-gate.yml`) calls the shared reusable
 - **`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` 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. Receiver groups also support a `model_ids` allow-list (SIENTIAPDE-2149), independent of the pre-existing `ignore_models_id` deny-list — both apply together (AND) when a group configures both, and `model_ids` follows the same union-at-query-time / precise-per-group-slice split as `levels`. `load_latest_data`'s Mongo query is capped at `NOTIFICATION_QUEUE_BATCH_SIZE` documents per run (default `0`, meaning unlimited — unchanged from pre-SIENTIAPDE-2013 behavior unless a deployment opts in), sorted by `timestamp` ascending so a capped run always returns the oldest pending batch first (SIENTIAPDE-2013) — a hard cap with no drain loop, since the incremental Redis cursor picks up any remainder on the next run.
 - **`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`); `pipeline_configs.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`); `tag_database.py` indexes the tag database (`connectors` + `connector_tags`) and resolves tags by `tag_id` (scouter read tag names and the `predictions_batch` write tags → `opc_output_config` keyed by `connector_id`, via `resolve_write_tags`; a pipeline's effective `updated_at` also follows its tags' `updated_at`, so tag changes redeploy it); `opc_slots.py` builds the OPC slot v2 the gateway reads (`gather_read_tags` / `build_tag_config`, contract C1: one entry per address with a `names[]` list; see the README section "OPC slots"). `resolve_tag_names` (in `tag_database.py`) is the single source of tag names for both the scouter's `model_tags` and the slot's `names` (SIENTIAPDE-1974 SC1); `converters.py` handles Temporal schedule frequency parsing.
+- **`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`); `pipeline_configs.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`); `tag_database.py` indexes the tag database (`connectors` + `connector_tags`) and resolves tags by `tag_id` (scouter read tag names and the `predictions_batch` write tags → `opc_output_config` keyed by `connector_id`, via `resolve_write_tags`; tag changes reach the schedules through `config_hash`, see below); `opc_slots.py` builds the OPC slot v2 the gateway reads (`gather_read_tags` / `build_tag_config`, contract C1: one entry per address with a `names[]` list; see the README section "OPC slots"). `resolve_tag_names` (in `tag_database.py`) is the single source of tag names for both the scouter's `model_tags` and the slot's `names` (SIENTIAPDE-1974 SC1); `converters.py` handles Temporal schedule frequency parsing.
+- **Schedule redeploys** — `orchestrated_schedules` stores, per schedule, the last deploy time (`updated_at`) and the `config_hash` it was deployed with (`formatters.schedule_config_hash`: SHA-256 of the built schedule without `updated_at`). `compare_config_timestamps` redeploys when the pipeline's `updated_at` is newer (manual force) or the hash differs (any content change, including tag database edits/deletes made without bumping `updated_at`). `create_schedules`/`update_schedules` pop `config_hash` before sending the schedule to Temporal and return it in their report for `create_pipelines_timestamps`/`update_pipelines_timestamps`.
 
 ### Temporal-specific conventions (all workflow/activity files follow this)
 - Every workflow and activity-consuming module wraps non-deterministic/side-effecting imports in `with workflow.unsafe.imports_passed_through(): ...` right after `from temporalio import workflow`, because Temporal's sandbox restricts what workflow code can import directly. When adding a new workflow file, follow the existing pattern (see `orchestrator/workflows/orchestrator.py`, `orchestrator/activities/activities.py`) rather than importing at module top level.

README.md

@@ -224,15 +224,21 @@ The legacy write tags (`server_id`, `type`, `addr`, `data_type` in the pipeline)
 - `ORCHESTRATOR_WRITE_TAG_CONNECTOR_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION`: the tag's connector is missing, inactive or incomplete in `connectors`;
 - `ORCHESTRATOR_INVALID_WRITE_TAG_DURING_SCHEDULE_CONFIGURATION`: the entry has no `tag_id` (a legacy write tag), or the tag's `data_type` is not one the gateway accepts.
 
-#### Redeploy on tag database changes
+#### Redeploy on changes (`updated_at` and `config_hash`)
 
-A schedule is redeployed when its `updated_at` is newer than the last time it was orchestrated. That `updated_at` is the latest of the pipeline's own and of the `updated_at` of every `connector_tags` document it references (read and write tags, active or not). So renaming a read tag, moving or retyping a write tag, or deactivating a tag redeploys the pipelines that use it on the next run, without touching them. The slot is compared by content and follows on its own. A pipeline without `updated_at` is still redeployed on every run. A deleted tag document (as opposed to a deactivated one) is not detected: deactivate tags instead of deleting them.
+`orchestrated_schedules` keeps, per schedule, when it was last deployed (`updated_at`) and the `config_hash` it was deployed with: a SHA-256 of the schedule as the Orchestrator builds it (the workflow input, frequency and offset, with tags and connectors already resolved), without `updated_at`. A schedule is redeployed when either:
+
+- the pipeline's `updated_at` is newer than its last deploy. Bumping `updated_at` by hand forces a redeploy even when nothing else changed; or
+- its `config_hash` differs from the stored one. Any change that reaches the schedule redeploys it, whether or not someone bumped an `updated_at`: a pipeline edited by hand, a tag renamed, moved, retyped, deactivated or deleted in `connector_tags`, a connector changed, or a new Orchestrator version that builds the schedule differently.
+
+A change that does not reach the schedule (e.g. a tag's `description`) does not redeploy. The slot is compared by content and follows on its own. A pipeline without `updated_at` is still redeployed on every run. Schedules deployed before `config_hash` existed have none stored, so the first run of this version redeploys all of them once and records it.
 
 #### Deploying the tag database write tags
 
 1. Run the migration that adds `data_type` to the `connector_tags` validator (`sientia-dataops-database-migrations`).
-2. Run `scripts/sientiapde_1974_backfill_write_tags.js`, first with `DRY_RUN=1` (the default). It creates the `connector_tags` of the legacy write tags, rewrites `write_tags` as `[{tag_id}]` (keeping the original in `legacy_write_tags`) and bumps the pipelines' `updated_at`.
-3. Deploy this version.
+2. Paste `scripts/sientiapde_1974_backfill_write_tags.js` into the MongoDB Compass shell (or mongosh), with `ACTION = 'backfill'`: first with `DRY_RUN = true` (the default) to see the plan, then with `DRY_RUN = false`. It creates the `connector_tags` of the legacy write tags, rewrites `write_tags` as `[{tag_id}]` (keeping the original in `legacy_write_tags`) and bumps the pipelines' `updated_at`. `ACTION = 'revert'` undoes it.
+3. Deploy this version. Its first run redeploys every schedule once, to record their `config_hash` (see above).
+4. Once the write tags are confirmed, run the script with `ACTION = 'cleanup'` to remove `legacy_write_tags`. After that, `revert` can no longer restore the pipelines.
 
 #### Architecture
 

e2e/scenarios.md

@@ -83,9 +83,12 @@ from `connector_tags` (`float` when absent), `Write` → `prediction_tags`, `Pre
 
 #### 1.9.2 A tag change redeploys its pipelines
 
-Renaming the read tag and moving the write tag in `connector_tags` (with a newer `updated_at`),
-without touching the pipelines → the next run redeploys both: the scouter `model_tags` and the
-slot `names` take the new name, the `opc_output_config` the new address.
+A second run with nothing changed redeploys nothing (`orchestrated_schedules` is not stamped
+again). Then renaming the read tag, moving the write tag and deleting the confidence tag in
+`connector_tags` by hand (no `updated_at` bumped anywhere, pipelines untouched) → the next run
+redeploys both through their `config_hash`: the scouter `model_tags` and the slot `names` take
+the new name, the `opc_output_config` the new address without the confidence tag, and
+`orchestrated_schedules` records the new `config_hash`.
 
 #### 1.9.3 OPC slot from the tag database
 

e2e/test_orchestrator_main_workflow.py

@@ -402,6 +402,21 @@ def _opc_slots(redis_client: Redis) -> dict[str, dict]:
     }
 
 
+def _orchestrated_schedules(mongo_uri: str) -> dict[str, dict]:
+    """orchestrated_schedules by schedule_name: updated_at (last deploy) and config_hash."""
+    mongo = MongoClient(mongo_uri)
+    try:
+        return {
+            doc['schedule_name']: {
+                'updated_at': doc['updated_at'],
+                'config_hash': doc.get('config_hash'),
+            }
+            for doc in mongo[E2E_DATABASE]['orchestrated_schedules'].find()
+        }
+    finally:
+        mongo.close()
+
+
 @pytest.mark.e2e
 @pytest.mark.asyncio
 async def test_scenario_1_9_1_write_tags_through_tag_database(
@@ -452,7 +467,10 @@ async def test_scenario_1_9_2_tag_change_redeploys_pipelines(
     temporal_host: str,
     redis_client: Redis,
 ):
-    """1.9.2: changing a tag in connector_tags redeploys the pipelines that reference it."""
+    """
+    1.9.2: changing or deleting a tag in connector_tags, without touching any updated_at,
+    redeploys the pipelines that reference it (config_hash); an unchanged run does not.
+    """
     _apply_orchestrator_seeds(
         load_scenario_input('orchestrator_tag_database'), mongo_uri, redis_client
     )
@@ -460,16 +478,21 @@ async def test_scenario_1_9_2_tag_change_redeploys_pipelines(
 
     scouter = await _schedule_input(temporal_host, 'opc-tagdb', namespace='scouter')
     assert set(scouter['model_tags']) == {'Pressure'}
+    deployed = _orchestrated_schedules(mongo_uri)
+    assert all(doc.get('config_hash') for doc in deployed.values()), deployed
 
-    # Rename the read tag and move the write tag, without touching the pipelines
-    changed_at = datetime.now(UTC) + timedelta(seconds=1)
+    # Nothing changed: no redeploy, so orchestrated_schedules is not stamped again
+    await _run_tag_database_scenario(temporal_env, 'unchanged')
+    assert _orchestrated_schedules(mongo_uri) == deployed
+
+    # As an edit by hand in Compass: rename the read tag, move a write tag and delete the
+    # confidence tag, with no updated_at bumped anywhere
     mongo = MongoClient(mongo_uri)
     try:
         tags = mongo[E2E_DATABASE]['connector_tags']
-        tags.update_one({'tag_id': 7}, {'$set': {'tag_name': 'Pressao', 'updated_at': changed_at}})
-        tags.update_one(
-            {'tag_id': 101}, {'$set': {'address': 'ns=3;i=3001', 'updated_at': changed_at}}
-        )
+        tags.update_one({'tag_id': 7}, {'$set': {'tag_name': 'Pressao'}})
+        tags.update_one({'tag_id': 101}, {'$set': {'address': 'ns=3;i=3001'}})
+        tags.delete_one({'tag_id': 102})
     finally:
         mongo.close()
 
@@ -478,9 +501,13 @@ async def test_scenario_1_9_2_tag_change_redeploys_pipelines(
     scouter = await _schedule_input(temporal_host, 'opc-tagdb', namespace='scouter')
     assert set(scouter['model_tags']) == {'Pressao'}
     pred = await _schedule_input(temporal_host, 'pred-tagdb')
-    assert pred['opc_output_config']['plant']['prediction_tags'] == {
-        'ns=3;i=3001': {'data_type': 'double'}
+    assert pred['opc_output_config'] == {
+        'plant': {'prediction_tags': {'ns=3;i=3001': {'data_type': 'double'}}}
     }
+    redeployed = _orchestrated_schedules(mongo_uri)
+    for name in ('opc-tagdb', 'pred-tagdb'):
+        assert redeployed[name]['config_hash'] != deployed[name]['config_hash']
+        assert redeployed[name]['updated_at'] > deployed[name]['updated_at']
     # The slot names the read tag like the scouter does (SC1)
     [slot] = _opc_slots(redis_client).values()
     [name] = slot['plant']['tags']['ns=3;i=1001']['names']

orchestrator/activities/formatters.py

@@ -1,6 +1,7 @@
 from temporalio import activity, workflow
 
 with workflow.unsafe.imports_passed_through():
+    import hashlib
     import json
     from collections.abc import Callable
     from logging import Logger
@@ -33,15 +34,15 @@ with workflow.unsafe.imports_passed_through():
         WRITE_WORKFLOW_TYPES,
         index_connector_tags,
         index_connectors,
-        index_tag_documents,
-        latest_updated_at,
         resolve_tag_names,
         resolve_write_tags,
-        tag_documents_updated_at,
     )
 
 topic_separator = '\n ========== \n'
 
+# Schedule fields left out of config_hash: updated_at is compared on its own
+CONFIG_HASH_EXCLUDED_FIELDS = ('updated_at', 'config_hash')
+
 # Read tags left out of the slots, by gather_read_tags issue kind
 SLOT_ISSUES = (
     (
@@ -168,6 +169,20 @@ Each entry specifies:
 """
 
 
+def schedule_config_hash(schedule: dict[str, Any]) -> str:
+    """
+    Fingerprint of a schedule as process_schedules builds it, without updated_at.
+
+    Stored in orchestrated_schedules when the schedule is deployed: any change to what
+    reaches Temporal (a pipeline edited without bumping updated_at, a tag renamed, moved,
+    retyped, deactivated or deleted, a connector changed, or a new Orchestrator version
+    building it differently) changes the hash and redeploys the schedule.
+    """
+    content = {k: v for k, v in schedule.items() if k not in CONFIG_HASH_EXCLUDED_FIELDS}
+    serialized = json.dumps(content, sort_keys=True, separators=(',', ':'), default=str)
+    return hashlib.sha256(serialized.encode()).hexdigest()
+
+
 class Formatters(SientiaMonitoring):
     """
     Schedule and slot configuration formatting and notification filtering activity.
@@ -248,8 +263,8 @@ class Formatters(SientiaMonitoring):
         Scouter read tags that carry only a tag_id are named from the tag database (the
         same names as the slot). A predictions_batch gets its opc_output_config from its
         write_tags ([{tag_id}]) through the tag database; what cannot be resolved is left
-        out and notified once per run. A pipeline's updated_at is the latest of its own and
-        of the tag database documents it references, so changing a tag redeploys it.
+        out and notified once per run. A change to a tag reaches the schedule through
+        create_schedule_config, which compares the resolved schedule (config_hash).
 
         Returns:
             - dict[str, Any]: The schedule configuration dictionary keyed by namespace
@@ -263,7 +278,6 @@ class Formatters(SientiaMonitoring):
         connector_tags = input_data.get('connector_tags', [])
         read_tags_by_id = index_connector_tags(connector_tags)
         write_tags_by_id = index_connector_tags(connector_tags, WRITE_CATEGORIES)
-        tag_documents = index_tag_documents(connector_tags)
         connectors = index_connectors(input_data.get('connectors', []))
         write_issues: dict[str, set[str]] = {}
 
@@ -298,13 +312,11 @@ class Formatters(SientiaMonitoring):
                 _merge_issues(write_issues, issues, schedule_name)
                 pipeline = {**pipeline, 'opc_output_config': opc_output_config}
 
-            # A pipeline without updated_at is redeployed on every run, as before
-            updated_at = pipeline.get('updated_at') or now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
             schedule_config[namespace][schedule_name] = {
                 **function(pipeline),
-                'updated_at': latest_updated_at(
-                    updated_at, tag_documents_updated_at(pipeline, tag_documents)
-                ),
+                # A pipeline without updated_at is redeployed on every run, as before
+                'updated_at': pipeline.get('updated_at')
+                or now().strftime(DATETIME_FORMAT_MS_WITH_TZ),
             }
 
         self._notify_issues(write_issues, WRITE_TAG_ISSUES, metadata)
@@ -401,6 +413,8 @@ class Formatters(SientiaMonitoring):
 
         Transforms a list of schedule configurations into a nested dictionary structure
         organized by namespace and schedule name for efficient lookup and comparison.
+        Each schedule keeps when it was last deployed (updated_at) and the config_hash
+        it was deployed with (None for schedules deployed before config_hash existed).
 
         Args:
             input_data (dict[str, Any]): The input data containing:
@@ -408,7 +422,7 @@ class Formatters(SientiaMonitoring):
                 - metadata (dict): Metadata for logging purposes
 
         Returns:
-            dict[str, Any]: The formatted schedule config organized by namespace and schedule name
+            dict[str, Any]: {namespace: {schedule_name: {'updated_at', 'config_hash'}}}
         """
         metadata = input_data['metadata']
 
@@ -416,16 +430,18 @@ class Formatters(SientiaMonitoring):
 
         schedule_config = input_data['schedule_config']
 
-        config: dict[str, dict[str, str]] = {}
+        config: dict[str, dict[str, dict[str, Any]]] = {}
         for schedule in schedule_config:
             namespace = schedule['namespace']
             schedule_name = schedule['schedule_name']
-            updated_at = schedule['updated_at']
 
             if namespace not in config:
                 config[namespace] = {}
 
-            config[namespace][schedule_name] = updated_at
+            config[namespace][schedule_name] = {
+                'updated_at': schedule['updated_at'],
+                'config_hash': schedule.get('config_hash'),
+            }
 
         self.info('Formatted schedule config', metadata=metadata)
         self.debug(json.dumps(config, indent=4, sort_keys=True), metadata=metadata)
@@ -444,32 +460,46 @@ class Formatters(SientiaMonitoring):
         """
         Compare new and current schedules to determine which should be updated or created.
 
-        Compares schedule timestamps to identify schedules that need updating (newer timestamp)
-        or creating (schedule doesn't exist). Results are accumulated in the provided dictionaries.
+        A deployed schedule is updated when its updated_at is newer than its last deploy
+        (bumping updated_at forces a redeploy) or when its config_hash differs from the one
+        it was deployed with (any change to the resolved schedule, including tag database
+        changes and deleted tags). A schedule not deployed yet is created. Each schedule
+        sent to update or create carries its config_hash, stored once it is deployed.
+        Results are accumulated in the provided dictionaries.
 
         Args:
             schedules (dict[str, Any]): New schedules to compare
-            current_schedules (dict[str, Any]): Existing schedules to compare against
+            current_schedules (dict[str, Any]): Existing schedules to compare against, as
+                format_schedule_config gives them
             to_update (dict[str, Any]): Output accumulator for schedules that need updating
             to_create (dict[str, Any]): Output accumulator for schedules that need creating
             namespace (str): Namespace for the schedules being compared
             metadata (dict[str, Any]): Metadata for logging purposes
         """
         for schedule_name, schedule in schedules.items():
+            config_hash = schedule_config_hash(schedule)
+            deployed = {**schedule, 'config_hash': config_hash}
+
             if schedule_name in current_schedules:
                 update_timestamp = schedule.get('updated_at', now())
 
-                old_timestamp = current_schedules[schedule_name]
+                current = current_schedules[schedule_name]
+                if not isinstance(current, dict):
+                    # A run started before config_hash replays format_schedule_config's
+                    # result, which held only the timestamp
+                    current = {'updated_at': current, 'config_hash': None}
+                old_timestamp = current['updated_at']
 
                 self.debug(
-                    f'Comparing schedule {schedule_name}:{update_timestamp} vs {old_timestamp}',
+                    f'Comparing schedule {schedule_name}:{update_timestamp} vs {old_timestamp}'
+                    f' (config_hash {config_hash} vs {current.get("config_hash")})',
                     metadata=metadata,
                 )
 
-                if update_timestamp > old_timestamp:
-                    to_update[namespace][schedule_name] = schedule
+                if update_timestamp > old_timestamp or config_hash != current.get('config_hash'):
+                    to_update[namespace][schedule_name] = deployed
             else:
-                to_create[namespace][schedule_name] = schedule
+                to_create[namespace][schedule_name] = deployed
 
     @activity.defn(name='create_schedule_config')
     def create_schedule_config(self, input_data: dict[str, Any]) -> dict[str, Any]:
@@ -478,7 +508,8 @@ class Formatters(SientiaMonitoring):
 
         Compares the current schedule configuration in Temporal with the new schedule
         configuration to determine which schedules need to be created, updated, or deleted.
-        Uses timestamp comparison to identify schedules that have changed.
+        Uses the updated_at timestamps and the config_hash (compare_config_timestamps) to
+        identify schedules that have changed.
         Also sends a SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR WARNING for laborious schedules
         whose frequency_alert is below 1m (the value is kept) or unparseable (it is ignored),
         and a DRIFT_CHUNK_PERIOD_INVALID WARNING for drift schedules whose configured
@@ -523,8 +554,8 @@ class Formatters(SientiaMonitoring):
                 schedules, current_schedules, to_update, to_create, namespace, metadata
             )
 
-        # Fires whenever the schedule is created or updated (each `updated_at` change,
-        # including escalation flips).
+        # Fires whenever the schedule is created or updated (each `updated_at` or
+        # config_hash change, including escalation flips).
         changed = {**to_create[self.laborious_namespace], **to_update[self.laborious_namespace]}
         for schedule_name, schedule in changed.items():
             warning = frequency_alert_warning(schedule.get('frequency_alert'))

orchestrator/activities/mongo_db.py

@@ -224,12 +224,14 @@ class MongoDB(SientiaMonitoring):
         """
         Update `updated_at` timestamps for successfully updated pipelines.
 
-        Updates the `updated_at` field in the `orchestrated_schedules` collection
-        for all pipelines that were successfully updated in Temporal.
+        Updates the `updated_at` and `config_hash` fields in the `orchestrated_schedules`
+        collection for all pipelines that were successfully updated in Temporal, so the
+        next run compares against what was deployed.
 
         Args:
             input_data (dict[str, Any]): Input data containing:
-                - updated_pipelines (list[dict]): Pipelines with success flags to consider
+                - updated_pipelines (list[dict]): Pipelines with success flags (and
+                  config_hash) to consider
                 - metadata (dict): Metadata for logging purposes
         """
         updated_pipelines = input_data.get('updated_pipelines', [])
@@ -238,22 +240,30 @@ class MongoDB(SientiaMonitoring):
 
         self.info('Updating pipelines timestamps...', metadata=metadata)
 
-        argument = [
-            {'schedule_name': pipeline['schedule_name'], 'namespace': pipeline['namespace']}
-            for pipeline in updated_pipelines
-            if pipeline['success']
-        ]
-        if not argument:
+        succeeded = [pipeline for pipeline in updated_pipelines if pipeline['success']]
+        if not succeeded:
             # Every update failed: stamp nothing so they are retried next tick (same guard as
-            # create_pipelines_timestamps). An empty filter would match every tracked schedule.
+            # create_pipelines_timestamps).
             self.info('No successful pipeline updates to timestamp', metadata=metadata)
             return
-        data_filter = {'$or': argument}
 
         try:
-            self.mongo_db_repository.update_many(
-                'orchestrated_schedules', data_filter, {'$set': {'updated_at': date_now}}, metadata
-            )
+            # One update per schedule: each has its own config_hash
+            for pipeline in succeeded:
+                self.mongo_db_repository.update_one(
+                    'orchestrated_schedules',
+                    {
+                        'schedule_name': pipeline['schedule_name'],
+                        'namespace': pipeline['namespace'],
+                    },
+                    {
+                        '$set': {
+                            'updated_at': date_now,
+                            'config_hash': pipeline.get('config_hash'),
+                        }
+                    },
+                    metadata,
+                )
         except Exception as e:
             trace = traceback.format_exc()
             self.send_notification(
@@ -278,11 +288,13 @@ class MongoDB(SientiaMonitoring):
         Insert `updated_at` timestamps for newly created pipelines.
 
         Inserts new documents into the `orchestrated_schedules` collection
-        for all pipelines that were successfully created in Temporal.
+        for all pipelines that were successfully created in Temporal, with the
+        `config_hash` they were created with.
 
         Args:
             input_data (dict[str, Any]): Input data containing:
-                - created_pipelines (list[dict]): Pipelines with success flags to consider
+                - created_pipelines (list[dict]): Pipelines with success flags (and
+                  config_hash) to consider
                 - metadata (dict): Metadata for logging purposes
         """
         created_pipelines = input_data.get('created_pipelines', [])
@@ -297,6 +309,7 @@ class MongoDB(SientiaMonitoring):
                 'schedule_name': pipeline['schedule_name'],
                 'namespace': pipeline['namespace'],
                 'updated_at': date_now,
+                'config_hash': pipeline.get('config_hash'),
             }
             for pipeline in created_pipelines
             if pipeline['success']

orchestrator/activities/temporal_manager.py

@@ -177,7 +177,9 @@ class TemporalManager(SientiaMonitoring):
                 - schedules (dict[str, Any]): The schedules to create.
 
         Returns:
-            - list[dict[str, Any]]: Report entries for each attempted schedule creation.
+            - list[dict[str, Any]]: Report entries for each attempted schedule creation,
+              with the schedule's config_hash (create_schedule_config) for
+              create_pipelines_timestamps.
         """
 
         schedules_to_create = input_data['schedules']
@@ -198,6 +200,8 @@ class TemporalManager(SientiaMonitoring):
                 )
 
             for schedule_name, schedule in schedules.items():
+                # Stored in orchestrated_schedules, not sent to the workflow
+                config_hash = schedule.pop('config_hash', None)
                 search_attributes = TypedSearchAttributes(
                     [
                         SearchAttributePair(key=self.model_id_id_key, value=schedule['model_id']),
@@ -258,6 +262,7 @@ class TemporalManager(SientiaMonitoring):
                             'schedule_name': schedule_name,
                             'success': True,
                             'message': 'Schedule created successfully',
+                            'config_hash': config_hash,
                         }
                     )
 
@@ -272,6 +277,7 @@ class TemporalManager(SientiaMonitoring):
                             'schedule_name': schedule_name,
                             'success': False,
                             'message': str(e),
+                            'config_hash': config_hash,
                         }
                     )
 
@@ -360,7 +366,9 @@ class TemporalManager(SientiaMonitoring):
                 - schedules (dict[str, Any]): The schedules to update.
 
         Returns:
-            - list[dict[str, Any]]: Report entries for each attempted schedule update.
+            - list[dict[str, Any]]: Report entries for each attempted schedule update,
+              with the schedule's config_hash (create_schedule_config) for
+              update_pipelines_timestamps.
         """
 
         schedules_to_update = input_data['schedules']
@@ -380,6 +388,8 @@ class TemporalManager(SientiaMonitoring):
                 )
 
             for schedule_name, schedule in schedules.items():
+                # Stored in orchestrated_schedules, not sent to the workflow
+                config_hash = schedule.pop('config_hash', None)
                 try:
                     await self._update_single_schedule(client, schedule_name, schedule, metadata)
 
@@ -389,6 +399,7 @@ class TemporalManager(SientiaMonitoring):
                             'schedule_name': schedule_name,
                             'success': True,
                             'message': 'Schedule updated successfully',
+                            'config_hash': config_hash,
                         }
                     )
 
@@ -403,6 +414,7 @@ class TemporalManager(SientiaMonitoring):
                             'schedule_name': schedule_name,
                             'success': False,
                             'message': str(e),
+                            'config_hash': config_hash,
                         }
                     )
 

orchestrator/utils/tag_database.py

@@ -123,7 +123,7 @@ def filter_by_tag_ids(query: dict[str, Any], tag_ids: list[Any]) -> dict[str, An
 def index_tag_documents(connector_tags: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
     """
     Every loaded connector_tags document keyed by tag_id text, whatever its status,
-    category or completeness: the documents whose updated_at a pipeline depends on.
+    category or completeness.
     """
     indexed = {}
     for tag in connector_tags:
@@ -181,41 +181,6 @@ def index_connectors(connectors: list[dict[str, Any]]) -> dict[str, dict[str, An
     }
 
 
-def latest_updated_at(*values: Any) -> str | None:
-    """
-    The latest of the given updated_at values, or None when none is set.
-
-    The values are text in DATETIME_FORMAT_MS_WITH_TZ (UTC), the format
-    aggregate_documents_in_mongodb gives with timestamp_fields, so they compare as text.
-    """
-    timestamps = [value for value in values if isinstance(value, str) and value]
-    return max(timestamps) if timestamps else None
-
-
-def tag_documents_updated_at(
-    pipeline: dict[str, Any], tag_documents: dict[str, dict[str, Any]]
-) -> str | None:
-    """
-    The latest updated_at among the tag database documents a pipeline references.
-
-    Every referenced document counts, inactive or of another category too, so renaming,
-    moving, retyping or deactivating a tag redeploys the pipelines that use it.
-
-    Args:
-        pipeline (dict[str, Any]): A pipeline configuration
-        tag_documents (dict[str, dict[str, Any]]): From index_tag_documents
-
-    Returns:
-        str | None: The latest updated_at, or None when no referenced document has one
-    """
-    values = []
-    for tag in pipeline_tag_refs(pipeline):
-        tag_id = tag_id_text(tag) if isinstance(tag, dict) else None
-        if tag_id is not None and tag_id in tag_documents:
-            values.append(tag_documents[tag_id].get('updated_at'))
-    return latest_updated_at(*values)
-
-
 def resolve_tag_names(
     read_tags: list[dict[str, Any]], tags_by_id: dict[str, dict[str, Any]]
 ) -> list[dict[str, Any]]:

orchestrator/workflows/orchestrator.py

@@ -21,9 +21,8 @@ PROJECT = '$project'
 # connector_tags_query); the workflow puts a tag_id $match in front of
 # connector_tags_query. Every category and status of the referenced tags is loaded: the
 # slot uses the active Read tags, predictions_batch the active Write / Prediction
-# Confidence ones, and every referenced tag's updated_at redeploys its pipelines. Only
-# the fields used are projected: other attributes may hold BSON types a Temporal payload
-# cannot carry (updated_at is turned into text by timestamp_fields).
+# Confidence ones (status and category are filtered in the code). Only the fields used
+# are projected: other attributes may hold BSON types a Temporal payload cannot carry.
 DEFAULT_CONNECTORS_QUERY = {
     'collection': 'connectors',
     'aggregation': [
@@ -54,7 +53,6 @@ DEFAULT_CONNECTOR_TAGS_QUERY = {
                 'category': 1,
                 'data_type': 1,
                 'status': 1,
-                'updated_at': 1,
             }
         },
     ],
@@ -125,7 +123,6 @@ class Orchestrator:
                 **metadata,
                 # A fresh aggregation list: the activity appends a stage to the one it gets
                 'query': {**connectors_query, 'aggregation': list(connectors_query['aggregation'])},
-                'timestamp_fields': ['updated_at'],
             },
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(seconds=60),
@@ -136,7 +133,6 @@ class Orchestrator:
             {
                 **metadata,
                 'query': filter_by_tag_ids(connector_tags_query, tag_ids),
-                'timestamp_fields': ['updated_at'],
             },
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(seconds=60),

scripts/sientiapde_1974_backfill_write_tags.js

@@ -19,18 +19,31 @@
  * Every check runs before the first write: any conflict aborts with nothing written.
  * Idempotent: a converted pipeline has no legacy entry left and is skipped.
  *
- * Usage (mongosh, against the sientia database):
- *   DRY_RUN=1 mongosh "$MONGO_URI" scripts/sientiapde_1974_backfill_write_tags.js   # plan only (default)
- *   DRY_RUN=0 mongosh "$MONGO_URI" scripts/sientiapde_1974_backfill_write_tags.js   # write
- *   REVERT=1 DRY_RUN=0 mongosh "$MONGO_URI" scripts/sientiapde_1974_backfill_write_tags.js
+ * Usage: paste the whole file into the MongoDB Compass shell (_MONGOSH) or mongosh, after
+ * setting ACTION and DRY_RUN below. It runs against DB_NAME whatever database the shell
+ * is on, and can be pasted again in the same shell.
+ *   1. ACTION 'backfill', DRY_RUN true: prints the plan, writes nothing.
+ *   2. ACTION 'backfill', DRY_RUN false: converts the pipelines.
+ *   3. Once the new write tags are confirmed in production, ACTION 'cleanup', DRY_RUN
+ *      false: removes legacy_write_tags from the pipelines. After it, 'revert' can no
+ *      longer restore them.
+ *   ACTION 'revert', DRY_RUN false: restores legacy_write_tags into write_tags and deletes
+ *   the connector_tags this script created that no other pipeline references. It only
+ *   restores pipelines that still have legacy_write_tags (not cleaned up).
  */
 
+(() => {
+// ---- settings ----------------------------------------------------------------------
+const ACTION = 'backfill'; // 'backfill' | 'revert' | 'cleanup'
+const DRY_RUN = true; // false writes
+const DB_NAME = 'sientia';
+// -------------------------------------------------------------------------------------
+
+const db = globalThis.db.getSiblingDB(DB_NAME);
 const SOURCE = 'backfill-SIENTIAPDE-1974-write-tags';
 const WRITE_TYPES = ['predictions_batch', 'xgboost_predictions_batch'];
 const WRITE_CATEGORY = { prediction: 'Write', confidence: 'Prediction Confidence' };
 const DATA_TYPES = ['float', 'double', 'real', 'int', 'integer', 'bool', 'boolean', 'str', 'string'];
-const DRY_RUN = (process.env.DRY_RUN ?? '1') !== '0';
-const REVERT = process.env.REVERT === '1';
 
 const hasValue = (value) => value !== undefined && value !== null && value !== '';
 
@@ -158,10 +171,10 @@ const backfill = () => {
   );
   if (plan.errors.length > 0) {
     print(`Aborted, nothing was written:\n- ${plan.errors.join('\n- ')}`);
-    quit(1);
+    return;
   }
   if (DRY_RUN) {
-    print('DRY_RUN: nothing was written (run with DRY_RUN=0 to apply)');
+    print('DRY_RUN: nothing was written (set DRY_RUN = false to apply)');
     return;
   }
 
@@ -216,10 +229,24 @@ const backfill = () => {
 
 const revert = () => {
   const pipelines = db.pipelines.find({ legacy_write_tags: { $exists: true } }).toArray();
-  const created = db.connector_tags.countDocuments({ source: SOURCE });
-  print(`Pipelines to restore: ${pipelines.length}; connector_tags to delete: ${created}`);
+  // A created tag still referenced by a pipeline that is not restored (cleaned up, or
+  // pointed at it later) is kept
+  const inUse = new Set(
+    db.pipelines
+      .find({ legacy_write_tags: { $exists: false } }, { read_tags: 1, write_tags: 1 })
+      .toArray()
+      .flatMap((p) => [...(Array.isArray(p.read_tags) ? p.read_tags : []), ...(Array.isArray(p.write_tags) ? p.write_tags : [])])
+      .filter((tag) => hasValue(tag?.tag_id))
+      .map((tag) => String(tag.tag_id))
+  );
+  const created = db.connector_tags.find({ source: SOURCE }, { tag_id: 1 }).toArray();
+  const toDelete = created.filter((tag) => !inUse.has(String(tag.tag_id))).map((tag) => tag.tag_id);
+  print(
+    `Pipelines to restore: ${pipelines.length}; connector_tags to delete: ${toDelete.length}` +
+      `; kept (still referenced): ${created.length - toDelete.length}`
+  );
   if (DRY_RUN) {
-    print('DRY_RUN: nothing was written (run with DRY_RUN=0 to apply)');
+    print('DRY_RUN: nothing was written (set DRY_RUN = false to apply)');
     return;
   }
   const now = new Date();
@@ -229,9 +256,29 @@ const revert = () => {
       { $set: { write_tags: pipeline.legacy_write_tags, updated_at: now }, $unset: { legacy_write_tags: '' } }
     );
   }
-  const { deletedCount } = db.connector_tags.deleteMany({ source: SOURCE });
+  const { deletedCount } = db.connector_tags.deleteMany({ source: SOURCE, tag_id: { $in: toDelete } });
   print(`pipelines restored: ${pipelines.length}; connector_tags deleted: ${deletedCount}`);
 };
 
-if (REVERT) revert();
-else backfill();
+// Run once the new write tags are confirmed: drops the copy that revert restores from
+const cleanup = () => {
+  const filter = { legacy_write_tags: { $exists: true } };
+  const count = db.pipelines.countDocuments(filter);
+  print(`Pipelines with legacy_write_tags to clean: ${count}`);
+  if (DRY_RUN) {
+    print('DRY_RUN: nothing was written (set DRY_RUN = false to apply)');
+    return;
+  }
+  // updated_at is not bumped: the Orchestrator does not read legacy_write_tags
+  const { modifiedCount } = db.pipelines.updateMany(filter, { $unset: { legacy_write_tags: '' } });
+  print(`legacy_write_tags removed from ${modifiedCount} pipeline(s)`);
+};
+
+const actions = { backfill, revert, cleanup };
+if (!actions[ACTION]) {
+  print(`Unknown ACTION ${ACTION}: use backfill, revert or cleanup`);
+  return;
+}
+print(`${ACTION}${DRY_RUN ? ' (DRY_RUN)' : ''} on ${DB_NAME}`);
+actions[ACTION]();
+})();

tests/orchestrator/activities/test_formatters.py

@@ -5,7 +5,7 @@ from pandas import DataFrame
 from pytest import fixture
 from sientia_do.notifications.models import NotificationLevel
 
-from orchestrator.activities.formatters import Formatters
+from orchestrator.activities.formatters import Formatters, schedule_config_hash
 from orchestrator.utils.pipeline_configs import predictions_batch
 
 
@@ -616,7 +616,13 @@ def test_process_schedules_names_scouter_tags_like_the_slot(formatters):
 def test_format_schedule_config(formatters):
     input_data = {
         'schedule_config': [
-            {'namespace': 'test_namespace1', 'schedule_name': 'test1', 'updated_at': '2021-01-01'},
+            {
+                'namespace': 'test_namespace1',
+                'schedule_name': 'test1',
+                'updated_at': '2021-01-01',
+                'config_hash': 'abc',
+            },
+            # Deployed before config_hash existed
             {'namespace': 'test_namespace2', 'schedule_name': 'test2', 'updated_at': '2021-01-02'},
         ],
         **metadata,
@@ -625,33 +631,34 @@ def test_format_schedule_config(formatters):
     result = formatters.format_schedule_config(input_data)
 
     assert result == {
-        'test_namespace1': {'test1': '2021-01-01'},
-        'test_namespace2': {'test2': '2021-01-02'},
+        'test_namespace1': {'test1': {'updated_at': '2021-01-01', 'config_hash': 'abc'}},
+        'test_namespace2': {'test2': {'updated_at': '2021-01-02', 'config_hash': None}},
     }
 
 
 def test_create_schedule_config(formatters):
+    to_create = {'frequency': 60, 'data': {'test': 'test'}, 'updated_at': '2021-01-03'}
+    to_update = {'frequency': 60, 'data': {'test': 'test2'}, 'updated_at': '2021-01-04'}
+    unchanged = {'frequency': 60, 'data': {'test': 'test3'}, 'updated_at': '2021-01-01'}
     input_data = {
         'current_schedule_config': {
             'scouter': {
-                'test_schedule_name_to_delete': '2021-01-01',
-                'test_schedule_name_to_update': '2021-01-02',
+                'test_schedule_name_to_delete': {'updated_at': '2021-01-01', 'config_hash': 'x'},
+                'test_schedule_name_to_update': {
+                    'updated_at': '2021-01-02',
+                    'config_hash': schedule_config_hash(to_update),
+                },
+                'test_schedule_name_unchanged': {
+                    'updated_at': '2021-01-02',
+                    'config_hash': schedule_config_hash(unchanged),
+                },
             }
         },
         'schedule_config': {
-            'laborious': {
-                'test_schedule_name_to_create': {
-                    'frequency': 60,
-                    'data': {'test': 'test'},
-                    'updated_at': '2021-01-03',
-                }
-            },
+            'laborious': {'test_schedule_name_to_create': to_create},
             'scouter': {
-                'test_schedule_name_to_update': {
-                    'frequency': 60,
-                    'data': {'test': 'test2'},
-                    'updated_at': '2021-01-04',
-                }
+                'test_schedule_name_to_update': to_update,
+                'test_schedule_name_unchanged': unchanged,
             },
         },
     }
@@ -662,9 +669,8 @@ def test_create_schedule_config(formatters):
         'to_create': {
             'laborious': {
                 'test_schedule_name_to_create': {
-                    'frequency': 60,
-                    'data': {'test': 'test'},
-                    'updated_at': '2021-01-03',
+                    **to_create,
+                    'config_hash': schedule_config_hash(to_create),
                 }
             },
             'scouter': {},
@@ -672,9 +678,8 @@ def test_create_schedule_config(formatters):
         'to_update': {
             'scouter': {
                 'test_schedule_name_to_update': {
-                    'frequency': 60,
-                    'data': {'test': 'test2'},
-                    'updated_at': '2021-01-04',
+                    **to_update,
+                    'config_hash': schedule_config_hash(to_update),
                 }
             },
             'laborious': {},
@@ -683,6 +688,75 @@ def test_create_schedule_config(formatters):
     }
 
 
+def test_schedule_config_hash_ignores_updated_at_and_key_order():
+    schedule = {'frequency': 60, 'data': {'a': 1, 'b': 2}, 'updated_at': '2021-01-01'}
+    same = {'data': {'b': 2, 'a': 1}, 'updated_at': '2021-02-02', 'frequency': 60}
+
+    assert schedule_config_hash(schedule) == schedule_config_hash(same)
+    assert schedule_config_hash(schedule) == schedule_config_hash(
+        {**schedule, 'config_hash': 'stale'}
+    )
+    assert schedule_config_hash(schedule) != schedule_config_hash({**schedule, 'frequency': 30})
+
+
+def test_create_schedule_config_updates_on_config_hash_change(formatters):
+    """Same updated_at, different content (e.g. a pipeline edited by hand): redeploy"""
+    schedule = {'frequency': 60, 'data': {'test': 'new'}, 'updated_at': '2021-01-01'}
+
+    result = formatters.create_schedule_config(
+        {
+            'current_schedule_config': {
+                'scouter': {
+                    's': {
+                        'updated_at': '2021-01-02',
+                        'config_hash': schedule_config_hash({**schedule, 'data': {'test': 'old'}}),
+                    }
+                }
+            },
+            'schedule_config': {'scouter': {'s': schedule}, 'laborious': {}},
+        }
+    )
+
+    assert set(result['to_update']['scouter']) == {'s'}
+
+
+def test_create_schedule_config_updates_on_newer_updated_at(formatters):
+    """Bumping updated_at forces a redeploy even when nothing else changed"""
+    schedule = {'frequency': 60, 'updated_at': '2021-01-05'}
+
+    result = formatters.create_schedule_config(
+        {
+            'current_schedule_config': {
+                'scouter': {
+                    's': {'updated_at': '2021-01-02', 'config_hash': schedule_config_hash(schedule)}
+                }
+            },
+            'schedule_config': {'scouter': {'s': schedule}, 'laborious': {}},
+        }
+    )
+
+    assert set(result['to_update']['scouter']) == {'s'}
+
+
+def test_create_schedule_config_accepts_timestamp_only_current_schedules(formatters):
+    """
+    A run started before config_hash replays format_schedule_config's timestamp-only
+    result: with no stored hash, the schedule is redeployed once to record it.
+    """
+    schedule = {'frequency': 60, 'updated_at': '2021-01-01'}
+
+    result = formatters.create_schedule_config(
+        {
+            'current_schedule_config': {'scouter': {'s': '2021-01-02'}},
+            'schedule_config': {'scouter': {'s': schedule}, 'laborious': {}},
+        }
+    )
+
+    assert result['to_update']['scouter'] == {
+        's': {**schedule, 'config_hash': schedule_config_hash(schedule)}
+    }
+
+
 def test_create_slot_config(formatters):
     input_data = {
         'current_slot_config': {
@@ -1426,14 +1500,22 @@ def test_create_schedule_config_warns_on_invalid_chunk_period(formatters):
 
 def test_create_schedule_config_no_chunk_period_warning_when_unchanged(formatters):
     input_data = _drift_schedule_config('hour')
-    input_data['current_schedule_config'] = {'laborious': {'m1-drift-1m': '2026-10-27'}}
+    schedule = input_data['schedule_config']['laborious']['m1-drift-1m']
+    input_data['current_schedule_config'] = {
+        'laborious': {
+            'm1-drift-1m': {
+                'updated_at': '2026-10-27',
+                'config_hash': schedule_config_hash(schedule),
+            }
+        }
+    }
 
     formatters.create_schedule_config(input_data)
 
     formatters.send_notification.assert_not_called()
 
 
-# PI Web API through the tag database (connectors-pi-web-api)
+# Write tags through the tag database (SIENTIAPDE-1974)
 
 PLANT_CONNECTOR = {
     'connector_id': 'plant',
@@ -1551,60 +1633,71 @@ def test_process_schedules_notifies_unresolved_write_tags_once(formatters):
     assert formatters.send_notification.call_count == 3
 
 
-def test_process_schedules_updated_at_follows_the_tag_database(formatters):
-    """A tag changed after the pipeline redeploys the pipelines that reference it"""
-    batch = _batch_pipeline('batch', [{'tag_id': 77}], updated_at='2026-10-01 00:00:00.000+0000')
-    scouter = {
-        'workflow_type': 'scouter',
-        'schedule_name': 'sin',
-        'model_id': 'm1',
-        'model': {'name': 'model m1'},
-        'read_tags': [{'tag_id': 7, 'frequency': 1000}],
-        'updated_at': '2026-10-01 00:00:00.000+0000',
+def test_tag_database_changes_redeploy_through_config_hash(formatters):
+    """
+    A tag changed or deleted in the tag database (without bumping any updated_at)
+    changes the resolved schedule, so its config_hash, and redeploys the pipelines that
+    use it. A change that does not reach the schedule (e.g. a description) does not.
+    """
+    pipelines = [
+        _batch_pipeline('moved', [{'tag_id': 77}]),
+        _batch_pipeline('deleted', [{'tag_id': 78}]),
+        _batch_pipeline('same', [{'tag_id': 79}]),
+        {
+            'workflow_type': 'scouter',
+            'schedule_name': 'renamed',
+            'model_id': 'm1',
+            'model': {'name': 'model m1'},
+            'read_tags': [{'tag_id': 7, 'frequency': 1000}],
+            'updated_at': '2026-10-01 00:00:00.000+0000',
+        },
+    ]
+    before = [
+        _write_tag(77, 'ns=2;s=P1'),
+        _write_tag(78, 'ns=2;s=P2'),
+        _write_tag(79, 'ns=2;s=P3', description='old'),
+        _write_tag(7, 'ns=3;i=1001', category='Read'),
+    ]
+    after = [
+        _write_tag(77, 'ns=2;s=P1-moved'),
+        _write_tag(79, 'ns=2;s=P3', description='new'),
+        {**_write_tag(7, 'ns=3;i=1001', category='Read'), 'tag_name': 'renamed-tag'},
+    ]
+
+    def process(connector_tags):
+        return formatters.process_schedules(
+            {
+                'pipelines': pipelines,
+                'connectors': [PLANT_CONNECTOR],
+                'connector_tags': connector_tags,
+                **metadata,
+            }
+        )
+
+    deployed = process(before)
+    # Last orchestrated after every pipeline's updated_at, with the hash of `before`
+    current = {
+        namespace: {
+            name: {
+                'updated_at': '2026-10-05 00:00:00.000+0000',
+                'config_hash': schedule_config_hash(schedule),
+            }
+            for name, schedule in schedules.items()
+        }
+        for namespace, schedules in deployed.items()
     }
-    untouched = _batch_pipeline('old', [{'tag_id': 78}], updated_at='2026-10-20 00:00:00.000+0000')
 
-    schedule_config = formatters.process_schedules(
-        {
-            'pipelines': [batch, scouter, untouched],
-            'connectors': [PLANT_CONNECTOR],
-            'connector_tags': [
-                _write_tag(77, 'ns=2;s=P2', updated_at='2026-10-10 00:00:00.000+0000'),
-                # Deactivated: no longer usable, but its change still redeploys
-                _write_tag(
-                    7,
-                    'ns=3;i=1001',
-                    category='Read',
-                    status='Inactive',
-                    updated_at='2026-10-11 00:00:00.000+0000',
-                ),
-                _write_tag(78, 'ns=2;s=C', updated_at='2026-10-02 00:00:00.000+0000'),
-            ],
-            **metadata,
-        }
+    unchanged = formatters.create_schedule_config(
+        {'current_schedule_config': current, 'schedule_config': process(before), **metadata}
     )
-
-    assert schedule_config['laborious']['batch']['updated_at'] == '2026-10-10 00:00:00.000+0000'
-    assert schedule_config['scouter']['sin']['updated_at'] == '2026-10-11 00:00:00.000+0000'
-    assert schedule_config['laborious']['old']['updated_at'] == '2026-10-20 00:00:00.000+0000'
-
-    # Last orchestrated on 2026-10-05: only the pipelines whose tags changed since update
     actions = formatters.create_schedule_config(
-        {
-            'current_schedule_config': {
-                'scouter': {'sin': '2026-10-05 00:00:00.000+0000'},
-                'laborious': {
-                    'batch': '2026-10-05 00:00:00.000+0000',
-                    'old': '2026-10-21 00:00:00.000+0000',
-                },
-            },
-            'schedule_config': schedule_config,
-            **metadata,
-        }
+        {'current_schedule_config': current, 'schedule_config': process(after), **metadata}
     )
 
-    assert set(actions['to_update']['scouter']) == {'sin'}
-    assert set(actions['to_update']['laborious']) == {'batch'}
+    assert unchanged['to_update'] == {'scouter': {}, 'laborious': {}}
+    assert set(actions['to_update']['laborious']) == {'moved', 'deleted'}
+    assert set(actions['to_update']['scouter']) == {'renamed'}
+    assert actions['to_update']['laborious']['deleted']['opc_output_config'] == {}
 
 
 def test_process_schedules_without_updated_at_redeploys_every_run(formatters):
@@ -1662,7 +1755,12 @@ def test_format_schedule_config_groups_schedules_of_one_namespace(formatters):
         }
     )
 
-    assert result == {'laborious': {'a': '2021-01-01', 'b': '2021-01-02'}}
+    assert result == {
+        'laborious': {
+            'a': {'updated_at': '2021-01-01', 'config_hash': None},
+            'b': {'updated_at': '2021-01-02', 'config_hash': None},
+        }
+    }
 
 
 def test_report_schedule_orchestration_success_only_and_empty_types(formatters):

tests/orchestrator/activities/test_mongo_db.py

@@ -1,5 +1,5 @@
 from datetime import UTC, datetime
-from unittest.mock import ANY, AsyncMock, MagicMock, patch
+from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
 
 from pytest import fixture, mark
 from sientia_do.notifications.models import NotificationLevel
@@ -270,23 +270,27 @@ def test_aggregate_documents_in_mongodb_missing_aggregation(mongo_db):
 def test_update_pipelines_timestamps_success(now_mock, mongo_db):
     input_data = {
         'updated_pipelines': [
-            {'schedule_name': 'test1', 'namespace': 'test1', 'success': True},
+            {'schedule_name': 'test1', 'namespace': 'test1', 'success': True, 'config_hash': 'h1'},
+            # Updated by a run started before config_hash
             {'schedule_name': 'test2', 'namespace': 'test2', 'success': True},
         ]
     }
-    mongo_db.mongo_db_repository.update_many = MagicMock(return_value=MagicMock())
+    mongo_db.mongo_db_repository.update_one = MagicMock(return_value=1)
     mongo_db.update_pipelines_timestamps(input_data)
-    mongo_db.mongo_db_repository.update_many.assert_called_once_with(
-        'orchestrated_schedules',
-        {
-            '$or': [
-                {'schedule_name': 'test1', 'namespace': 'test1'},
-                {'schedule_name': 'test2', 'namespace': 'test2'},
-            ]
-        },
-        {'$set': {'updated_at': now_mock.return_value}},
-        {},
-    )
+    assert mongo_db.mongo_db_repository.update_one.call_args_list == [
+        call(
+            'orchestrated_schedules',
+            {'schedule_name': 'test1', 'namespace': 'test1'},
+            {'$set': {'updated_at': now_mock.return_value, 'config_hash': 'h1'}},
+            {},
+        ),
+        call(
+            'orchestrated_schedules',
+            {'schedule_name': 'test2', 'namespace': 'test2'},
+            {'$set': {'updated_at': now_mock.return_value, 'config_hash': None}},
+            {},
+        ),
+    ]
 
 
 @patch('orchestrator.activities.mongo_db.now')
@@ -295,31 +299,30 @@ def test_update_pipelines_timestamps_success(now_mock, mongo_db):
 )
 def test_update_pipelines_timestamps_all_failed_stamps_nothing(now_mock, mongo_db, pipelines):
     input_data = {'updated_pipelines': pipelines}
-    mongo_db.mongo_db_repository.update_many = MagicMock()
+    mongo_db.mongo_db_repository.update_one = MagicMock()
 
     mongo_db.update_pipelines_timestamps(input_data)
 
-    # An empty filter would match and stamp every tracked schedule, so the failed one
-    # would never be retried.
-    mongo_db.mongo_db_repository.update_many.assert_not_called()
+    # A failed update is not stamped, so it is retried on the next tick
+    mongo_db.mongo_db_repository.update_one.assert_not_called()
 
 
 @patch('orchestrator.activities.mongo_db.now')
 def test_update_pipelines_timestamps_mixed_stamps_only_successes(now_mock, mongo_db):
     input_data = {
         'updated_pipelines': [
-            {'schedule_name': 'test1', 'namespace': 'test1', 'success': True},
-            {'schedule_name': 'test2', 'namespace': 'test2', 'success': False},
+            {'schedule_name': 'test1', 'namespace': 'test1', 'success': True, 'config_hash': 'h1'},
+            {'schedule_name': 'test2', 'namespace': 'test2', 'success': False, 'config_hash': 'h2'},
         ]
     }
-    mongo_db.mongo_db_repository.update_many = MagicMock()
+    mongo_db.mongo_db_repository.update_one = MagicMock()
 
     mongo_db.update_pipelines_timestamps(input_data)
 
-    mongo_db.mongo_db_repository.update_many.assert_called_once_with(
+    mongo_db.mongo_db_repository.update_one.assert_called_once_with(
         'orchestrated_schedules',
-        {'$or': [{'schedule_name': 'test1', 'namespace': 'test1'}]},
-        {'$set': {'updated_at': now_mock.return_value}},
+        {'schedule_name': 'test1', 'namespace': 'test1'},
+        {'$set': {'updated_at': now_mock.return_value, 'config_hash': 'h1'}},
         {},
     )
 
@@ -334,7 +337,7 @@ def test_update_pipelines_timestamps_failure(now_mock, mongo_db):
         **metadata,
     }
 
-    mongo_db.mongo_db_repository.update_many.side_effect = Exception('Error')
+    mongo_db.mongo_db_repository.update_one.side_effect = Exception('Error')
     try:
         mongo_db.update_pipelines_timestamps(input_data)
     except Exception as e:
@@ -356,7 +359,7 @@ def test_update_pipelines_timestamps_failure(now_mock, mongo_db):
 def test_create_pipelines_timestamps_success(now_mock, mongo_db):
     input_data = {
         'created_pipelines': [
-            {'schedule_name': 'test1', 'namespace': 'test1', 'success': True},
+            {'schedule_name': 'test1', 'namespace': 'test1', 'success': True, 'config_hash': 'h1'},
             {'schedule_name': 'test2', 'namespace': 'test2', 'success': True},
         ]
     }
@@ -365,8 +368,18 @@ def test_create_pipelines_timestamps_success(now_mock, mongo_db):
     mongo_db.mongo_db_repository.insert_many.assert_called_once_with(
         'orchestrated_schedules',
         [
-            {'schedule_name': 'test1', 'namespace': 'test1', 'updated_at': now_mock.return_value},
-            {'schedule_name': 'test2', 'namespace': 'test2', 'updated_at': now_mock.return_value},
+            {
+                'schedule_name': 'test1',
+                'namespace': 'test1',
+                'updated_at': now_mock.return_value,
+                'config_hash': 'h1',
+            },
+            {
+                'schedule_name': 'test2',
+                'namespace': 'test2',
+                'updated_at': now_mock.return_value,
+                'config_hash': None,
+            },
         ],
         {},
     )

tests/orchestrator/activities/test_temporal_manager.py

@@ -145,6 +145,7 @@ async def test_create_schedule(
                     'data': {'test': 'test'},
                     'execution_timeout_seconds': 100,
                     'task_timeout_seconds': 100,
+                    'config_hash': 'h1',
                 },
                 'test-schedule-invalid-frequency': {
                     'model_id': 2,
@@ -291,24 +292,29 @@ async def test_create_schedule(
         ]
     )
 
+    # config_hash goes to the report (orchestrated_schedules), not to the workflow
+    assert 'config_hash' not in input_data['schedules']['scouter']['test-schedule']
     assert report == [
         {
             'schedule_name': 'test-schedule',
             'namespace': 'scouter',
             'success': True,
             'message': 'Schedule created successfully',
+            'config_hash': 'h1',
         },
         {
             'schedule_name': 'test-schedule-invalid-frequency',
             'namespace': 'scouter',
             'success': False,
             'message': 'Invalid frequency',
+            'config_hash': None,
         },
         {
             'schedule_name': 'test-schedule-laborious',
             'namespace': 'laborious',
             'success': True,
             'message': 'Schedule created successfully',
+            'config_hash': None,
         },
     ]
 
@@ -355,6 +361,7 @@ async def test_update_schedules(
                     'workflow_type': 'scouter',
                     'frequency': '1m',
                     'data': {'test': 'test'},
+                    'config_hash': 'h1',
                 },
                 'test-schedule_no_handler': {
                     'workflow_type': 'scouter',
@@ -400,24 +407,28 @@ async def test_update_schedules(
     handler_scouter.update.assert_called_once()
     handler_laborious.update.assert_called_once()
 
+    assert 'config_hash' not in input_data['schedules']['scouter']['test-schedule']
     assert report == [
         {
             'schedule_name': 'test-schedule',
             'namespace': 'scouter',
             'success': True,
             'message': 'Schedule updated successfully',
+            'config_hash': 'h1',
         },
         {
             'schedule_name': 'test-schedule_no_handler',
             'namespace': 'scouter',
             'success': False,
             'message': 'Schedule test-schedule_no_handler not found',
+            'config_hash': None,
         },
         {
             'schedule_name': 'test-schedule-laborious',
             'namespace': 'laborious',
             'success': True,
             'message': 'Schedule updated successfully',
+            'config_hash': None,
         },
     ]
 

tests/orchestrator/utils/test_tag_database.py

@@ -4,11 +4,9 @@ from orchestrator.utils.tag_database import (
     index_connector_tags,
     index_connectors,
     index_tag_documents,
-    latest_updated_at,
     referenced_tag_ids,
     resolve_tag_names,
     resolve_write_tags,
-    tag_documents_updated_at,
     tag_id_text,
 )
 from tests.orchestrator.utils.tag_database_fixtures import (
@@ -216,34 +214,3 @@ def test_resolve_write_tags_issues():
         "{'tag_id': None}",
         'ns=2;s=raw',
     }
-
-
-def test_latest_updated_at():
-    assert latest_updated_at(
-        '2026-10-01 00:00:00.000+0000', None, '', '2026-10-02 00:00:00.000+0000'
-    ) == ('2026-10-02 00:00:00.000+0000')
-    assert latest_updated_at(None, '') is None
-
-
-def test_tag_documents_updated_at():
-    documents = index_tag_documents(
-        [
-            write_tag_doc(
-                7, 'ns=2;s=R', category='Read', updated_at='2026-10-05 00:00:00.000+0000'
-            ),
-            # Inactive documents count: deactivating a tag redeploys its pipelines
-            write_tag_doc(
-                77, 'ns=2;s=P', status='Inactive', updated_at='2026-10-09 00:00:00.000+0000'
-            ),
-            write_tag_doc(78, 'ns=2;s=C', updated_at='2026-10-03 00:00:00.000+0000'),
-        ]
-    )
-
-    scouter_pipeline = scouter_pipeline_doc(
-        'sin', 'm1', [{'tag_id': 7}, legacy_tag('ns=2;s=A', 'a')]
-    )
-    batch_pipeline = batch_pipeline_doc('pred', [{'tag_id': '77'}, {'tag_id': 78}, {'tag_id': 404}])
-
-    assert tag_documents_updated_at(scouter_pipeline, documents) == '2026-10-05 00:00:00.000+0000'
-    assert tag_documents_updated_at(batch_pipeline, documents) == '2026-10-09 00:00:00.000+0000'
-    assert tag_documents_updated_at(batch_pipeline_doc('none', []), documents) is None

tests/orchestrator/workflows/test_orchestrator.py

@@ -125,11 +125,7 @@ async def test_run(workflow_mock, referenced_tag_ids_mock, orchestrator):
         [
             call(
                 Activities.aggregate_documents_in_mongodb,
-                {
-                    **metadata,
-                    'query': DEFAULT_CONNECTORS_QUERY,
-                    'timestamp_fields': ['updated_at'],
-                },
+                {**metadata, 'query': DEFAULT_CONNECTORS_QUERY},
                 retry_policy=ANY,
                 start_to_close_timeout=ANY,
             ),
@@ -144,7 +140,6 @@ async def test_run(workflow_mock, referenced_tag_ids_mock, orchestrator):
                             *DEFAULT_CONNECTOR_TAGS_QUERY['aggregation'],
                         ],
                     },
-                    'timestamp_fields': ['updated_at'],
                 },
                 retry_policy=ANY,
                 start_to_close_timeout=ANY,
@@ -508,12 +503,16 @@ def _process_schedules_input(workflow_mock):
 
 
 def test_default_connector_tags_query_loads_every_category_and_status():
-    """The status / category filters are applied in the code, so changes still redeploy"""
+    """
+    The status / category filters are applied in the code. updated_at is not loaded: a
+    tag change reaches the schedules through their config_hash.
+    """
     stages = DEFAULT_CONNECTOR_TAGS_QUERY['aggregation']
     assert all('$match' not in stage for stage in stages)
     projected = stages[-1]['$project']
-    for field in ('category', 'data_type', 'status', 'updated_at'):
+    for field in ('category', 'data_type', 'status'):
         assert projected[field] == 1
+    assert 'updated_at' not in projected
 
 
 @mark.asyncio
@@ -535,8 +534,6 @@ async def test_run_hands_the_tag_database_to_process_schedules(workflow_mock, _i
     assert schedules_input['connectors'] is loaded
     assert schedules_input['connector_tags'] is loaded
     assert 'pi_connectors' not in schedules_input
-    for c in _tag_database_calls(workflow_mock):
-        assert c.args[1]['timestamp_fields'] == ['updated_at']
 
 
 @mark.asyncio