chore: bump version to 1.3.5

fix: bump sientia_do to 1.12.5 (psycopg2 driver pin)

.github/workflows/deploy.yml

@@ -0,0 +1,29 @@
+name: Deploy DataOps Worker
+
+on:
+  pull_request:
+    types: [closed]
+    branches: [main, 'release/**', 'feature/**']
+  workflow_dispatch:
+
+permissions:
+  contents: write
+  id-token: write
+  pull-requests: read
+
+jobs:
+  deploy:
+    name: Deploy
+    if: github.event_name == 'workflow_dispatch' || github.event.pull_request.merged == true
+    uses: Aignosi/github_workflow_templates/.github/workflows/reusable-deploy-dataops.yml@main
+    with:
+      branch_name: ${{ github.event.pull_request.head.ref }}
+      project_name: 'scouter'
+      chart_name: 'scouter-worker'
+      env_list_path: '[".env"]'
+      update_version_in: '["pyproject"]'
+      create_tarball: ${{ github.event_name == 'pull_request' }}
+      mirror: true
+      bump_chart: true
+      deploy: true
+    secrets: inherit

.github/workflows/release.yml

@@ -1,16 +0,0 @@
-name: Create Release on Merge to Main
-
-on:
-  pull_request:
-    types: [closed]
-    branches:
-      - main
-
-jobs:
-  release:
-    if: github.event.pull_request.merged == true
-    uses: Aignosi/github_workflow_templates/.github/workflows/dataops-module-release.yml@main
-    permissions: write-all
-    with:
-      project_name: 'scouter'
-    secrets: inherit

.github/workflows/require-chart-ref-label.yml

@@ -0,0 +1,16 @@
+name: Require chart-ref label
+
+on:
+  pull_request:
+    types: [opened, synchronize, reopened, labeled, unlabeled]
+    branches:
+      - main
+      - 'release/**'
+      - 'feature/**'
+
+permissions:
+  pull-requests: write
+
+jobs:
+  require-chart-ref-label:
+    uses: Aignosi/github_workflow_templates/.github/workflows/require-chart-ref-label.yml@main

CLAUDE.md

@@ -48,7 +48,7 @@ Parent/child Temporal workflow pattern, two independent worker task queues shari
 ```
 Scouter (scouter/workflow/scouter.py)              PIWebAPIScouter (scouter/workflow/pi_web_api_scouter.py)
   reads raw_{schedule_name} from MongoDB              queries PI Web API directly (get_tag_values)
-  tracks last-processed timestamp in Redis             normalizes all timestamps to the batch max
+  tracks last-processed timestamp in Redis             preserves each record's real timestamp
         │                                                     │
         └───────────────► CoreScouter (scouter/workflow/sub_workflows/core_scouter.py) ◄───────┘
                               1. data_quality_gate   (NULL_VALUES_FILTER / OUT_OF_BOUNDS_FILTER, DISCARD policy)

e2e/scenario_inputs/pi_web_api_scouter_distinct_tag_timestamps_multi_point.json

@@ -0,0 +1,36 @@
+{
+  "workflow_input": {
+    "model_id": "13",
+    "model_name": "PI Distinct Tag Timestamps (max_count>1)",
+    "schedule_name": "pi-distinct-ts-n",
+    "trigger_laborious": false,
+    "filters": {},
+    "schema": "sientia_data",
+    "table_name": "laborious_data",
+    "retention_time": 3600,
+    "fill_missing_tags": false,
+    "model_tags": {
+      "tag_early": {"webid": "w1", "aggr_func": "lts", "data_range": [0, 1000], "frequency": 60000},
+      "tag_late": {"webid": "w2", "aggr_func": "lts", "data_range": [0, 1000], "frequency": 60000}
+    },
+    "pi_web_api_query": {
+      "endpoint": "/streamsets/recorded",
+      "period": "*-1d",
+      "max_count": 2,
+      "api_timeout": 30
+    }
+  },
+  "pi_web_api_server": {
+    "mode": "success",
+    "rows": [
+      {"name": "tag_early", "timestamp": "2024-06-01T12:00:00Z", "value": 10},
+      {"name": "tag_early", "timestamp": "2024-06-01T12:01:00Z", "value": 20},
+      {"name": "tag_late", "timestamp": "2024-06-01T15:00:00Z", "value": 100},
+      {"name": "tag_late", "timestamp": "2024-06-01T15:05:00Z", "value": 200}
+    ]
+  },
+  "expected_rows": {
+    "tag_early": {"value": 20.0, "timestamp": "2024-06-01T12:01:00Z"},
+    "tag_late": {"value": 200.0, "timestamp": "2024-06-01T15:05:00Z"}
+  }
+}

e2e/scenario_inputs/pi_web_api_scouter_distinct_tag_timestamps_single_point.json

@@ -0,0 +1,34 @@
+{
+  "workflow_input": {
+    "model_id": "14",
+    "model_name": "PI Distinct Tag Timestamps (max_count=1)",
+    "schedule_name": "pi-distinct-ts-1",
+    "trigger_laborious": false,
+    "filters": {},
+    "schema": "sientia_data",
+    "table_name": "laborious_data",
+    "retention_time": 3600,
+    "fill_missing_tags": false,
+    "model_tags": {
+      "tag_early": {"webid": "w1", "aggr_func": "lts", "data_range": [0, 1000], "frequency": 60000},
+      "tag_late": {"webid": "w2", "aggr_func": "lts", "data_range": [0, 1000], "frequency": 60000}
+    },
+    "pi_web_api_query": {
+      "endpoint": "/streamsets/recorded",
+      "period": "*-1d",
+      "max_count": 1,
+      "api_timeout": 30
+    }
+  },
+  "pi_web_api_server": {
+    "mode": "success",
+    "rows": [
+      {"name": "tag_early", "timestamp": "2024-06-01T12:00:00Z", "value": 10},
+      {"name": "tag_late", "timestamp": "2024-06-01T15:00:00Z", "value": 100}
+    ]
+  },
+  "expected_rows": {
+    "tag_early": {"value": 10.0, "timestamp": "2024-06-01T12:00:00Z"},
+    "tag_late": {"value": 100.0, "timestamp": "2024-06-01T15:00:00Z"}
+  }
+}

e2e/scenarios.md

@@ -61,6 +61,14 @@ Five tags with `avg` / `mdn` / `max` / `min` / `lts`; assert five distinct `vari
 
 `debug_data_package=True`; assert `data_package_pi_web_api_scouter_*` Redis key with `data` and `held_data`.
 
+### 2.1.4 Distinct tag timestamps preserved
+
+Regression test for SIENTIAPDE-1993, parametrized over `max_count=1` and `max_count>1`. Two
+tags (`tag_early` around 12:00, `tag_late` around 15:00) per case; assert each tag's stored
+`timestamp` in Postgres equals its own latest point after aggregation, not the other tag's
+(the batch-wide max across tags) — the invariant must hold whether a tag has one point or
+several historical points.
+
 ### 2.2.1 Empty response early exit
 
 Server `empty` mode; zero Postgres rows for `model_id`, one request recorded.

e2e/test_pi_web_api_scouter_main_workflow.py

@@ -3,6 +3,7 @@ End-to-end tests for the PIWebAPIScouter main workflow.
 """
 
 import json
+from datetime import datetime
 
 import pytest
 from redis import Redis
@@ -103,6 +104,52 @@ async def test_scenario_2_1_3_debug_data_package(
     assert 'data' in payload and 'held_data' in payload
 
 
+@pytest.mark.e2e
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'scenario_name, workflow_id_prefix',
+    [
+        ('pi_web_api_scouter_distinct_tag_timestamps_single_point', 'pi-distinct-ts-1'),
+        ('pi_web_api_scouter_distinct_tag_timestamps_multi_point', 'pi-distinct-ts-n'),
+    ],
+    ids=['max_count_1', 'max_count_gt_1'],
+)
+async def test_scenario_2_1_4_distinct_tag_timestamps_preserved(
+    temporal_env: WorkflowEnvironment,
+    temporal_worker: Worker,
+    pi_web_api_server: PIWebAPITestServer,
+    postgres_engine,
+    scenario_name: str,
+    workflow_id_prefix: str,
+):
+    """
+    Regression test for SIENTIAPDE-1993: whether a tag has a single point
+    (max_count=1) or several historical points (max_count>1), each tag's
+    stored row must keep the timestamp of its own latest point after
+    aggregation, not another tag's timestamp from the same batch.
+    """
+    scenario = load_scenario_input(scenario_name)
+    apply_pi_web_api_server_config(pi_web_api_server, scenario.get('pi_web_api_server'))
+    workflow_input = scenario['workflow_input']
+
+    await start_and_await_workflow(
+        temporal_env.client,
+        PIWebAPIScouter.run,
+        workflow_input,
+        make_workflow_id(workflow_id_prefix),
+    )
+
+    rows = {
+        row['variable']: row
+        for row in fetch_laborious_rows(postgres_engine, workflow_input['model_id'])
+    }
+    for tag, expected in scenario['expected_rows'].items():
+        assert tag in rows
+        assert rows[tag]['value'] == pytest.approx(expected['value'])
+        expected_timestamp = datetime.fromisoformat(expected['timestamp'].replace('Z', '+00:00'))
+        assert rows[tag]['timestamp'] == expected_timestamp
+
+
 @pytest.mark.e2e
 @pytest.mark.asyncio
 @pytest.mark.xfail(

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "scouter"
-version = "1.3.0"
+version = "1.3.5"
 description = "Sientia DataOps Scouter - ML Model Orchestration System"
 readme = "README.md"
 requires-python = ">=3.11"

requirements-local.txt

@@ -3,6 +3,6 @@ psycopg2-binary
 sqlalchemy
 redis
 pymongo
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.5
 prometheus-client
 pycurl

requirements.txt

@@ -3,6 +3,6 @@ psycopg2-binary
 sqlalchemy
 redis
 pymongo
-sientia_do
+sientia_do>=1.12.5
 prometheus-client
 pycurl

scouter/activities/api.py

@@ -86,10 +86,8 @@ class API(SientiaMonitoring):
         for a set of configured tags. It returns the data as a list of dictionaries
         suitable for further processing in the workflow.
 
-        The timestamps are normalized to ensure consistency across all records in the
+        Timestamps are converted to string format and each record keeps its own real
-        response. After converting timestamps to string format, all timestamps are
+        timestamp as returned by the PI Web API.
-        set to the maximum timestamp value (lexicographically) found in the dataset.
-        This ensures all records in a single batch share the same timestamp value.
 
         Args:
             input_data (dict[str, Any]): Activity input parameters.
@@ -103,7 +101,7 @@ class API(SientiaMonitoring):
 
         Returns:
             list[dict]: List of data records, each containing:
-                - timestamp: Normalized timestamp string (all records share the same value)
+                - timestamp: Each record's real timestamp string
                 - name: Tag name
                 - value: Numeric value
                 - tag: WebId
@@ -148,10 +146,6 @@ class API(SientiaMonitoring):
 
         self.debug(f'Latest values: {latest_values.to_string()}', metadata=metadata)
 
-        # Normalize the package timestamp
-        valid_timestamp_values = latest_values['timestamp'].dropna()
-        latest_values['timestamp'] = valid_timestamp_values.max()
-
         self.info(f'Gathered {len(latest_values)} tag values', metadata=metadata)
 
         return latest_values.to_dict(orient='records')

tests/activities/test_api.py

@@ -87,7 +87,7 @@ def test_close(mock_sientia_monitoring, api_activity):
 
 
 def test_get_tag_values_success(api_activity):
-    """Test get_tag_values with successful data retrieval."""
+    """Test get_tag_values with successful data retrieval preserves each record's real timestamp."""
     # Setup test data
     test_data = {
         **metadata,
@@ -155,8 +155,55 @@ def test_get_tag_values_success(api_activity):
     assert result[0]['value'] == pytest.approx(10.5)
     assert result[1]['name'] == 'tag2'
     assert result[2]['name'] == 'tag3'
-    assert result[0]['timestamp'] == '2023-01-01 12:02:00+0000'
+    # Each record keeps its own real timestamp, no cross-tag normalization
-    assert result[1]['timestamp'] == '2023-01-01 12:02:00+0000'
+    assert result[0]['timestamp'] == '2023-01-01 12:00:00+0000'
+    assert result[1]['timestamp'] == '2023-01-01 12:01:00+0000'
+    assert result[2]['timestamp'] == '2023-01-01 12:02:00+0000'
+
+
+def test_get_tag_values_with_max_count_greater_than_one_preserves_timestamps(api_activity):
+    """Regression test for SIENTIAPDE-1993: with max_count > 1, each historical point's
+    real timestamp must be preserved, otherwise distinct records collide on the
+    (model_id, timestamp, variable) unique constraint downstream and are silently
+    dropped."""
+    test_data = {
+        **metadata,
+        'endpoint': '/streamsets/recorded',
+        'web_ids': {
+            'tag1': {
+                'webid': 'webid1',
+                'aggr_function': 'avg',
+                'data_range': [0, 100],
+            },
+        },
+        'period': '*-1d',
+        'max_count': 3,
+        'api_timeout': 30,
+    }
+
+    # Multiple historical points for the SAME tag, with distinct timestamps
+    mock_df = pd.DataFrame(
+        {
+            'timestamp': [
+                '2023-01-01 12:00:00+0000',
+                '2023-01-01 12:01:00+0000',
+                '2023-01-01 12:02:00+0000',
+            ],
+            'name': ['tag1', 'tag1', 'tag1'],
+            'value': [10.5, 20.3, 30.7],
+            'tag': ['webid1', 'webid1', 'webid1'],
+        }
+    )
+
+    mock_df['timestamp'] = pd.to_datetime(mock_df['timestamp'])
+
+    api_activity.pi_web_api_client.get_latest_values_df = Mock(return_value=mock_df)
+
+    result = api_activity.get_tag_values(test_data)
+
+    assert len(result) == 3
+    assert result[0]['timestamp'] == '2023-01-01 12:00:00+0000'
+    assert result[1]['timestamp'] == '2023-01-01 12:01:00+0000'
     assert result[2]['timestamp'] == '2023-01-01 12:02:00+0000'