style: fix ruff formatting in test_mongo_db.py

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

.github/workflows/quality-gate.yml

@@ -1,16 +1,31 @@
 name: Quality gate
 
 on:
+  push:
+    branches:
+      - main
+      - 'release/**'
+      - 'feature/**'
   pull_request:
     branches:
       - main
+      - 'release/**'
+      - 'feature/**'
     types: [ opened, synchronize, reopened ]
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
 jobs:
   quality-gate:
-    uses: Aignosi/github_workflow_templates/.github/workflows/python-module-quality-gate.yml@main
-    permissions: write-all
+    uses: Aignosi/github_workflow_templates/.github/workflows/dataops-module-quality-gate.yml@main
+    permissions:
+      contents: read
+      pull-requests: write
+      issues: write
     with:
       project_name: 'orchestrator'
       repositories: 'sientia-dataops-library'
+      requirements_file: 'requirements-local.txt'
     secrets: inherit

.gitignore

@@ -27,6 +27,7 @@ __pycache__/
 .vscode/
 .pytest_cache/
 .idea/
+.obsidian/
 *.swp
 
 # Ignorar arquivos temporários

CLAUDE.md

@@ -0,0 +1,65 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## Project
+
+SIENTIA DataOps Orchestrator — a Temporal.io workflow application for pipeline orchestration, real-time alerting, and scheduled reporting on the SIENTIA platform. Python 3.11+, package name `orchestrator`.
+
+## Commands
+
+Install deps:
+```bash
+pip install -r requirements.txt
+pip install -r requirements-dev.txt   # ruff, mypy, bandit, pytest, pytest-cov, pytest-asyncio
+```
+
+Run the worker locally (activates venv, loads `.env`, runs `python -m orchestrator.worker.worker`):
+```bash
+./run_local.sh
+```
+
+Tests:
+```bash
+pytest tests/ --cov=orchestrator --cov-report=html   # full suite with coverage
+pytest tests/orchestrator/activities/test_mongo_db.py  # single file
+pytest tests/orchestrator/activities/test_mongo_db.py::TestClass::test_name  # single test
+./run_coverage.sh   # runs pytest w/ coverage, opens htmlcov/index.html
+```
+
+Lint/format/type/security (run individually — `validate.sh` referenced by README no longer exists in the repo, so don't rely on it):
+```bash
+ruff format --check orchestrator/ tests/
+ruff check orchestrator/ tests/
+mypy orchestrator/
+bandit -r orchestrator/ -ll
+
+# autofix
+ruff format orchestrator/ tests/
+ruff check --fix orchestrator/ tests/
+```
+
+CI quality gate (`.github/workflows/quality-gate.yml`) calls the shared reusable workflow `Aignosi/github_workflow_templates/.github/workflows/dataops-module-quality-gate.yml` on PRs to `main` — it's the source of truth for what must pass, not a local script.
+
+## Architecture
+
+### 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/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.
+
+### 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.
+- Workflows are registered with `@workflow.defn(name='...')`; the `run` method is `@workflow.run`. Activity calls from within a workflow go through `workflow.start_local_activity_method` / `workflow.execute_activity_method` referencing `Activities.<method>` (the class method, not an instance) with a `retry_policy` from `sientia_do.temporal.policies`.
+- Task queue isolation is deliberate: don't move activities between the orchestrator/alerts/reports worker activity lists without checking which queue actually needs them — each list in `worker.py` is scoped to what that workflow (+ its subworkflows) calls.
+- Namespaces: Temporal schedules span two namespaces, `scouter` and `laborious` (`TEMPORAL_SCOUTER_NAMESPACE` / `TEMPORAL_LABORIOUS_NAMESPACE`), handled inside `TemporalManager`.
+
+### External dependency
+`sientia-dataops-library` (pinned via `requirements.txt` as a git dependency, currently `@1.8.2`) supplies `sientia_do.observability.logger`, `sientia_do.observability.metrics_controller`, `sientia_do.notifications.handlers`, `sientia_do.temporal.policies`, and `sientia_do.temporal.activities.postgres.Postgres`. It is a separate Aignosi repo — its version bump is a deliberate, explicit change, not incidental.
+
+### Tests
+`tests/` mirrors `orchestrator/` (`tests/orchestrator/activities/`, `tests/orchestrator/utils/`, `tests/orchestrator/workflows/...`). Async tests use `pytest-asyncio`; coverage targets the `orchestrator` package. When testing Temporal workflows, follow the existing pattern in `tests/orchestrator/workflows/test_orchestrator.py` / `test_alerts.py` / `test_reports.py` (using Temporal's test env / activity mocking) rather than hitting real Temporal/Mongo/Redis/Postgres.
+
+### Ruff config notes (`pyproject.toml`)
+Single quotes enforced by formatter. Notably ignored lint rules with reasons already documented inline: `B023` (closures over loop vars needed for schedule-action assembly), `BLE001` (blind `except` needed so activities can always notify on any error), `N802`/`N806` (Temporal decorators/variables don't follow standard naming), `S105`/`S106` (false-positive hardcoded-password matches).

orchestrator/activities/mongo_db.py

@@ -2,7 +2,7 @@ from temporalio import activity, workflow
 
 with workflow.unsafe.imports_passed_through():
     import traceback
-    from datetime import UTC
+    from datetime import UTC, datetime
     from logging import Logger
     from typing import Any
 
@@ -452,7 +452,8 @@ class MongoDB(SientiaMonitoring):
                 Required fields:
                 - metadata (dict[str, Any]): Workflow execution metadata
                 - collection_name (str): Name of the MongoDB collection
-                - last_data_timestamp (str | None): Last processed timestamp for filtering
+                - last_data_timestamp (str | None): Last processed timestamp
+                  (ISO-format string) for filtering
                 - base_data_filter (dict[str, Any]): Base query filter conditions
 
         Returns:
@@ -472,19 +473,14 @@ class MongoDB(SientiaMonitoring):
             if last_data_timestamp is None:
                 data_filter = base_data_filter
             else:
-                # ``notification_queue.timestamp`` is stored as a string in
-                # ``DATETIME_FORMAT_WITH_TZ`` (``Notification`` writes it as
-                # ``now().strftime(DATETIME_FORMAT_WITH_TZ)``). Coercing
-                # ``last_data_timestamp`` to ``datetime`` here would force a
-                # BSON ``String`` vs ``Date`` comparison, which always yields
-                # ``False`` (``String < Date`` in BSON sort order) and breaks
-                # incremental loading entirely. Comparing strings preserves the
-                # intended chronological filter because the format is
-                # lexicographically ordered when the timezone is fixed
-                # (``Notification.timestamp`` always uses UTC).
+                # ``notification_queue.timestamp`` is stored as a native BSON
+                # ``Date`` (``Notification.timestamp`` is a ``datetime``).
+                # ``last_data_timestamp`` arrives here as an ISO-format string
+                # (round-tripped through Redis), so it must be parsed back to
+                # ``datetime`` for the ``$gt`` comparison to be type-correct.
                 data_filter = {
                     **base_data_filter,
-                    'timestamp': {'$gt': last_data_timestamp},
+                    'timestamp': {'$gt': datetime.fromisoformat(last_data_timestamp)},
                 }
 
             self.debug(f'Data filter: {data_filter}', metadata=metadata)

orchestrator/activities/slot_manager.py

@@ -335,6 +335,11 @@ class SlotManager(SientiaMonitoring):
 
         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 an ISO-format string instead.
+        last_data_timestamp = last_data_timestamp.isoformat()
+
         try:
             self.redis_repository.set(key, last_data_timestamp, ttl=60 * 60 * 5)
         except Exception as e:

orchestrator/activities/temporal_manager.py

@@ -27,6 +27,7 @@ with workflow.unsafe.imports_passed_through():
 
 RUNTIME_WORKFLOWS = ['predictions_batch', 'minimal_retrain', 'drift', 'simple_metrics']
 
+
 class TemporalManager(SientiaMonitoring):
     """
     Temporal workflow and schedule management activity.
@@ -215,9 +216,8 @@ class TemporalManager(SientiaMonitoring):
                     self.debug(
                         f'{json.dumps(schedule, indent=4, sort_keys=True)}', metadata=metadata
                     )
-                    
-                    runtime_name = schedule.get('runtime', 'legacy') if workflow_type in RUNTIME_WORKFLOWS else None
-                    task_queue_name = build_queue_name(workflow_type, runtime_name)
+
+                    task_queue_name = self._build_task_queue_name(workflow_type, schedule)
                     schedule['task_queue'] = task_queue_name
 
                     await client.create_schedule(
@@ -280,6 +280,68 @@ class TemporalManager(SientiaMonitoring):
 
         return report
 
+    @staticmethod
+    def _build_task_queue_name(workflow_type: str, schedule: dict[str, Any]) -> str:
+        """
+        Build the Temporal task queue name for a schedule.
+
+        Only workflow types listed in ``RUNTIME_WORKFLOWS`` get an environment/tenant
+        specific ``runtime`` suffix; every other workflow type gets a plain queue name.
+        """
+        runtime_name = (
+            schedule.get('runtime', 'legacy') if workflow_type in RUNTIME_WORKFLOWS else None
+        )
+        return build_queue_name(workflow_type, runtime_name)
+
+    def _make_schedule_updater(self, schedule: dict[str, Any], metadata: dict[str, Any]):
+        """Build the ``ScheduleUpdate`` callback used by ``handler.update`` for one schedule."""
+
+        # fmt: off
+        async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate:  # NOSONAR
+            schedule_action = input_data.description.schedule.action
+
+            self.debug("Updating schedule:", metadata=metadata)
+
+            if hasattr(schedule_action, "args"):
+                self.debug("New schedule:", metadata=metadata)
+                self.debug(
+                    f"{json.dumps(schedule, indent=4, sort_keys=True)}", metadata=metadata)  # NOSONAR
+
+                schedule_action.args = [schedule]
+
+            input_data.description.schedule.spec.intervals = [
+                ScheduleIntervalSpec(
+                    every=timedelta(
+                        seconds=parse_frequency(schedule.get('frequency', '1m'))),
+                    offset=timedelta(
+                        seconds=parse_frequency(schedule.get('offset', '0m'))),
+                )
+            ]
+
+            return ScheduleUpdate(schedule=input_data.description.schedule)
+
+        # fmt: on
+        return update_schedule
+
+    async def _update_single_schedule(
+        self,
+        client: Client,
+        schedule_name: str,
+        schedule: dict[str, Any],
+        metadata: dict[str, Any],
+    ) -> None:
+        """Update a single schedule in Temporal, raising if the schedule handle is missing."""
+        handler = client.get_schedule_handle(schedule_name)
+
+        if not handler:
+            raise ValueError(f'Schedule {schedule_name} not found')
+
+        workflow_type = schedule['workflow_type']
+        schedule['task_queue'] = self._build_task_queue_name(workflow_type, schedule)
+
+        update_schedule = self._make_schedule_updater(schedule, metadata)
+        await handler.update(update_schedule)
+
     @activity.defn(name='update_schedules')
     async def update_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
         """
@@ -316,43 +378,7 @@ class TemporalManager(SientiaMonitoring):
 
             for schedule_name, schedule in schedules.items():
                 try:
-                    handler = client.get_schedule_handle(schedule_name)
-
-                    if not handler:
-                        raise ValueError(f'Schedule {schedule_name} not found')
-
-                    workflow_type = schedule['workflow_type']
-                    runtime_name = schedule.get('runtime', 'legacy') if workflow_type in RUNTIME_WORKFLOWS else None
-                    schedule['task_queue'] = build_queue_name(workflow_type, runtime_name)
-
-                    # fmt: off
-                    async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate: # NOSONAR
-                        schedule_action = input_data.description.schedule.action
-
-                        self.debug("Updating schedule:", metadata=metadata)
-
-                        if hasattr(schedule_action, "args"):
-                            self.debug("New schedule:", metadata=metadata)
-                            self.debug(
-                                f"{json.dumps(schedule, indent=4, sort_keys=True)}", metadata=metadata)  # NOSONAR
-
-                            schedule_action.args = [schedule]
-
-                        input_data.description.schedule.spec.intervals = [
-                            ScheduleIntervalSpec(
-                                every=timedelta(
-                                    seconds=parse_frequency(schedule.get('frequency', '1m'))),
-                                offset=timedelta(
-                                    seconds=parse_frequency(schedule.get('offset', '0m'))),
-                            )
-                        ]
-
-                        return ScheduleUpdate(schedule=input_data.description.schedule)
-
-                    # fmt: on
-                    await handler.update(update_schedule)
-
-                    del update_schedule
+                    await self._update_single_schedule(client, schedule_name, schedule, metadata)
 
                     report.append(
                         {

pipelines_sample.json

@@ -0,0 +1,43 @@
+{
+  "schedule_name": "scouter-test_linreg",
+  "workflow_type": "scouter",
+  "active": true,
+  "model_id": "1",
+  "frequency": "1m",
+  "offset": "0m",
+  "max_retry_policy": 1,
+  "execution_timeout_seconds": 300,
+  "task_timeout_seconds": 300,
+  "on_conflict": "error",
+  "debug_data_package": false,
+  "fill_missing_tags": false,
+  "filters": [
+    { "filter_name": "OUT_OF_RANGE", "policy": "STOP" }
+  ],
+  "read_tags": [
+    {
+      "server_id": "1",
+      "tag_name": "Counter",
+      "tag_address": "ns=2;s=GatewayProSYS.ProSYS.Simulation.Counter",
+      "aggr_func": "lts",
+      "data_range": [1, 30],
+      "frequency": 30000
+    },
+    {
+      "server_id": "1",
+      "tag_name": "Square",
+      "tag_address": "ns=2;s=GatewayProSYS.ProSYS.Simulation.Square",
+      "aggr_func": "lts",
+      "data_range": [-2, 2],
+      "frequency": 30000
+    },
+    {
+      "server_id": "1",
+      "tag_name": "Rollout",
+      "tag_address": "ns=2;s=GatewayProSYS.ProSYS.Simulation.Rollout",
+      "aggr_func": "lts",
+      "data_range": [-2, 2],
+      "frequency": 30000
+    }
+  ]
+}

requirements-local.txt

@@ -4,5 +4,5 @@ sqlalchemy
 redis
 pymongo
 jinja2
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.1
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2
 prometheus-client

requirements.txt

@@ -4,5 +4,5 @@ sqlalchemy
 redis
 pymongo
 jinja2
-sientia_do>=1.12.1
+sientia_do>=1.12.2
 prometheus-client

sonar-project.properties

@@ -1,4 +1,4 @@
-sonar.projectKey=Aignosi_sientia-dataops-orchestrator_temporal_099ace99-66d1-4df4-a557-10fde4f9d2f8
+sonar.projectKey=Aignosi_sientia-dataops-orchestrator_temporal_22f04ada-022d-4345-a74f-bbda07dc17a6
 sonar.projectName=sientia-dataops-orchestrator_temporal
 sonar.sources=orchestrator
 sonar.tests=tests
@@ -7,5 +7,4 @@ sonar.qualitygate.timeout=300
 sonar.python.coverage.reportPaths=coverage.xml
 sonar.python.xunit.reportPath=pytest.xml
 sonar.python.version=3.11
-sonar.projectVersion=1.0.0
 sonar.coverage.exclusions=orchestrator/worker/*

tests/orchestrator/activities/test_mongo_db.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import UTC, datetime
 from unittest.mock import ANY, AsyncMock, MagicMock, patch
 
 from pytest import fixture
@@ -462,7 +462,7 @@ def test_load_latest_data_none_last_data_timestamp(mongo_db):
             {
                 'name': 'test1',
                 'value': 1,
-                'timestamp': '2023-01-01 12:00:00+0000',
+                'timestamp': datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC),
             }
         ]
     )
@@ -482,18 +482,20 @@ def test_load_latest_data_none_last_data_timestamp(mongo_db):
         {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
     )
 
-    assert result == [{'name': 'test1', 'value': 1, 'timestamp': '2023-01-01 12:00:00+0000'}]
+    assert result == [
+        {'name': 'test1', 'value': 1, 'timestamp': datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)}
+    ]
 
 
 def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
-    """Test load_latest_data"""
+    """Test load_latest_data compares native datetime against native datetime in $gt"""
 
     mongo_db.mongo_db_repository.find = MagicMock(
         return_value=[
             {
                 'name': 'test1',
                 'value': 1,
-                'timestamp': '2023-01-01 12:00:00+0000',
+                'timestamp': datetime(2023, 1, 1, 12, 0, 1, tzinfo=UTC),
             }
         ]
     )
@@ -502,7 +504,7 @@ def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
         {
             'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
             'collection_name': 'test_collection',
-            'last_data_timestamp': '2023-01-01 12:00:00+0000',
+            'last_data_timestamp': '2023-01-01T12:00:00+00:00',
             'base_data_filter': {'level': 'ERROR'},
         }
     )
@@ -511,12 +513,60 @@ def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
         'test_collection',
         {
             'level': 'ERROR',
-            'timestamp': {'$gt': '2023-01-01 12:00:00+0000'},
+            'timestamp': {'$gt': datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)},
         },
         {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
     )
 
-    assert result == [{'name': 'test1', 'value': 1, 'timestamp': '2023-01-01 12:00:00+0000'}]
+    assert result == [
+        {'name': 'test1', 'value': 1, 'timestamp': datetime(2023, 1, 1, 12, 0, 1, tzinfo=UTC)}
+    ]
+
+
+def test_load_latest_data_run_boundary_no_skip_or_duplicate(mongo_db):
+    """Regression: the $gt filter built from the previous run's timestamp must not
+    skip the document that landed exactly on the boundary, nor re-return it."""
+
+    boundary_timestamp = datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)
+
+    def fake_find(collection_name, data_filter, metadata):
+        gt = data_filter.get('timestamp', {}).get('$gt')
+        all_docs = [
+            {
+                'name': 'before',
+                'value': 1,
+                'timestamp': datetime(2023, 1, 1, 11, 59, 59, tzinfo=UTC),
+            },
+            {'name': 'boundary', 'value': 2, 'timestamp': boundary_timestamp},
+            {'name': 'after', 'value': 3, 'timestamp': datetime(2023, 1, 1, 12, 0, 1, tzinfo=UTC)},
+        ]
+        if gt is None:
+            return all_docs
+        return [doc for doc in all_docs if doc['timestamp'] > gt]
+
+    mongo_db.mongo_db_repository.find = MagicMock(side_effect=fake_find)
+
+    first_run = mongo_db.load_latest_data(
+        {
+            'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
+            'collection_name': 'test_collection',
+            'last_data_timestamp': None,
+            'base_data_filter': {'level': 'ERROR'},
+        }
+    )
+
+    assert [doc['name'] for doc in first_run] == ['before', 'boundary', 'after']
+
+    second_run = mongo_db.load_latest_data(
+        {
+            'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
+            'collection_name': 'test_collection',
+            'last_data_timestamp': boundary_timestamp.isoformat(),
+            'base_data_filter': {'level': 'ERROR'},
+        }
+    )
+
+    assert [doc['name'] for doc in second_run] == ['after']
 
 
 def test_load_latest_data_error(mongo_db):
@@ -528,7 +578,7 @@ def test_load_latest_data_error(mongo_db):
             {
                 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
                 'collection_name': 'test_collection',
-                'last_data_timestamp': '2023-01-01 12:00:00+0000',
+                'last_data_timestamp': '2023-01-01T12:00:00+00:00',
                 'base_data_filter': {'level': 'ERROR'},
             }
         )

tests/orchestrator/activities/test_slot_manager.py

@@ -1,4 +1,4 @@
-from datetime import timedelta
+from datetime import UTC, datetime, timedelta
 from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
 
 from pandas import DataFrame
@@ -279,7 +279,10 @@ def test_put_last_data_timestamp_not_empty_dataframe(slot_manager):
         {
             'name': ['sensor1', 'sensor2'],
             'value': [25.5, 30.0],
-            'timestamp': ['2023-01-01 12:00:00', '2023-01-01 12:00:01'],
+            'timestamp': [
+                datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC),
+                datetime(2023, 1, 1, 12, 0, 1, tzinfo=UTC),
+            ],
         }
     )
     test_data = {
@@ -294,10 +297,10 @@ def test_put_last_data_timestamp_not_empty_dataframe(slot_manager):
 
     result = slot_manager.put_last_data_timestamp(test_data)
 
-    assert result == '2023-01-01 12:00:01'
+    assert result == '2023-01-01T12:00:01+00:00'
 
     slot_manager.redis_repository.set.assert_called_once_with(
-        'notification_last_timestamp:test_mail_type', '2023-01-01 12:00:01', ttl=18000
+        'notification_last_timestamp:test_mail_type', '2023-01-01T12:00:01+00:00', ttl=18000
     )
 
 
@@ -311,7 +314,7 @@ def test_put_last_data_timestamp_error(slot_manager):
             {
                 'name': ['sensor1', 'sensor2'],
                 'value': [25.5, 30.0],
-                'timestamp': ['2023-01-01 12:00:00'] * 2,
+                'timestamp': [datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)] * 2,
             }
         ).to_dict('records'),
         'mail_type': 'test_mail_type',

tests/orchestrator/activities/test_temporal_manager.py

@@ -206,7 +206,7 @@ async def test_create_schedule(
                 'test-workflow',
                 input_data['schedules']['scouter']['test-schedule'],
                 id='test-schedule',
-                task_queue='test-workflow-legacy-queue',
+                task_queue='test-workflow-queue',
                 execution_timeout=timedelta(seconds=100),
                 run_timeout=timedelta(seconds=100),
                 task_timeout=timedelta(seconds=100),
@@ -216,7 +216,7 @@ async def test_create_schedule(
                 'test-workflow',
                 input_data['schedules']['scouter']['test-schedule-invalid-frequency'],
                 id='test-schedule-invalid-frequency',
-                task_queue='test-workflow-legacy-queue',
+                task_queue='test-workflow-queue',
                 execution_timeout=timedelta(seconds=400),
                 run_timeout=timedelta(seconds=400),
                 task_timeout=timedelta(seconds=400),
@@ -226,7 +226,7 @@ async def test_create_schedule(
                 'test-workflow',
                 input_data['schedules']['laborious']['test-schedule-laborious'],
                 id='test-schedule-laborious',
-                task_queue='test-workflow-legacy-queue',
+                task_queue='test-workflow-queue',
                 execution_timeout=timedelta(seconds=500),
                 run_timeout=timedelta(seconds=500),
                 task_timeout=timedelta(seconds=500),
@@ -350,10 +350,24 @@ async def test_update_schedules(
     input_data = {
         'schedules': {
             'scouter': {
-                'test-schedule': {'frequency': '1m', 'data': {'test': 'test'}},
-                'test-schedule_no_handler': {'frequency': '1m', 'data': {'test': 'test'}},
+                'test-schedule': {
+                    'workflow_type': 'scouter',
+                    'frequency': '1m',
+                    'data': {'test': 'test'},
+                },
+                'test-schedule_no_handler': {
+                    'workflow_type': 'scouter',
+                    'frequency': '1m',
+                    'data': {'test': 'test'},
+                },
+            },
+            'laborious': {
+                'test-schedule-laborious': {
+                    'workflow_type': 'laborious',
+                    'frequency': '2m',
+                    'data': {'test': 'test'},
+                }
             },
-            'laborious': {'test-schedule-laborious': {'frequency': '2m', 'data': {'test': 'test'}}},
         }
     }
 
@@ -490,6 +504,91 @@ async def test_delete_schedules_with_no_client(temporal_manager):
         )
 
 
+def test_build_task_queue_name_non_runtime_workflow(temporal_manager):
+    task_queue_name = temporal_manager._build_task_queue_name(
+        'test-workflow', {'workflow_type': 'test-workflow'}
+    )
+
+    assert task_queue_name == 'test-workflow-queue'
+
+
+def test_build_task_queue_name_default_runtime_legacy(temporal_manager):
+    task_queue_name = temporal_manager._build_task_queue_name('drift', {'workflow_type': 'drift'})
+
+    assert task_queue_name == 'drift-legacy-queue'
+
+
+def test_build_task_queue_name_explicit_runtime(temporal_manager):
+    task_queue_name = temporal_manager._build_task_queue_name(
+        'drift', {'workflow_type': 'drift', 'runtime': 'tenant-x'}
+    )
+
+    assert task_queue_name == 'drift-tenant-x-queue'
+
+
+@mark.asyncio
+@patch('orchestrator.activities.temporal_manager.parse_frequency', side_effect=parse_frequency)
+@patch('orchestrator.activities.temporal_manager.ScheduleIntervalSpec')
+async def test_make_schedule_updater_patches_args_and_intervals(
+    _mock_schedule_interval_spec, _mock_parse_frequency, temporal_manager
+):
+    schedule = {'workflow_type': 'scouter', 'frequency': '2m', 'offset': '1h', 'data': 'new'}
+    update_schedule = temporal_manager._make_schedule_updater(schedule, metadata)
+
+    schedule_action = MagicMock()
+    input_data = MagicMock()
+    input_data.description.schedule.action = schedule_action
+
+    result = await update_schedule(input_data)
+
+    assert schedule_action.args == [schedule]
+    assert input_data.description.schedule.spec.intervals == [
+        _mock_schedule_interval_spec.return_value
+    ]
+    _mock_schedule_interval_spec.assert_called_once_with(
+        every=timedelta(seconds=120), offset=timedelta(seconds=3600)
+    )
+    assert result.schedule == input_data.description.schedule
+
+
+@mark.asyncio
+async def test_make_schedule_updater_skips_args_without_attribute(temporal_manager):
+    schedule = {'workflow_type': 'scouter', 'frequency': '1m', 'data': 'new'}
+    update_schedule = temporal_manager._make_schedule_updater(schedule, metadata)
+
+    input_data = MagicMock()
+    input_data.description.schedule.action = object()
+
+    await update_schedule(input_data)
+
+
+@mark.asyncio
+async def test_update_single_schedule_not_found(temporal_manager):
+    client = MagicMock(get_schedule_handle=MagicMock(return_value=None))
+
+    try:
+        await temporal_manager._update_single_schedule(
+            client, 'missing-schedule', {'workflow_type': 'scouter'}, metadata
+        )
+    except ValueError as e:
+        assert str(e) == 'Schedule missing-schedule not found'
+    else:
+        raise AssertionError('Expected a ValueError to be raised')
+
+
+@mark.asyncio
+async def test_update_single_schedule_updates_handle(temporal_manager):
+    handler = MagicMock(update=AsyncMock())
+    client = MagicMock(get_schedule_handle=MagicMock(return_value=handler))
+    schedule = {'workflow_type': 'drift', 'frequency': '1m'}
+
+    await temporal_manager._update_single_schedule(client, 'test-schedule', schedule, metadata)
+
+    client.get_schedule_handle.assert_called_once_with('test-schedule')
+    handler.update.assert_awaited_once()
+    assert schedule['task_queue'] == 'drift-legacy-queue'
+
+
 @mark.asyncio
 @patch('orchestrator.activities.temporal_manager.parse_frequency', side_effect=parse_frequency)
 @patch('orchestrator.activities.temporal_manager.Schedule')
@@ -514,7 +613,7 @@ async def test_create_schedules_default_runtime_legacy_queue(
                 'test-schedule': {
                     'model_id': 1,
                     'model_name': 'test-model-name',
-                    'workflow_type': 'scouter',
+                    'workflow_type': 'drift',
                     'frequency': '1m',
                     'data': {'test': 'test'},
                 }
@@ -527,10 +626,10 @@ async def test_create_schedules_default_runtime_legacy_queue(
     await temporal_manager.create_schedules(input_data)
 
     mock_schedule_action_start_workflow.assert_called_once_with(
-        'scouter',
+        'drift',
         input_data['schedules']['scouter']['test-schedule'],
         id='test-schedule',
-        task_queue='scouter-legacy-queue',
+        task_queue='drift-legacy-queue',
         execution_timeout=timedelta(seconds=300),
         run_timeout=timedelta(seconds=300),
         task_timeout=timedelta(seconds=300),
@@ -562,7 +661,7 @@ async def test_create_schedules_tenant_runtime_queue(
                 'test-schedule': {
                     'model_id': 1,
                     'model_name': 'test-model-name',
-                    'workflow_type': 'scouter',
+                    'workflow_type': 'drift',
                     'frequency': '1m',
                     'runtime': 'tenant-x',
                     'data': {'test': 'test'},
@@ -576,10 +675,10 @@ async def test_create_schedules_tenant_runtime_queue(
     await temporal_manager.create_schedules(input_data)
 
     mock_schedule_action_start_workflow.assert_called_once_with(
-        'scouter',
+        'drift',
         input_data['schedules']['scouter']['test-schedule'],
         id='test-schedule',
-        task_queue='scouter-tenant-x-queue',
+        task_queue='drift-tenant-x-queue',
         execution_timeout=timedelta(seconds=300),
         run_timeout=timedelta(seconds=300),
         task_timeout=timedelta(seconds=300),

validate.sh

@@ -1,124 +0,0 @@
-#!/bin/bash
-# Model Manager Code Validation Script
-# This script runs all code quality checks before committing or deploying
-
-set -e  # Exit on any error
-
-# Colors for output
-RED='\033[0;31m'
-GREEN='\033[0;32m'
-YELLOW='\033[1;33m'
-BLUE='\033[0;34m'
-NC='\033[0m' # No Color
-
-# Args
-FIX_MODE=false
-while [[ $# -gt 0 ]]; do
-    case "$1" in
-        --fix)
-            FIX_MODE=true
-            shift
-            ;;
-        -h|--help)
-            echo "Usage: $0 [--fix]"
-            echo "  --fix   Apply Ruff auto-fixes (format and lint fixes)."
-            exit 0
-            ;;
-        *)
-            echo -e "${RED}Unknown option: $1${NC}"
-            echo "Usage: $0 [--fix]"
-            exit 2
-            ;;
-    esac
-done
-
-echo -e "${BLUE}╔════════════════════════════════════════════════════════╗${NC}"
-echo -e "${BLUE}║  Model Manager - Code Validation Suite                ║${NC}"
-echo -e "${BLUE}╚════════════════════════════════════════════════════════╝${NC}"
-echo ""
-
-# Check if virtual environment is activated
-if [[ -z "${VIRTUAL_ENV}" ]] && [[ -z "${CONDA_DEFAULT_ENV}" ]]; then
-    echo -e "${YELLOW}⚠️  Warning: No virtual environment detected${NC}"
-    echo -e "${YELLOW}   Consider activating your venv/conda environment${NC}"
-    echo ""
-fi
-
-# Function to run a validation step
-run_step() {
-    local step_name=$1
-    local step_command=$2
-    
-    echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
-    echo -e "${BLUE}▶ ${step_name}${NC}"
-    echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
-    
-    if eval "$step_command"; then
-        echo -e "${GREEN}✅ ${step_name} - PASSED${NC}"
-        echo ""
-        return 0
-    else
-        echo -e "${RED}❌ ${step_name} - FAILED${NC}"
-        echo ""
-        return 1
-    fi
-}
-
-# Track failures
-FAILED_STEPS=()
-
-# Step 1: Code Formatting Check (Ruff)
-# - default: check only
-# - --fix: write changes
-if ! run_step "1. Code Formatting (Ruff)" "if \$FIX_MODE; then ruff format orchestrator/ tests/; else ruff format --check orchestrator/ tests/; fi"; then
-    FAILED_STEPS+=("Code Formatting")
-fi
-
-# Step 2: Linting (Ruff)
-# - default: check only
-# - --fix: apply autofixes
-if ! run_step "2. Code Linting (Ruff)" "if \$FIX_MODE; then ruff check --fix orchestrator/ tests/; else ruff check orchestrator/ tests/; fi"; then
-    FAILED_STEPS+=("Linting")
-fi
-
-# Step 3: Type Checking (mypy)
-if ! run_step "3. Type Checking (mypy)" "mypy orchestrator/"; then
-    FAILED_STEPS+=("Type Checking")
-fi
-
-# Step 4: Security Analysis (Bandit)
-if ! run_step "4. Security Analysis (Bandit)" "bandit -r orchestrator/ -ll -q"; then
-    FAILED_STEPS+=("Security Analysis")
-fi
-
-# Step 5: Unit Tests (pytest)
-if ! run_step "5. Unit Tests (pytest)" "pytest tests/ --cov=orchestrator --cov-report=term-missing --cov-report=xml --cov-report=html --cov-fail-under=80 -q"; then
-    FAILED_STEPS+=("Unit Tests")
-fi
-
-# Summary
-echo -e "${BLUE}╔════════════════════════════════════════════════════════╗${NC}"
-echo -e "${BLUE}║  Validation Summary                                    ║${NC}"
-echo -e "${BLUE}╚════════════════════════════════════════════════════════╝${NC}"
-echo ""
-
-if [ ${#FAILED_STEPS[@]} -eq 0 ]; then
-    echo -e "${GREEN}✅ All validation checks passed!${NC}"
-    echo -e "${GREEN}   Your code is ready for commit/deployment.${NC}"
-    echo ""
-    exit 0
-else
-    echo -e "${RED}❌ Validation failed for the following steps:${NC}"
-    for step in "${FAILED_STEPS[@]}"; do
-        echo -e "${RED}   • ${step}${NC}"
-    done
-    echo ""
-    echo -e "${YELLOW}💡 Tips:${NC}"
-    echo -e "${YELLOW}   • Run 'ruff format orchestrator/ tests/' to auto-fix formatting${NC}"
-    echo -e "${YELLOW}   • Run 'ruff check --fix orchestrator/ tests/' to auto-fix linting issues${NC}"
-    echo -e "${YELLOW}   • Review mypy errors and add type hints where needed${NC}"
-    echo -e "${YELLOW}   • Check bandit warnings for security issues${NC}"
-    echo -e "${YELLOW}   • Fix failing tests or improve test coverage${NC}"
-    echo ""
-    exit 1
-fi

values.yaml

@@ -1,265 +0,0 @@
-# Default values for sientia-module.
-# This is a YAML-formatted file.
-# Declare variables to be passed into your templates.
-
-# This will set the replicaset count more information can be found here: https://kubernetes.io/docs/concepts/workloads/controllers/replicaset/
-replicaCount: 1
-
-# This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/
-image:
-  repository: aignosi.azurecr.io/sientia-module
-  # This sets the pull policy for images.
-  pullPolicy: Always
-  # Overrides the image tag whose default is the chart appVersion.
-  tag: "1.0.0"
-
-# This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
-imagePullSecrets:
-- name: docker-hub-secret
-# This is to override the chart name.
-nameOverride: "sientia-orchestrator-worker"
-fullnameOverride: "sientia-orchestrator-worker"
-namespace: sientia
-
-# This section builds out the service account more information can be found here: https://kubernetes.io/docs/concepts/security/service-accounts/
-serviceAccount:
-  # Specifies whether a service account should be created
-  create: true
-  # Automatically mount a ServiceAccount's API credentials?
-  automount: true
-  # Annotations to add to the service account
-  annotations: {}
-  # The name of the service account to use.
-  # If not set and create is true, a name is generated using the fullname template
-  name: "sientia-orchestrator-worker"
-
-# This is for setting Kubernetes Annotations to a Pod.
-# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/
-podAnnotations: {}
-# This is for setting Kubernetes Labels to a Pod.
-# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/
-podLabels: {}
-
-podSecurityContext: {}
-  # fsGroup: 2000
-
-securityContext: {}
-  # capabilities:
-  #   drop:
-  #   - ALL
-  # readOnlyRootFilesystem: true
-  # runAsNonRoot: true
-  # runAsUser: 1000
-
-
-resources: 
-  # We usually recommend not to specify default resources and to leave this as a conscious
-  # choice for the user. This also increases chances charts run on environments with little
-  # resources, such as Minikube. If you do want to specify resources, uncomment the following
-  # lines, adjust them as necessary, and remove the curly braces after 'resources:'.
-  # limits:
-  #   cpu: 100m
-  #   memory: 128Mi
-  # requests:
-  #   cpu: 100m
-  #   memory: 128Mi
-
-  limits:
-    cpu: 1000m
-    memory: 2048Mi
-  requests:
-    cpu: 100m
-    memory: 128Mi
-
-# This is to setup the liveness and readiness probes more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/
-livenessProbe:
-  exec:
-    command:
-      - sh
-      - -c
-      - |
-        curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1'
-  initialDelaySeconds: 30
-  periodSeconds: 15
-  timeoutSeconds: 5
-  failureThreshold: 3
-
-readinessProbe:
-  exec:
-    command:
-      - sh
-      - -c
-      - |
-        curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1'
-  initialDelaySeconds: 20
-  periodSeconds: 10
-  timeoutSeconds: 3
-  failureThreshold: 2
-
-
-# This section is for setting up autoscaling more information can be found here: https://kubernetes.io/docs/concepts/workloads/autoscaling/
-autoscaling:
-  enabled: false
-  minReplicas: 1
-  maxReplicas: 100
-  targetCPUUtilizationPercentage: 80
-  # targetMemoryUtilizationPercentage: 80
-
-# Additional volumes on the output Deployment definition.
-volumes: []
-# - name: foo
-#   secret:
-#     secretName: mysecret
-#     optional: false
-
-# Additional volumeMounts on the output Deployment definition.
-volumeMounts: []
-# - name: foo
-#   mountPath: "/etc/foo"
-#   readOnly: true
-
-nodeSelector: {}
-
-tolerations: []
-
-affinity: {}
-
-services:
-  sdk-metrics:
-    enabled: true
-    type: ClusterIP
-    port: 9091
-    targetPort: 9091
-    name: sdk-metrics
-
-  metrics:
-    enabled: true
-    type: ClusterIP
-    port: 9090
-    targetPort: 9090
-    name: metrics
-
-# Configuração do ServiceMonitor para o Prometheus Operator
-# ref: https://github.com/prometheus-operator/prometheus-operator
-serviceMonitor:
-  # Se true, um recurso ServiceMonitor será criado.
-  enabled: true
-  # O intervalo no qual as métricas devem ser coletadas (ex: 30s, 1m).
-  endpoints:
-  - port: metrics
-    path: /metrics
-    interval: 30s
-    relabelings: []
-  - port: sdk-metrics
-    path: /metrics
-    interval: 30s
-    relabelings: []
-
-  additionalLabels:
-    release: kube-prometheus-stack
-  
-
-env:
-  # Entrypoint variables 
-  - name: GITHUB_REPO_URL
-    value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
-  - name: GITHUB_BRANCH
-    value: "feature/SIENTIAPDE-1646"
-  - name: PYTHON_APP
-    value: "orchestrator.worker.worker"
-
-  # Application variables
-  - name: REDIS_HOST
-    value: "redis-master.redis.svc.cluster.local"
-  - name: REDIS_PORT
-    value: "6379"
-  - name: REDIS_USERNAME
-    valueFrom:
-      secretKeyRef:
-        name: redis
-        key: redis-username
-  - name: REDIS_PASSWORD
-    valueFrom:
-      secretKeyRef:
-        name: redis
-        key: redis-password
-
-  - name: MONGODB_USERNAME
-    value: "root"
-  - name: MONGODB_PASSWORD
-    value: "wKZDbMNU1c"
-  - name: MONGODB_URL
-    value: "my-release-mongodb.mongodb.svc.cluster.local:27017"
-  - name: MONGODB_DATABASE
-    value: "sientia"
-  - name: MONGODB_TTL_INDEX_HOURS
-    value: "1"
-
-  # - name: EMAIL_SENDER
-  #   value: "vitor.santos@aignosi.com.br"
-  # - name: EMAIL_SENDER_PASSWORD
-  #   valueFrom:
-  #     secretKeyRef:
-  #       name: smtp-credentials
-  #       key: app_password
-  # - name: EMAIL_SMTP_SERVER
-  #   value: "smtp.gmail.com"
-  # - name: EMAIL_SMTP_PORT
-  #   value: "587"
-
-  # Application variables
-  - name: POSTGRES_HOST
-    value: "paradedb-rw.paradedb.svc.cluster.local"
-  - name: POSTGRES_PORT
-    value: "5432"
-  - name: POSTGRES_USER
-    value: "sientia"
-  - name: POSTGRES_PASSWORD
-    value: "sientia"
-  - name: POSTGRES_DBNAME
-    value: "sientia"
-  - name: POSTGRES_MIN_CONNECTIONS
-    value: "10"
-  - name: POSTGRES_MAX_CONNECTIONS
-    value: "40"
-
-  - name: KAFKA_BOOTSTRAP_SERVERS
-    value: "kafka.kafka.svc.cluster.local:9092"
-
-  - name: LOG_LEVEL
-    value: "DEBUG"
-  - name: HTTP_METRICS_PORT
-    value: "9090"
-  - name: PROJECT_NAME
-    value: "sientia-orchestrator"
-
-  - name: TEMPORAL_HOST
-    value: "temporal-frontend.temporal.svc.cluster.local:7233"
-  - name: TEMPORAL_NAMESPACE
-    value: "default"
-  - name: TEMPORAL_SCOUTER_NAMESPACE
-    value: "scouter"
-  - name: TEMPORAL_LABORIOUS_NAMESPACE
-    value: "laborious"
-
-  - name: PYPI_SERVER
-    value: "http://library-distribution-server.library.svc.cluster.local:5000"
-
-ssh:
-  enabled: true
-  secretName: git-ssh-key-sientia-orchestrator-worker
-  sshPath: /mnt/.ssh
-  knownHostsPath: /mnt/known_hosts
-
-# kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp
-
-# helm upgrade --install sientia-orchestrator-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.6.6
-
-# kubectl create secret generic git-ssh-key-sientia-orchestrator-worker \
-#   --namespace sientia \
-#   --from-file=ssh-privatekey=git_key \
-#   --type=kubernetes.io/ssh-auth
-
-# kubectl create secret generic smtp-credentials \
-#   --namespace sientia \
-#   --from-literal=app_password='sua-senha-de-app-de-16-digitos'