From dcf83696388779e943240dbfcd074e9e3b0d87ac Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 14 Jul 2026 16:59:43 -0300 Subject: [PATCH 01/17] feat(gateway): WriteEngine (status_code/timestamp/flags, safe converter, rate-limit) --- ingestor/engines/write.py | 178 +++++++++++++++++++++++++ tests/unit/engines/test_write.py | 220 +++++++++++++++++++++++++++++++ 2 files changed, 398 insertions(+) create mode 100644 ingestor/engines/write.py create mode 100644 tests/unit/engines/test_write.py diff --git a/ingestor/engines/write.py b/ingestor/engines/write.py new file mode 100644 index 0000000..ec1242d --- /dev/null +++ b/ingestor/engines/write.py @@ -0,0 +1,178 @@ +"""OPC write execution (F2 capstone) from a NATS C2 command dict. + +Fixes 1879 correctness gaps: safe value coercion (never raises on bad +input), StatusCode/SourceTimestamp fidelity on the DataValue actually sent +to the server, a rate-limit floor between writes, and OPC error typing. +""" + +import asyncio +from dataclasses import dataclass +from datetime import datetime + +from asyncua import ua +from asyncua.ua.uaerrors import UaStatusCodeError +from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler +from sientia_do.observability.logger import Logger +from sientia_do.observability.metrics_controller import MetricsController +from sientia_do.observability.sientia_monitoring import SientiaMonitoring + +import ingestor.metrics as metrics +from ingestor.connection.clock import MonotonicClock +from ingestor.connection.pool import OpcConnectionPool +from ingestor.converters import ConversionError, to_typed_value +from ingestor.io.data_sink import DataSink + +_VARIANT_TYPES = { + 'float': ua.VariantType.Float, + 'double': ua.VariantType.Double, + 'real': ua.VariantType.Double, + 'int': ua.VariantType.Int32, + 'integer': ua.VariantType.Int32, + 'bool': ua.VariantType.Boolean, + 'boolean': ua.VariantType.Boolean, + 'str': ua.VariantType.String, + 'string': ua.VariantType.String, +} + +# ponytail: same reconnect-worthy Bad*-status subset as laborious's +# opc_repository.py (different repo, not importable here); keep in sync by +# hand or promote to a shared lib if it drifts. +_SESSION_BAD_NAMES = frozenset({ + 'BadSessionIdInvalid', 'BadSessionClosed', 'BadSessionNotActivated', + 'BadSecureChannelIdInvalid', 'BadSecureChannelClosed', 'BadSecureChannelTokenUnknown', + 'BadTcpSecureChannelUnknown', 'BadServerNotConnected', 'BadConnectionClosed', + 'BadConnectionRejected', 'BadNotConnected', 'BadTimeout', 'BadRequestTimeout', + 'BadTooManySessions', +}) + + +@dataclass +class WriteResult: + ok: bool + status: str # 'ok' | 'error' | 'session_bad' + opc_status_code: str + message: str + + +class WriteEngine(SientiaMonitoring): + """Executes one OPC write from a NATS C2 command.""" + + def __init__( + self, + pool: OpcConnectionPool, + sink: DataSink, + logger: Logger, + notification_handler: NotificationHandler, + metrics_controller: MetricsController, + metadata: dict | None = None, + write_timestamp: bool = True, + write_quality: bool = True, + min_pause_s: float = 0.0, + ): + self.pool = pool + self.sink = sink + self.metadata = metadata or {} + self.write_timestamp = write_timestamp + self.write_quality = write_quality + self.min_pause_s = min_pause_s + self.clock = MonotonicClock() + self._last_write_at: float | None = None + + SientiaMonitoring.__init__( + self, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + + async def write(self, cmd: dict) -> WriteResult: + await self._rate_limit() + + try: + typed = to_typed_value(cmd['value'], cmd['data_type']) + except ConversionError as e: + return await self._emit_error(cmd, 'conversion', str(e), 'error') + + dv = self._build_datavalue(cmd, typed) + + start = self.clock.elapsed() + try: + client = await self.pool.get(cmd['server_id'], self._server_cfg(cmd)) + node = client.get_node(cmd['addr']) + await node.write_value(dv) + except Exception as e: + status = 'session_bad' if self._is_session_bad(e) else 'error' + return await self._emit_error(cmd, type(e).__name__, str(e), status) + + return await self._emit_success(cmd, start) + + async def _rate_limit(self) -> None: + now = self.clock.elapsed() + if self.min_pause_s > 0 and self._last_write_at is not None: + remaining = self.min_pause_s - (now - self._last_write_at) + if remaining > 0: + await asyncio.sleep(remaining) + now = self.clock.elapsed() + self._last_write_at = now + + def _server_cfg(self, cmd: dict) -> dict: + # ponytail: minimal cfg built straight from the C2 message. Resolving + # the full connector config (url/security) from the connectors DB by + # server_id is F4 wiring, not this task — pool is mocked in tests. + return { + 'url': cmd.get('url'), + 'use_security': cmd.get('use_security', False), + 'server_uri': cmd.get('server_uri'), + } + + def _variant_type(self, data_type: str) -> ua.VariantType: + return _VARIANT_TYPES.get(data_type.strip().lower(), ua.VariantType.String) + + def _build_datavalue(self, cmd: dict, typed_value) -> ua.DataValue: + variant = ua.Variant(typed_value, self._variant_type(cmd['data_type'])) + write_quality = cmd.get('write_quality', self.write_quality) + write_timestamp = cmd.get('write_timestamp', self.write_timestamp) + # NOTE: DataValue.StatusCode_ default_factory populates a Good status + # even when omitted, so "off" must pass StatusCode_=None explicitly + # or the flag silently no-ops. + status_code = ua.StatusCode(ua.StatusCodes.Good) if write_quality else None + source_ts = datetime.fromisoformat(cmd['timestamp']) if write_timestamp else None + return ua.DataValue(variant, StatusCode_=status_code, SourceTimestamp=source_ts) + + def _is_session_bad(self, exc: Exception) -> bool: + return isinstance(exc, UaStatusCodeError) and type(exc).__name__ in _SESSION_BAD_NAMES + + async def _emit_error(self, cmd: dict, error_type: str, message: str, status: str): + tags = {'pod_id': self.pod_id, 'server_name': cmd.get('server_id', 'unknown')} + await self.emit_metric( + metric_object=metrics.OPC_ERROR_BY_TYPE, tags={**tags, 'error_type': error_type} + ) + await self.emit_metric( + metric_object=metrics.OPC_WRITE_TOTAL, + tags={**tags, 'model_id': cmd.get('model_id', 'unknown'), 'result': 'error'}, + ) + result = WriteResult(ok=False, status=status, opc_status_code=error_type, message=message) + await self.sink.update_write_status(cmd, result) + return result + + async def _emit_success(self, cmd: dict, start: float): + tags = {'pod_id': self.pod_id, 'server_name': cmd['server_id']} + await self.emit_metric( + metric_object=metrics.OPC_WRITE_TOTAL, + tags={**tags, 'model_id': cmd.get('model_id', 'unknown'), 'result': 'success'}, + ) + await self.emit_metric( + metric_object=metrics.OPC_WRITE_LATENCY, + method='observe', + value=self.clock.elapsed() - start, + tags=tags, + ) + await self.emit_metric( + metric_object=metrics.OPC_WRITE_STATUS_BY_MODEL, + method='set', + value=1, + tags={'pod_id': self.pod_id, 'model_id': cmd.get('model_id', 'unknown'), 'status': 'ok'}, + ) + result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='') + await self.sink.update_write_status(cmd, result) + return result diff --git a/tests/unit/engines/test_write.py b/tests/unit/engines/test_write.py new file mode 100644 index 0000000..e92d058 --- /dev/null +++ b/tests/unit/engines/test_write.py @@ -0,0 +1,220 @@ +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from asyncua import ua + +import ingestor.metrics as metrics +from ingestor.engines.write import WriteEngine + + +def make_engine(**kwargs): + eng = WriteEngine( + pool=AsyncMock(), + sink=AsyncMock(), + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=AsyncMock(), + metadata={}, + **kwargs, + ) + eng.emit_metric = AsyncMock() + return eng + + +def make_client(node=None): + node = node or make_node() + client = MagicMock() + client.get_node = MagicMock(return_value=node) + return client + + +def make_node(): + node = MagicMock() + node.write_value = AsyncMock() + return node + + +def base_cmd(**overrides): + cmd = { + 'server_id': 'srv1', + 'addr': 'ns=2;i=5', + 'value': 42.0, + 'data_type': 'float', + 'tag_type': 'prediction', + 'model_id': 'm1', + 'model_name': 'model-1', + 'timestamp': '2026-07-13T12:00:00Z', + 'schema': 'sientia_data', + 'table_name': 'predictions_m1', + } + cmd.update(overrides) + return cmd + + +@pytest.mark.asyncio +async def test_write_success_writes_datavalue_and_returns_ok(): + eng = make_engine() + node = make_node() + client = make_client(node) + eng.pool.get.return_value = client + + result = await eng.write(base_cmd()) + + client.get_node.assert_called_once_with('ns=2;i=5') + node.write_value.assert_awaited_once() + dv = node.write_value.await_args.args[0] + assert isinstance(dv, ua.DataValue) + assert dv.Value.Value == 42.0 + assert dv.Value.VariantType == ua.VariantType.Float + assert result.ok is True + assert result.status == 'ok' + assert result.opc_status_code == 'Good' + eng.sink.update_write_status.assert_awaited_once_with(base_cmd(), result) + + +@pytest.mark.asyncio +async def test_write_success_emits_metrics(): + eng = make_engine() + eng.pool.get.return_value = make_client() + + await eng.write(base_cmd()) + + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_WRITE_TOTAL, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'model_id': 'm1', 'result': 'success'}, + ) + status_call = next( + c + for c in eng.emit_metric.await_args_list + if c.kwargs['metric_object'] is metrics.OPC_WRITE_STATUS_BY_MODEL + ) + assert status_call.kwargs['value'] == 1 + assert status_call.kwargs['tags'] == {'pod_id': eng.pod_id, 'model_id': 'm1', 'status': 'ok'} + latency_call = next( + c + for c in eng.emit_metric.await_args_list + if c.kwargs['metric_object'] is metrics.OPC_WRITE_LATENCY + ) + assert latency_call.kwargs['method'] == 'observe' + assert latency_call.kwargs['value'] >= 0 + + +@pytest.mark.asyncio +async def test_write_defaults_carry_statuscode_and_timestamp(): + eng = make_engine(write_timestamp=True, write_quality=True) + node = make_node() + eng.pool.get.return_value = make_client(node) + + await eng.write(base_cmd()) + + dv = node.write_value.await_args.args[0] + assert dv.StatusCode_ is not None + assert dv.StatusCode_.is_good() + assert dv.SourceTimestamp == datetime.fromisoformat('2026-07-13T12:00:00Z') + + +@pytest.mark.asyncio +async def test_write_engine_flags_off_omit_statuscode_and_timestamp(): + eng = make_engine(write_timestamp=False, write_quality=False) + node = make_node() + eng.pool.get.return_value = make_client(node) + + await eng.write(base_cmd()) + + dv = node.write_value.await_args.args[0] + assert dv.StatusCode_ is None + assert dv.SourceTimestamp is None + + +@pytest.mark.asyncio +async def test_per_message_flag_override_disables_timestamp(): + # Engine default is True, but the command explicitly turns it off. + eng = make_engine(write_timestamp=True, write_quality=True) + node = make_node() + eng.pool.get.return_value = make_client(node) + + await eng.write(base_cmd(write_timestamp=False)) + + dv = node.write_value.await_args.args[0] + assert dv.SourceTimestamp is None + assert dv.StatusCode_ is not None # write_quality untouched, stays on + + +@pytest.mark.asyncio +async def test_write_bad_type_returns_error_without_raising(): + eng = make_engine() + + result = await eng.write(base_cmd(value='abc', data_type='int')) + + assert result.ok is False + assert result.status == 'error' + eng.sink.update_write_status.assert_awaited_once() + eng.pool.get.assert_not_awaited() + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_ERROR_BY_TYPE, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'error_type': 'conversion'}, + ) + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_WRITE_TOTAL, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'model_id': 'm1', 'result': 'error'}, + ) + + +@pytest.mark.asyncio +async def test_write_value_raises_maps_to_error_result_not_raised(): + eng = make_engine() + node = make_node() + node.write_value.side_effect = RuntimeError('boom') + eng.pool.get.return_value = make_client(node) + + result = await eng.write(base_cmd()) + + assert result.ok is False + assert result.status == 'error' + assert result.opc_status_code == 'RuntimeError' + eng.sink.update_write_status.assert_awaited_once() + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_ERROR_BY_TYPE, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'error_type': 'RuntimeError'}, + ) + + +@pytest.mark.asyncio +async def test_write_session_bad_status_code_error_maps_to_session_bad(): + from asyncua.ua.uaerrors import BadSessionClosed + + eng = make_engine() + node = make_node() + node.write_value.side_effect = BadSessionClosed('closed') + eng.pool.get.return_value = make_client(node) + + result = await eng.write(base_cmd()) + + assert result.ok is False + assert result.status == 'session_bad' + + +@pytest.mark.asyncio +async def test_rate_limit_sleeps_between_back_to_back_writes(): + eng = make_engine(min_pause_s=0.05) + eng.pool.get.return_value = make_client() + + with patch('ingestor.engines.write.asyncio.sleep', new_callable=AsyncMock) as sleep_mock: + await eng.write(base_cmd()) + sleep_mock.assert_not_awaited() + await eng.write(base_cmd()) + sleep_mock.assert_awaited_once() + remaining = sleep_mock.await_args.args[0] + assert 0 < remaining <= 0.05 + + +@pytest.mark.asyncio +async def test_no_rate_limit_when_min_pause_is_zero(): + eng = make_engine(min_pause_s=0.0) + eng.pool.get.return_value = make_client() + + with patch('ingestor.engines.write.asyncio.sleep', new_callable=AsyncMock) as sleep_mock: + await eng.write(base_cmd()) + await eng.write(base_cmd()) + sleep_mock.assert_not_awaited() From 86d0f6cf234fa409e859798ec30b6d85a406097e Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 14 Jul 2026 19:18:52 -0300 Subject: [PATCH 02/17] fix(gateway): make WriteEngine + converter raise-safe against malformed write commands --- ingestor/converters.py | 2 ++ ingestor/engines/write.py | 6 ++++- tests/unit/engines/test_write.py | 40 ++++++++++++++++++++++++++++++++ tests/unit/test_converters.py | 7 ++++++ 4 files changed, 54 insertions(+), 1 deletion(-) diff --git a/ingestor/converters.py b/ingestor/converters.py index f785136..d8b5951 100644 --- a/ingestor/converters.py +++ b/ingestor/converters.py @@ -25,6 +25,8 @@ def to_typed_value(raw, data_type: str): data_type: 'boolean' | 'int' | 'float' | 'double' | 'string'. Raises ConversionError on any invalid input (never a bare TypeError). """ + if not isinstance(data_type, str): + raise ConversionError(f'data_type must be a string, got {type(data_type).__name__}') dtype = data_type.strip().lower() try: if dtype in ('bool', 'boolean'): diff --git a/ingestor/engines/write.py b/ingestor/engines/write.py index ec1242d..7bf8f68 100644 --- a/ingestor/engines/write.py +++ b/ingestor/engines/write.py @@ -88,6 +88,10 @@ class WriteEngine(SientiaMonitoring): async def write(self, cmd: dict) -> WriteResult: await self._rate_limit() + missing = [k for k in ('value', 'data_type', 'server_id', 'addr') if k not in cmd] + if missing: + return await self._emit_error(cmd, 'bad_command', f'missing fields: {missing}', 'error') + try: typed = to_typed_value(cmd['value'], cmd['data_type']) except ConversionError as e: @@ -156,7 +160,7 @@ class WriteEngine(SientiaMonitoring): return result async def _emit_success(self, cmd: dict, start: float): - tags = {'pod_id': self.pod_id, 'server_name': cmd['server_id']} + tags = {'pod_id': self.pod_id, 'server_name': cmd.get('server_id', 'unknown')} await self.emit_metric( metric_object=metrics.OPC_WRITE_TOTAL, tags={**tags, 'model_id': cmd.get('model_id', 'unknown'), 'result': 'success'}, diff --git a/tests/unit/engines/test_write.py b/tests/unit/engines/test_write.py index e92d058..d888ee7 100644 --- a/tests/unit/engines/test_write.py +++ b/tests/unit/engines/test_write.py @@ -161,6 +161,46 @@ async def test_write_bad_type_returns_error_without_raising(): ) +@pytest.mark.asyncio +async def test_write_none_data_type_returns_error_without_raising(): + eng = make_engine() + + result = await eng.write(base_cmd(data_type=None)) + + assert result.ok is False + assert result.status == 'error' + eng.sink.update_write_status.assert_awaited_once() + eng.pool.get.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_write_missing_value_key_returns_error_without_raising(): + eng = make_engine() + cmd = base_cmd() + del cmd['value'] + + result = await eng.write(cmd) + + assert result.ok is False + assert result.status == 'error' + eng.sink.update_write_status.assert_awaited_once() + eng.pool.get.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_write_missing_data_type_key_returns_error_without_raising(): + eng = make_engine() + cmd = base_cmd() + del cmd['data_type'] + + result = await eng.write(cmd) + + assert result.ok is False + assert result.status == 'error' + eng.sink.update_write_status.assert_awaited_once() + eng.pool.get.assert_not_awaited() + + @pytest.mark.asyncio async def test_write_value_raises_maps_to_error_result_not_raised(): eng = make_engine() diff --git a/tests/unit/test_converters.py b/tests/unit/test_converters.py index b5a4ff3..6fcfaa7 100644 --- a/tests/unit/test_converters.py +++ b/tests/unit/test_converters.py @@ -22,3 +22,10 @@ def test_float_and_int(): def test_invalid_raises_conversion_error(raw, dtype): with pytest.raises(ConversionError): to_typed_value(raw, dtype) + + +@pytest.mark.parametrize('dtype', [None, 123]) +def test_non_string_data_type_raises_conversion_error(dtype): + # contract: never a bare AttributeError/TypeError, always ConversionError + with pytest.raises(ConversionError): + to_typed_value('5', dtype) From 905da5151b440a7c481cabe5645e9c03ebf8314f Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 14 Jul 2026 19:24:24 -0300 Subject: [PATCH 03/17] feat(gateway): DataSink.update_write_status Postgres by timestamp+model_id (C3); drop unused metadata param --- ingestor/io/data_sink.py | 41 ++++++++++++--- tests/unit/io/test_data_sink_status.py | 72 ++++++++++++++++++++++++++ 2 files changed, 107 insertions(+), 6 deletions(-) create mode 100644 tests/unit/io/test_data_sink_status.py diff --git a/ingestor/io/data_sink.py b/ingestor/io/data_sink.py index 52b526c..ba60404 100644 --- a/ingestor/io/data_sink.py +++ b/ingestor/io/data_sink.py @@ -1,16 +1,18 @@ +import re + +from sqlalchemy import text + import ingestor.metrics as metrics +_IDENTIFIER_RE = re.compile(r'^[A-Za-z0-9_]+$') + class DataSink: - """Gateway egress wrapper: fans a reading out to every configured name/collection. + """Gateway egress wrapper: fans a reading out to every configured name/collection.""" - `postgres` is unused for now — task 12 adds `update_write_status` on top of it. - """ - - def __init__(self, data_manager, postgres, metadata: dict | None = None) -> None: + def __init__(self, data_manager, postgres) -> None: self.data_manager = data_manager self.postgres = postgres - self.metadata = metadata async def emit_metric(self, **kwargs) -> None: # ponytail: delegate to data_manager, it's already a SientiaMonitoring @@ -37,3 +39,30 @@ class DataSink: 'model_id': name['model_id'], }, ) + + async def update_write_status(self, cmd: dict, result) -> None: + """UPDATE the per-model prediction row's write_status (C3), by timestamp+model_id. + + schema/table_name are per-model SQL identifiers, not values — they can't be + bound params, so they're checked against an allowlist before interpolation. + status/timestamp/model_id ARE bound params, never f-stringed. + """ + schema = self._validate_identifier(cmd['schema']) + table = self._validate_identifier(cmd['table_name']) + # schema/table are allowlist-validated identifiers above; status/timestamp/model_id + # are bound params below, never interpolated — not a real injection vector. + stmt = text( + f'UPDATE {schema}.{table} SET write_status = :status ' # noqa: S608 + 'WHERE timestamp = :timestamp AND model_id = :model_id' + ) + params = {'status': result.status, 'timestamp': cmd['timestamp'], 'model_id': cmd['model_id']} + # ponytail: sync engine call inside async def — same pattern sientia_do's own + # Postgres activities use (e.g. export_data_to_postgres); no async pg driver here. + with self.postgres.engine.begin() as connection: + connection.execute(stmt, params) + + @staticmethod + def _validate_identifier(name: str) -> str: + if not _IDENTIFIER_RE.match(name): + raise ValueError(f'invalid SQL identifier: {name!r}') + return name diff --git a/tests/unit/io/test_data_sink_status.py b/tests/unit/io/test_data_sink_status.py new file mode 100644 index 0000000..6cdaaa3 --- /dev/null +++ b/tests/unit/io/test_data_sink_status.py @@ -0,0 +1,72 @@ +from unittest.mock import MagicMock + +import pytest + +from ingestor.engines.write import WriteResult +from ingestor.io.data_sink import DataSink + + +def make_sink(): + postgres = MagicMock() + connection = MagicMock() + postgres.engine.begin.return_value.__enter__.return_value = connection + sink = DataSink(data_manager=MagicMock(), postgres=postgres) + return sink, connection + + +@pytest.mark.asyncio +async def test_update_write_status_targets_model_row(): + sink, connection = make_sink() + cmd = { + 'schema': 'sientia_data', + 'table_name': 'pred_m1', + 'model_id': '1', + 'timestamp': '2026-07-13T00:00:00Z', + } + result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='') + + await sink.update_write_status(cmd, result) + + assert connection.execute.call_count == 1 + stmt, params = connection.execute.call_args[0] + sql = str(stmt) + assert 'update' in sql.lower() and 'write_status' in sql.lower() + assert 'sientia_data.pred_m1' in sql + # values must be bound params, never interpolated into the SQL text itself + assert params == {'status': 'ok', 'timestamp': '2026-07-13T00:00:00Z', 'model_id': '1'} + assert 'ok' not in sql + assert '2026-07-13T00:00:00Z' not in sql + + +@pytest.mark.asyncio +async def test_update_write_status_rejects_bad_schema(): + sink, connection = make_sink() + cmd = { + 'schema': 'a; DROP TABLE x', + 'table_name': 'pred_m1', + 'model_id': '1', + 'timestamp': '2026-07-13T00:00:00Z', + } + result = WriteResult(ok=False, status='error', opc_status_code='Bad', message='x') + + with pytest.raises(ValueError): + await sink.update_write_status(cmd, result) + + connection.execute.assert_not_called() + + +@pytest.mark.asyncio +async def test_update_write_status_rejects_bad_table_name(): + sink, connection = make_sink() + cmd = { + 'schema': 'sientia_data', + 'table_name': 'pred_m1; DROP TABLE x', + 'model_id': '1', + 'timestamp': '2026-07-13T00:00:00Z', + } + result = WriteResult(ok=False, status='error', opc_status_code='Bad', message='x') + + with pytest.raises(ValueError): + await sink.update_write_status(cmd, result) + + connection.execute.assert_not_called() From abff80e0d3e5e99d2e6577c788b6b64a767990ee Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 14 Jul 2026 19:31:11 -0300 Subject: [PATCH 04/17] fix(gateway): tighten SQL identifier validation to fullmatch (reject trailing newline) --- ingestor/io/data_sink.py | 6 ++++-- tests/unit/io/test_data_sink_status.py | 17 +++++++++++++++++ 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/ingestor/io/data_sink.py b/ingestor/io/data_sink.py index ba60404..eef8a50 100644 --- a/ingestor/io/data_sink.py +++ b/ingestor/io/data_sink.py @@ -4,7 +4,9 @@ from sqlalchemy import text import ingestor.metrics as metrics -_IDENTIFIER_RE = re.compile(r'^[A-Za-z0-9_]+$') +# fullmatch (no ^/$ anchors) so a trailing newline can't slip through — `$` would +# match before a trailing '\n', which is not what we want for a SQL identifier. +_IDENTIFIER_RE = re.compile(r'[A-Za-z0-9_]+') class DataSink: @@ -63,6 +65,6 @@ class DataSink: @staticmethod def _validate_identifier(name: str) -> str: - if not _IDENTIFIER_RE.match(name): + if not isinstance(name, str) or not _IDENTIFIER_RE.fullmatch(name): raise ValueError(f'invalid SQL identifier: {name!r}') return name diff --git a/tests/unit/io/test_data_sink_status.py b/tests/unit/io/test_data_sink_status.py index 6cdaaa3..a626036 100644 --- a/tests/unit/io/test_data_sink_status.py +++ b/tests/unit/io/test_data_sink_status.py @@ -55,6 +55,23 @@ async def test_update_write_status_rejects_bad_schema(): connection.execute.assert_not_called() +@pytest.mark.asyncio +async def test_update_write_status_rejects_trailing_newline_identifier(): + sink, connection = make_sink() + cmd = { + 'schema': 'sientia_data\n', + 'table_name': 'pred_m1', + 'model_id': '1', + 'timestamp': '2026-07-13T00:00:00Z', + } + result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='') + + with pytest.raises(ValueError): + await sink.update_write_status(cmd, result) + + connection.execute.assert_not_called() + + @pytest.mark.asyncio async def test_update_write_status_rejects_bad_table_name(): sink, connection = make_sink() From 165ac79783f87b366a09faa5782eb2a92f3a0b8e Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 14 Jul 2026 21:31:53 -0300 Subject: [PATCH 05/17] feat(gateway): NatsWriteConsumer (Core NATS, queue group) (C2) --- ingestor/adapters/__init__.py | 0 ingestor/adapters/nats_consumer.py | 53 ++++++++++++++++ tests/unit/adapters/__init__.py | 0 tests/unit/adapters/test_nats_consumer.py | 76 +++++++++++++++++++++++ 4 files changed, 129 insertions(+) create mode 100644 ingestor/adapters/__init__.py create mode 100644 ingestor/adapters/nats_consumer.py create mode 100644 tests/unit/adapters/__init__.py create mode 100644 tests/unit/adapters/test_nats_consumer.py diff --git a/ingestor/adapters/__init__.py b/ingestor/adapters/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ingestor/adapters/nats_consumer.py b/ingestor/adapters/nats_consumer.py new file mode 100644 index 0000000..d0f9960 --- /dev/null +++ b/ingestor/adapters/nats_consumer.py @@ -0,0 +1,53 @@ +"""NATS input adapter for OPC write commands (contract C2). + +Thin adapter: decode the message JSON and hand the dict to WriteEngine, +which is already raise-safe against malformed commands. No metrics of its +own — the engine records write metrics; a malformed-JSON drop is logged +here. No reconnect logic — nats-py handles reconnection internally. +""" + +import json + +import nats +from sientia_do.observability.logger import Logger + +from ingestor.engines.write import WriteEngine + + +class NatsWriteConsumer: + def __init__( + self, + nats_url: str, + write_engine: WriteEngine, + logger: Logger, + subject: str = 'opc.write', + queue: str = 'gateway-writers', + ): + self.nats_url = nats_url + self.write_engine = write_engine + self.logger = logger + self.subject = subject + self.queue = queue + self.nc = None + self.sub = None + + async def start(self) -> None: + self.nc = await nats.connect(servers=self.nats_url) + self.sub = await self.nc.subscribe(self.subject, queue=self.queue, cb=self._handle) + self.logger.info(f'NatsWriteConsumer started: subject={self.subject} queue={self.queue}') + + async def _handle(self, msg) -> None: + try: + payload = json.loads(msg.data) + except (ValueError, UnicodeDecodeError) as e: + preview = msg.data[:200] + self.logger.error(f'NatsWriteConsumer dropped malformed JSON ({e}): {preview!r}') + return + await self.write_engine.write(payload) + + async def stop(self) -> None: + if self.nc is None: + return + if self.sub is not None: + await self.sub.unsubscribe() + await self.nc.drain() diff --git a/tests/unit/adapters/__init__.py b/tests/unit/adapters/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/adapters/test_nats_consumer.py b/tests/unit/adapters/test_nats_consumer.py new file mode 100644 index 0000000..7410dab --- /dev/null +++ b/tests/unit/adapters/test_nats_consumer.py @@ -0,0 +1,76 @@ +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from ingestor.adapters.nats_consumer import NatsWriteConsumer + + +def make_consumer(**kwargs): + return NatsWriteConsumer( + nats_url='nats://localhost:4222', + write_engine=AsyncMock(), + logger=MagicMock(), + **kwargs, + ) + + +@pytest.mark.asyncio +async def test_handle_valid_json_dispatches_to_write_engine(): + consumer = make_consumer() + payload = {'server_id': 'srv1', 'addr': 'ns=2;i=5', 'value': 1.0, 'data_type': 'float'} + msg = MagicMock(data=json.dumps(payload).encode()) + + await consumer._handle(msg) + + consumer.write_engine.write.assert_awaited_once_with(payload) + + +@pytest.mark.asyncio +async def test_handle_invalid_json_drops_message_and_logs_error(): + consumer = make_consumer() + msg = MagicMock(data=b'not-json') + + await consumer._handle(msg) + + consumer.write_engine.write.assert_not_awaited() + consumer.logger.error.assert_called_once() + + +@pytest.mark.asyncio +async def test_start_subscribes_with_subject_and_queue(): + consumer = make_consumer(subject='opc.write', queue='gateway-writers') + mock_nc = AsyncMock() + mock_nc.subscribe = AsyncMock() + + with patch('ingestor.adapters.nats_consumer.nats') as mock_nats: + mock_nats.connect = AsyncMock(return_value=mock_nc) + await consumer.start() + + mock_nats.connect.assert_awaited_once_with(servers='nats://localhost:4222') + mock_nc.subscribe.assert_awaited_once_with( + 'opc.write', queue='gateway-writers', cb=consumer._handle + ) + + +@pytest.mark.asyncio +async def test_stop_before_start_does_not_raise(): + consumer = make_consumer() + + await consumer.stop() # no raise + + +@pytest.mark.asyncio +async def test_stop_after_start_drains_and_unsubscribes(): + consumer = make_consumer() + mock_nc = AsyncMock() + mock_sub = AsyncMock() + mock_nc.subscribe = AsyncMock(return_value=mock_sub) + + with patch('ingestor.adapters.nats_consumer.nats') as mock_nats: + mock_nats.connect = AsyncMock(return_value=mock_nc) + await consumer.start() + await consumer.stop() + + mock_sub.unsubscribe.assert_awaited_once() + mock_nc.drain.assert_awaited_once() From ce0db62848ea9e9eabf19a7875d1f81add7298ac Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 14 Jul 2026 22:18:28 -0300 Subject: [PATCH 06/17] SIENTIAPDE-1972: F2 close-out - ledger de commits + carry-forward notes (Task 13.end) --- .../superpowers/plans/2026-07-13-opc-gateway.md | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/docs/superpowers/plans/2026-07-13-opc-gateway.md b/docs/superpowers/plans/2026-07-13-opc-gateway.md index 258a904..687310b 100644 --- a/docs/superpowers/plans/2026-07-13-opc-gateway.md +++ b/docs/superpowers/plans/2026-07-13-opc-gateway.md @@ -80,7 +80,16 @@ At the end of each phase Fx, before starting the next: - **Task 18 (wiring):** decide how many `SubscriptionEngine` instances the app builds. Current code keeps a single engine-wide `non_receive_count` and `check_opc_listenning(server_name)` takes a param — safe only if **one engine per OPC server**. If one engine is shared across servers, a datachange on server A masks silence on server B → must partition cycle/non-receive state per server. Make this deliberate at wiring time. - **Task 12 (DataSink):** drop the currently-unused `metadata` param from `DataSink.__init__` while adding `update_write_status`. - **F4 cleanup:** `README.md` and `tests/functional/test_single_node.py`/`conftest.py` still reference Kafka (stale/commented dead code) — remove during F4 hardening (Task 21 sweep). -**F2 — Active write via NATS (C2, C3):** _(fill at F2 close-out)_ +**F2 — Active write via NATS (C2, C3)** — base `2a9af05`. Status: ✅ complete, 192 unit tests green, ruff clean. +- `dcf8369` feat(gateway): WriteEngine (status_code/timestamp/flags, safe converter, rate-limit) (Task 11) +- `86d0f6c` fix(gateway): make WriteEngine + converter raise-safe against malformed write commands (Task 11) +- `905da51` feat(gateway): DataSink.update_write_status Postgres by timestamp+model_id (C3); drop unused metadata param (Task 12) +- `abff80e` fix(gateway): tighten SQL identifier validation to fullmatch (reject trailing newline) (Task 12) +- `165ac79` feat(gateway): NatsWriteConsumer (Core NATS, queue group) (C2) (Task 13) +- Carry-forward: + - **asyncua version hygiene:** requirements pin `asyncua==1.1.5` but the dev/test env runs 1.1.8. The DataValue/StatusCode_ behavior was verified identical on both, but align the pin (bump to 1.1.8) or the env during F4 (Task 21) so version-specific bugs can't slip through. + - **`_SESSION_BAD_NAMES` in write.py** duplicates a subset of Laborious's reconnectable-error list (cross-repo) — move to `sientia_do` if it drifts (ponytail note in code). + - **`_server_cfg` in write.py** is minimal (url/security off the cmd); full connector-DB lookup by `server_id` is Task 18 (F4 wiring) — ponytail note in code. **F3 — Active reads + gRPC (C4):** _(fill at F3 close-out)_ **F4 — Wiring, health, deploy, hardening:** _(fill at F4 close-out)_ @@ -726,9 +735,9 @@ async def test_bad_json_does_not_raise(): --- ### Task 13.end: F2 close-out (see [Phase Delivery](#phase-delivery--commit-ledger)) -- [ ] Full `pytest tests` green + `ruff` clean. -- [ ] Record F2 commit SHAs (base + range) in the [Commit Ledger](#commit-ledger). -- [ ] Do NOT open the PR yet. +- [x] Full `pytest tests` green (192) + `ruff` clean. +- [x] F2 commit SHAs recorded in the [Commit Ledger](#commit-ledger) (base `2a9af05`). +- [ ] PR intentionally NOT opened yet — opened only after F4 closes. --- From f7f356fbd5883f0b204619f1db3ea4eec38d7c57 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 14 Jul 2026 22:24:02 -0300 Subject: [PATCH 07/17] refactor(gateway): ponytail F2 - drop dead metadata, unreachable variant fallback, redundant except member --- ingestor/adapters/nats_consumer.py | 2 +- ingestor/engines/write.py | 5 ++--- tests/unit/engines/test_write.py | 1 - 3 files changed, 3 insertions(+), 5 deletions(-) diff --git a/ingestor/adapters/nats_consumer.py b/ingestor/adapters/nats_consumer.py index d0f9960..189220d 100644 --- a/ingestor/adapters/nats_consumer.py +++ b/ingestor/adapters/nats_consumer.py @@ -39,7 +39,7 @@ class NatsWriteConsumer: async def _handle(self, msg) -> None: try: payload = json.loads(msg.data) - except (ValueError, UnicodeDecodeError) as e: + except ValueError as e: # includes JSONDecodeError and UnicodeDecodeError preview = msg.data[:200] self.logger.error(f'NatsWriteConsumer dropped malformed JSON ({e}): {preview!r}') return diff --git a/ingestor/engines/write.py b/ingestor/engines/write.py index 7bf8f68..2f90d64 100644 --- a/ingestor/engines/write.py +++ b/ingestor/engines/write.py @@ -64,14 +64,12 @@ class WriteEngine(SientiaMonitoring): logger: Logger, notification_handler: NotificationHandler, metrics_controller: MetricsController, - metadata: dict | None = None, write_timestamp: bool = True, write_quality: bool = True, min_pause_s: float = 0.0, ): self.pool = pool self.sink = sink - self.metadata = metadata or {} self.write_timestamp = write_timestamp self.write_quality = write_quality self.min_pause_s = min_pause_s @@ -130,7 +128,8 @@ class WriteEngine(SientiaMonitoring): } def _variant_type(self, data_type: str) -> ua.VariantType: - return _VARIANT_TYPES.get(data_type.strip().lower(), ua.VariantType.String) + # direct index: unknown dtypes already became ConversionError in to_typed_value + return _VARIANT_TYPES[data_type.strip().lower()] def _build_datavalue(self, cmd: dict, typed_value) -> ua.DataValue: variant = ua.Variant(typed_value, self._variant_type(cmd['data_type'])) diff --git a/tests/unit/engines/test_write.py b/tests/unit/engines/test_write.py index d888ee7..6fc1963 100644 --- a/tests/unit/engines/test_write.py +++ b/tests/unit/engines/test_write.py @@ -15,7 +15,6 @@ def make_engine(**kwargs): logger=MagicMock(), notification_handler=MagicMock(), metrics_controller=AsyncMock(), - metadata={}, **kwargs, ) eng.emit_metric = AsyncMock() From 4855205aa43b7209531be81a61b23b2b6c525e30 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 14 Jul 2026 22:24:28 -0300 Subject: [PATCH 08/17] SIENTIAPDE-1972: adiciona commit ponytail ao ledger F2 --- docs/superpowers/plans/2026-07-13-opc-gateway.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/superpowers/plans/2026-07-13-opc-gateway.md b/docs/superpowers/plans/2026-07-13-opc-gateway.md index 687310b..95df039 100644 --- a/docs/superpowers/plans/2026-07-13-opc-gateway.md +++ b/docs/superpowers/plans/2026-07-13-opc-gateway.md @@ -86,6 +86,7 @@ At the end of each phase Fx, before starting the next: - `905da51` feat(gateway): DataSink.update_write_status Postgres by timestamp+model_id (C3); drop unused metadata param (Task 12) - `abff80e` fix(gateway): tighten SQL identifier validation to fullmatch (reject trailing newline) (Task 12) - `165ac79` feat(gateway): NatsWriteConsumer (Core NATS, queue group) (C2) (Task 13) +- `f7f356f` refactor(gateway): ponytail F2 - drop dead metadata, unreachable variant fallback, redundant except member - Carry-forward: - **asyncua version hygiene:** requirements pin `asyncua==1.1.5` but the dev/test env runs 1.1.8. The DataValue/StatusCode_ behavior was verified identical on both, but align the pin (bump to 1.1.8) or the env during F4 (Task 21) so version-specific bugs can't slip through. - **`_SESSION_BAD_NAMES` in write.py** duplicates a subset of Laborious's reconnectable-error list (cross-repo) — move to `sientia_do` if it drifts (ponytail note in code). From 99a715cc1661c78668d44a47a69d07a7aa528067 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 14 Jul 2026 22:30:06 -0300 Subject: [PATCH 09/17] feat(gateway): ActiveReadEngine (scheduled + read_now) --- ingestor/engines/active_read.py | 124 +++++++++++++++++ tests/unit/engines/test_active_read.py | 177 +++++++++++++++++++++++++ 2 files changed, 301 insertions(+) create mode 100644 ingestor/engines/active_read.py create mode 100644 tests/unit/engines/test_active_read.py diff --git a/ingestor/engines/active_read.py b/ingestor/engines/active_read.py new file mode 100644 index 0000000..9748517 --- /dev/null +++ b/ingestor/engines/active_read.py @@ -0,0 +1,124 @@ +"""Active OPC reads (F3): on-demand (`read_now`, used by the gRPC server - +Task 16) and scheduled (`scheduled_tick`, driven by the app main loop each +collect cycle). Emits metrics only -- unlike WriteEngine there is no +notification path here.""" + +from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler +from sientia_do.observability.logger import Logger +from sientia_do.observability.metrics_controller import MetricsController +from sientia_do.observability.sientia_monitoring import SientiaMonitoring +from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, OPC_TIMEZONE + +import ingestor.metrics as metrics +from ingestor.connection.clock import MonotonicClock +from ingestor.connection.pool import OpcConnectionPool +from ingestor.io.data_sink import DataSink + + +class ActiveReadEngine(SientiaMonitoring): + """On-demand + scheduled OPC reads, sharing the pool/sink with WriteEngine + and SubscriptionEngine.""" + + def __init__( + self, + pool: OpcConnectionPool, + sink: DataSink, + logger: Logger, + notification_handler: NotificationHandler, + metrics_controller: MetricsController, + ): + self.pool = pool + self.sink = sink + self.clock = MonotonicClock() + self._ticks: dict[tuple[str, str], int] = {} + + SientiaMonitoring.__init__( + self, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + + async def read_now( + self, server_id: str, addr: str, names: list[dict] | None = None, server_cfg: dict | None = None + ) -> dict: + """Read `addr` on `server_id` right now; persist via sink only if `names` given.""" + tags = {'pod_id': self.pod_id, 'server_name': server_id} + start = self.clock.elapsed() + try: + client = await self.pool.get(server_id, server_cfg or {}) + dv = await client.get_node(addr).read_data_value() + except Exception as e: + await self.emit_metric( + metric_object=metrics.OPC_ACTIVE_READ_TOTAL, tags={**tags, 'result': 'error'} + ) + self.logger.error(f'Active read failed for {server_id}:{addr}: {e}') + raise + + await self.emit_metric( + metric_object=metrics.OPC_ACTIVE_READ_LATENCY, + method='observe', + value=self.clock.elapsed() - start, + tags=tags, + ) + await self.emit_metric( + metric_object=metrics.OPC_ACTIVE_READ_TOTAL, tags={**tags, 'result': 'success'} + ) + return await self._build_result(dv, addr, names, tags) + + async def _build_result(self, dv, addr: str, names: list[dict] | None, tags: dict) -> dict: + value, status_name, source_ts, server_ts = self._extract(dv) + await self._record_read_lag(source_ts, tags) + source_str = self._format_ts(source_ts) + persisted = bool(names) + if persisted: + await self.sink.publish_reading(tag=addr, value=value, timestamp=source_str, names=names) + return { + 'value': value, + 'opc_status_code': status_name, + 'source_timestamp': source_str, + 'server_timestamp': self._format_ts(server_ts), + 'persisted': persisted, + } + + @staticmethod + def _extract(dv): + status_name = dv.StatusCode_.name if dv.StatusCode_ is not None else None + return dv.Value.Value, status_name, dv.SourceTimestamp, dv.ServerTimestamp + + @staticmethod + def _format_ts(ts) -> str | None: + if ts is None: + return None + return ts.replace(tzinfo=OPC_TIMEZONE).strftime(DATETIME_FORMAT_WITH_TZ) + + async def _record_read_lag(self, source_ts, tags: dict) -> None: + if source_ts is None: + return + lag = MonotonicClock.now_wall() - source_ts.replace(tzinfo=OPC_TIMEZONE).timestamp() + await self.emit_metric(metric_object=metrics.OPC_READ_LAG, method='set', value=lag, tags=tags) + + async def scheduled_tick(self, slot: dict) -> None: + """Sweep every server/tag in a parsed slot, reading tags due for an active read.""" + for server in slot.values(): + server_id = server['server_id'] + server_cfg = {k: v for k, v in server.items() if k != 'tags'} + for addr, tag in server['tags'].items(): + await self._tick_tag(server_id, addr, tag, server_cfg) + + async def _tick_tag(self, server_id: str, addr: str, tag: dict, server_cfg: dict) -> None: + active_read = tag.get('active_read') or {} + if not active_read.get('enabled'): + return + if not self._should_read(server_id, addr, active_read.get('cycle_multiple', 1)): + return + try: + await self.read_now(server_id, addr, names=tag['names'], server_cfg=server_cfg) + except Exception as e: + self.logger.error(f'Scheduled active read failed for {server_id}:{addr}: {e}') + + def _should_read(self, server_id: str, addr: str, cycle_multiple: int) -> bool: + key = (server_id, addr) + count = self._ticks.get(key, 0) + 1 + self._ticks[key] = count + return count % max(cycle_multiple, 1) == 0 diff --git a/tests/unit/engines/test_active_read.py b/tests/unit/engines/test_active_read.py new file mode 100644 index 0000000..d880a9e --- /dev/null +++ b/tests/unit/engines/test_active_read.py @@ -0,0 +1,177 @@ +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest +from asyncua import ua + +import ingestor.metrics as metrics +from ingestor.engines.active_read import ActiveReadEngine + + +def make_engine(): + eng = ActiveReadEngine( + pool=AsyncMock(), + sink=AsyncMock(), + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=AsyncMock(), + ) + eng.emit_metric = AsyncMock() + return eng + + +def make_dv(value=42.0, status='Good', source_ts=None, server_ts=None): + source_ts = source_ts or datetime(2026, 7, 13, 12, 0, 0) + server_ts = server_ts or source_ts + return ua.DataValue( + ua.Variant(value), + StatusCode_=ua.StatusCode(getattr(ua.StatusCodes, status)), + SourceTimestamp=source_ts, + ServerTimestamp=server_ts, + ) + + +def make_client(dv=None): + node = MagicMock() + node.read_data_value = AsyncMock(return_value=dv or make_dv()) + client = MagicMock() + client.get_node = MagicMock(return_value=node) + return client, node + + +NAMES = [{'tag_name': 'a', 'model_id': 'm1', 'collection': 'coll'}] + + +@pytest.mark.asyncio +async def test_read_now_persists_and_returns_fields_when_names_given(): + eng = make_engine() + client, node = make_client() + eng.pool.get.return_value = client + + result = await eng.read_now('srv1', 'ns=2;i=5', names=NAMES) + + client.get_node.assert_called_once_with('ns=2;i=5') + node.read_data_value.assert_awaited_once() + eng.sink.publish_reading.assert_awaited_once() + _, kwargs = eng.sink.publish_reading.call_args + assert kwargs['tag'] == 'ns=2;i=5' + assert kwargs['value'] == 42.0 + assert kwargs['names'] == NAMES + assert result['value'] == 42.0 + assert result['opc_status_code'] == 'Good' + assert result['persisted'] is True + assert result['source_timestamp'].startswith('2026-07-13 12:00:00') + assert result['server_timestamp'].startswith('2026-07-13 12:00:00') + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_ACTIVE_READ_TOTAL, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'result': 'success'}, + ) + latency_call = next( + c + for c in eng.emit_metric.await_args_list + if c.kwargs['metric_object'] is metrics.OPC_ACTIVE_READ_LATENCY + ) + assert latency_call.kwargs['method'] == 'observe' + assert latency_call.kwargs['value'] >= 0 + lag_call = next( + c for c in eng.emit_metric.await_args_list if c.kwargs['metric_object'] is metrics.OPC_READ_LAG + ) + assert lag_call.kwargs['method'] == 'set' + + +@pytest.mark.asyncio +async def test_read_now_without_names_does_not_persist(): + eng = make_engine() + client, _ = make_client() + eng.pool.get.return_value = client + + result = await eng.read_now('srv1', 'ns=2;i=5') + + eng.sink.publish_reading.assert_not_awaited() + assert result['persisted'] is False + + +@pytest.mark.asyncio +async def test_read_now_error_emits_metric_and_reraises(): + eng = make_engine() + node = MagicMock() + node.read_data_value = AsyncMock(side_effect=RuntimeError('boom')) + client = MagicMock() + client.get_node = MagicMock(return_value=node) + eng.pool.get.return_value = client + + with pytest.raises(RuntimeError, match='boom'): + await eng.read_now('srv1', 'ns=2;i=5') + + eng.emit_metric.assert_any_await( + metric_object=metrics.OPC_ACTIVE_READ_TOTAL, + tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'result': 'error'}, + ) + eng.sink.publish_reading.assert_not_awaited() + + +def make_slot(cycle_multiple=None, enabled=True): + active_read = {'enabled': enabled} + if cycle_multiple is not None: + active_read['cycle_multiple'] = cycle_multiple + return { + 'srv_name': { + 'server_id': 'srv1', + 'url': 'opc.tcp://x', + 'tags': { + 'ns=2;i=5': {'frequency': 1000, 'names': NAMES, 'active_read': active_read}, + }, + } + } + + +@pytest.mark.asyncio +async def test_scheduled_tick_respects_cycle_multiple(): + eng = make_engine() + eng.read_now = AsyncMock() + slot = make_slot(cycle_multiple=3) + + await eng.scheduled_tick(slot) + await eng.scheduled_tick(slot) + eng.read_now.assert_not_awaited() + await eng.scheduled_tick(slot) + eng.read_now.assert_awaited_once_with( + 'srv1', 'ns=2;i=5', names=NAMES, server_cfg={'server_id': 'srv1', 'url': 'opc.tcp://x'} + ) + + +@pytest.mark.asyncio +async def test_scheduled_tick_skips_disabled_tags(): + eng = make_engine() + eng.read_now = AsyncMock() + slot = make_slot(enabled=False) + + await eng.scheduled_tick(slot) + + eng.read_now.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_scheduled_tick_continues_past_tag_error(): + eng = make_engine() + bad_slot = make_slot(cycle_multiple=1) + good_names = [{'tag_name': 'b', 'model_id': 'm2', 'collection': 'coll'}] + bad_slot['srv_name']['tags']['ns=2;i=6'] = { + 'frequency': 1000, + 'names': good_names, + 'active_read': {'enabled': True, 'cycle_multiple': 1}, + } + + calls = [] + + async def fake_read_now(server_id, addr, names=None, server_cfg=None): + calls.append(addr) + if addr == 'ns=2;i=5': + raise RuntimeError('boom') + + eng.read_now = AsyncMock(side_effect=fake_read_now) + + await eng.scheduled_tick(bad_slot) + + assert set(calls) == {'ns=2;i=5', 'ns=2;i=6'} + eng.logger.error.assert_called_once() From 1798b69c8b8676ad4f201382159f3d79e891dee6 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 15 Jul 2026 08:36:08 -0300 Subject: [PATCH 10/17] feat(gateway): gRPC proto + generated stubs (C4) --- ingestor/proto/__init__.py | 0 ingestor/proto/opc_gateway.proto | 18 +++++ ingestor/proto/opc_gateway_pb2.py | 40 +++++++++++ ingestor/proto/opc_gateway_pb2_grpc.py | 97 ++++++++++++++++++++++++++ pyproject.toml | 1 + tests/unit/proto/__init__.py | 0 tests/unit/proto/test_proto.py | 36 ++++++++++ 7 files changed, 192 insertions(+) create mode 100644 ingestor/proto/__init__.py create mode 100644 ingestor/proto/opc_gateway.proto create mode 100644 ingestor/proto/opc_gateway_pb2.py create mode 100644 ingestor/proto/opc_gateway_pb2_grpc.py create mode 100644 tests/unit/proto/__init__.py create mode 100644 tests/unit/proto/test_proto.py diff --git a/ingestor/proto/__init__.py b/ingestor/proto/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ingestor/proto/opc_gateway.proto b/ingestor/proto/opc_gateway.proto new file mode 100644 index 0000000..6def8d3 --- /dev/null +++ b/ingestor/proto/opc_gateway.proto @@ -0,0 +1,18 @@ +syntax = "proto3"; +package opcgateway; + +service OpcGateway { + rpc ReadTag(ReadTagRequest) returns (ReadTagResponse); +} +message ReadTagRequest { + string server_id = 1; + string addr = 2; + string data_type = 3; +} +message ReadTagResponse { + string value = 1; + string opc_status_code = 2; + string source_timestamp = 3; + string server_timestamp = 4; + bool persisted = 5; +} diff --git a/ingestor/proto/opc_gateway_pb2.py b/ingestor/proto/opc_gateway_pb2.py new file mode 100644 index 0000000..0a5f08c --- /dev/null +++ b/ingestor/proto/opc_gateway_pb2.py @@ -0,0 +1,40 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# NO CHECKED-IN PROTOBUF GENCODE +# source: opc_gateway.proto +# Protobuf Python Version: 7.35.0 +"""Generated protocol buffer code.""" +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import runtime_version as _runtime_version +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder +_runtime_version.ValidateProtobufRuntimeVersion( + _runtime_version.Domain.PUBLIC, + 7, + 35, + 0, + '', + 'opc_gateway.proto' +) +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x11opc_gateway.proto\x12\nopcgateway\"D\n\x0eReadTagRequest\x12\x11\n\tserver_id\x18\x01 \x01(\t\x12\x0c\n\x04\x61\x64\x64r\x18\x02 \x01(\t\x12\x11\n\tdata_type\x18\x03 \x01(\t\"\x80\x01\n\x0fReadTagResponse\x12\r\n\x05value\x18\x01 \x01(\t\x12\x17\n\x0fopc_status_code\x18\x02 \x01(\t\x12\x18\n\x10source_timestamp\x18\x03 \x01(\t\x12\x18\n\x10server_timestamp\x18\x04 \x01(\t\x12\x11\n\tpersisted\x18\x05 \x01(\x08\x32P\n\nOpcGateway\x12\x42\n\x07ReadTag\x12\x1a.opcgateway.ReadTagRequest\x1a\x1b.opcgateway.ReadTagResponseb\x06proto3') + +_globals = globals() +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'opc_gateway_pb2', _globals) +if not _descriptor._USE_C_DESCRIPTORS: + DESCRIPTOR._loaded_options = None + _globals['_READTAGREQUEST']._serialized_start=33 + _globals['_READTAGREQUEST']._serialized_end=101 + _globals['_READTAGRESPONSE']._serialized_start=104 + _globals['_READTAGRESPONSE']._serialized_end=232 + _globals['_OPCGATEWAY']._serialized_start=234 + _globals['_OPCGATEWAY']._serialized_end=314 +# @@protoc_insertion_point(module_scope) diff --git a/ingestor/proto/opc_gateway_pb2_grpc.py b/ingestor/proto/opc_gateway_pb2_grpc.py new file mode 100644 index 0000000..1f8ac2d --- /dev/null +++ b/ingestor/proto/opc_gateway_pb2_grpc.py @@ -0,0 +1,97 @@ +# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! +"""Client and server classes corresponding to protobuf-defined services.""" +import grpc +import warnings + +from . import opc_gateway_pb2 as opc__gateway__pb2 + +GRPC_GENERATED_VERSION = '1.82.1' +GRPC_VERSION = grpc.__version__ +_version_not_supported = False + +try: + from grpc._utilities import first_version_is_lower + _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION) +except ImportError: + _version_not_supported = True + +if _version_not_supported: + raise RuntimeError( + f'The grpc package installed is at version {GRPC_VERSION},' + + ' but the generated code in opc_gateway_pb2_grpc.py depends on' + + f' grpcio>={GRPC_GENERATED_VERSION}.' + + f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}' + + f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.' + ) + + +class OpcGatewayStub: + """Missing associated documentation comment in .proto file.""" + + def __init__(self, channel): + """Constructor. + + Args: + channel: A grpc.Channel. + """ + self.ReadTag = channel.unary_unary( + '/opcgateway.OpcGateway/ReadTag', + request_serializer=opc__gateway__pb2.ReadTagRequest.SerializeToString, + response_deserializer=opc__gateway__pb2.ReadTagResponse.FromString, + _registered_method=True) + + +class OpcGatewayServicer: + """Missing associated documentation comment in .proto file.""" + + def ReadTag(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + +def add_OpcGatewayServicer_to_server(servicer, server): + rpc_method_handlers = { + 'ReadTag': grpc.unary_unary_rpc_method_handler( + servicer.ReadTag, + request_deserializer=opc__gateway__pb2.ReadTagRequest.FromString, + response_serializer=opc__gateway__pb2.ReadTagResponse.SerializeToString, + ), + } + generic_handler = grpc.method_handlers_generic_handler( + 'opcgateway.OpcGateway', rpc_method_handlers) + server.add_generic_rpc_handlers((generic_handler,)) + server.add_registered_method_handlers('opcgateway.OpcGateway', rpc_method_handlers) + + + # This class is part of an EXPERIMENTAL API. +class OpcGateway: + """Missing associated documentation comment in .proto file.""" + + @staticmethod + def ReadTag(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/opcgateway.OpcGateway/ReadTag', + opc__gateway__pb2.ReadTagRequest.SerializeToString, + opc__gateway__pb2.ReadTagResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) diff --git a/pyproject.toml b/pyproject.toml index e4841ad..163b85f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,6 +23,7 @@ exclude = [ "*.pyc", ".pytest_cache", "htmlcov", + "ingestor/proto/*_pb2*.py", ] [tool.ruff.lint] diff --git a/tests/unit/proto/__init__.py b/tests/unit/proto/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/proto/test_proto.py b/tests/unit/proto/test_proto.py new file mode 100644 index 0000000..de8a319 --- /dev/null +++ b/tests/unit/proto/test_proto.py @@ -0,0 +1,36 @@ +from ingestor.proto import opc_gateway_pb2, opc_gateway_pb2_grpc + + +def test_read_tag_request_roundtrip(): + req = opc_gateway_pb2.ReadTagRequest(server_id='s1', addr='ns=2;s=Tag1', data_type='Float') + + data = req.SerializeToString() + parsed = opc_gateway_pb2.ReadTagRequest() + parsed.ParseFromString(data) + + assert parsed.server_id == 's1' + assert parsed.addr == 'ns=2;s=Tag1' + assert parsed.data_type == 'Float' + + +def test_read_tag_response_roundtrip(): + resp = opc_gateway_pb2.ReadTagResponse( + value='42.0', + opc_status_code='Good', + source_timestamp='2026-07-14T00:00:00Z', + server_timestamp='2026-07-14T00:00:01Z', + persisted=True, + ) + + data = resp.SerializeToString() + parsed = opc_gateway_pb2.ReadTagResponse() + parsed.ParseFromString(data) + + assert parsed.value == '42.0' + assert parsed.opc_status_code == 'Good' + assert parsed.persisted is True + + +def test_grpc_stub_module_loads(): + assert hasattr(opc_gateway_pb2_grpc, 'OpcGatewayStub') + assert hasattr(opc_gateway_pb2_grpc, 'OpcGatewayServicer') From 52223e6c09b1adaa06a3dfcefebdfee5f90e8f2e Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 15 Jul 2026 08:39:54 -0300 Subject: [PATCH 11/17] feat(gateway): gRPC server ReadTag (C4) --- ingestor/adapters/grpc_server.py | 68 ++++++++++++++++++ tests/unit/adapters/test_grpc_server.py | 96 +++++++++++++++++++++++++ 2 files changed, 164 insertions(+) create mode 100644 ingestor/adapters/grpc_server.py create mode 100644 tests/unit/adapters/test_grpc_server.py diff --git a/ingestor/adapters/grpc_server.py b/ingestor/adapters/grpc_server.py new file mode 100644 index 0000000..05cf6e8 --- /dev/null +++ b/ingestor/adapters/grpc_server.py @@ -0,0 +1,68 @@ +"""gRPC input adapter for on-demand tag reads (contract C4). + +Thin adapter: unpack the request, call ActiveReadEngine.read_now, map the +result onto ReadTagResponse. Internal-only for now -- no auth/TLS, per the +feature owner (no consumer exists yet, likely a frontend later). +""" + +import grpc +from sientia_do.observability.logger import Logger + +from ingestor.engines.active_read import ActiveReadEngine +from ingestor.proto import opc_gateway_pb2, opc_gateway_pb2_grpc + + +class ReadTagServicer(opc_gateway_pb2_grpc.OpcGatewayServicer): + def __init__(self, active_read: ActiveReadEngine, logger: Logger, slot_lookup=None): + self.active_read = active_read + self.logger = logger + # ponytail: slot_lookup wiring (address -> names from the current slot) + # lands in Task 18 (app wiring); optional callable keeps this servicer usable now. + self.slot_lookup = slot_lookup + + async def ReadTag(self, request, context): + names = None + if self.slot_lookup is not None: + names = self.slot_lookup(request.server_id, request.addr) + try: + result = await self.active_read.read_now(request.server_id, request.addr, names=names) + except Exception as e: + self.logger.error(f'ReadTag failed for {request.server_id}:{request.addr}: {e}') + await context.abort(grpc.StatusCode.INTERNAL, f'read failed: {e}') + return + return opc_gateway_pb2.ReadTagResponse( + value=str(result['value']), + opc_status_code=result['opc_status_code'] or '', + source_timestamp=result['source_timestamp'] or '', + server_timestamp=result['server_timestamp'] or '', + persisted=result['persisted'], + ) + + +class GrpcServer: + def __init__( + self, + active_read: ActiveReadEngine, + logger: Logger, + port: int = 50051, + slot_lookup=None, + ): + self.active_read = active_read + self.logger = logger + self.port = port + self.slot_lookup = slot_lookup + self.server = None + + async def start(self) -> None: + self.server = grpc.aio.server() + opc_gateway_pb2_grpc.add_OpcGatewayServicer_to_server( + ReadTagServicer(self.active_read, self.logger, self.slot_lookup), self.server + ) + self.server.add_insecure_port(f'[::]:{self.port}') + await self.server.start() + self.logger.info(f'GrpcServer started on port {self.port}') + + async def stop(self) -> None: + if self.server is None: + return + await self.server.stop(grace=5) diff --git a/tests/unit/adapters/test_grpc_server.py b/tests/unit/adapters/test_grpc_server.py new file mode 100644 index 0000000..2e542a2 --- /dev/null +++ b/tests/unit/adapters/test_grpc_server.py @@ -0,0 +1,96 @@ +from unittest.mock import AsyncMock, MagicMock + +import grpc +import pytest + +from ingestor.adapters.grpc_server import GrpcServer, ReadTagServicer +from ingestor.proto.opc_gateway_pb2 import ReadTagRequest + + +def make_request(server_id='srv1', addr='ns=2;i=5'): + return ReadTagRequest(server_id=server_id, addr=addr, data_type='float') + + +@pytest.mark.asyncio +async def test_read_tag_maps_full_result_onto_response(): + active_read = AsyncMock() + active_read.read_now.return_value = { + 'value': 1.5, + 'opc_status_code': 'Good', + 'source_timestamp': '2026-07-15T10:00:00-03:00', + 'server_timestamp': '2026-07-15T10:00:01-03:00', + 'persisted': False, + } + servicer = ReadTagServicer(active_read, MagicMock()) + + response = await servicer.ReadTag(make_request(), AsyncMock()) + + assert response.value == '1.5' + assert response.opc_status_code == 'Good' + assert response.source_timestamp == '2026-07-15T10:00:00-03:00' + assert response.server_timestamp == '2026-07-15T10:00:01-03:00' + assert response.persisted is False + + +@pytest.mark.asyncio +async def test_read_tag_none_timestamps_become_empty_string(): + active_read = AsyncMock() + active_read.read_now.return_value = { + 'value': 1.5, + 'opc_status_code': None, + 'source_timestamp': None, + 'server_timestamp': None, + 'persisted': False, + } + servicer = ReadTagServicer(active_read, MagicMock()) + + response = await servicer.ReadTag(make_request(), AsyncMock()) + + assert response.opc_status_code == '' + assert response.source_timestamp == '' + assert response.server_timestamp == '' + + +@pytest.mark.asyncio +async def test_read_tag_aborts_with_internal_on_engine_failure(): + active_read = AsyncMock() + active_read.read_now.side_effect = RuntimeError('boom') + logger = MagicMock() + servicer = ReadTagServicer(active_read, logger) + context = AsyncMock() + context.abort.side_effect = grpc.RpcError('aborted') + + with pytest.raises(grpc.RpcError): + await servicer.ReadTag(make_request(), context) + + context.abort.assert_awaited_once() + args = context.abort.call_args.args + assert args[0] == grpc.StatusCode.INTERNAL + logger.error.assert_called_once() + + +@pytest.mark.asyncio +async def test_read_tag_uses_slot_lookup_names_when_provided(): + active_read = AsyncMock() + active_read.read_now.return_value = { + 'value': 1, + 'opc_status_code': 'Good', + 'source_timestamp': None, + 'server_timestamp': None, + 'persisted': True, + } + names = [{'collection': 'c1', 'name': 't1'}] + slot_lookup = MagicMock(return_value=names) + servicer = ReadTagServicer(active_read, MagicMock(), slot_lookup=slot_lookup) + + await servicer.ReadTag(make_request(server_id='srv1', addr='ns=2;i=5'), AsyncMock()) + + slot_lookup.assert_called_once_with('srv1', 'ns=2;i=5') + active_read.read_now.assert_awaited_once_with('srv1', 'ns=2;i=5', names=names) + + +@pytest.mark.asyncio +async def test_grpc_server_stop_before_start_does_not_raise(): + server = GrpcServer(active_read=AsyncMock(), logger=MagicMock()) + + await server.stop() # no raise From e9ce9e50e2a31b193dedf29fe29646a09a9e5127 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 15 Jul 2026 08:50:59 -0300 Subject: [PATCH 12/17] SIENTIAPDE-1972: F3 close-out - ledger de commits + carry-forward notes (Task 16.end) --- docs/superpowers/plans/2026-07-13-opc-gateway.md | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/docs/superpowers/plans/2026-07-13-opc-gateway.md b/docs/superpowers/plans/2026-07-13-opc-gateway.md index 95df039..812e52a 100644 --- a/docs/superpowers/plans/2026-07-13-opc-gateway.md +++ b/docs/superpowers/plans/2026-07-13-opc-gateway.md @@ -91,7 +91,14 @@ At the end of each phase Fx, before starting the next: - **asyncua version hygiene:** requirements pin `asyncua==1.1.5` but the dev/test env runs 1.1.8. The DataValue/StatusCode_ behavior was verified identical on both, but align the pin (bump to 1.1.8) or the env during F4 (Task 21) so version-specific bugs can't slip through. - **`_SESSION_BAD_NAMES` in write.py** duplicates a subset of Laborious's reconnectable-error list (cross-repo) — move to `sientia_do` if it drifts (ponytail note in code). - **`_server_cfg` in write.py** is minimal (url/security off the cmd); full connector-DB lookup by `server_id` is Task 18 (F4 wiring) — ponytail note in code. -**F3 — Active reads + gRPC (C4):** _(fill at F3 close-out)_ +**F3 — Active reads + gRPC (C4)** — base `4855205`. Status: ✅ complete, 206 unit tests green, ruff clean. +- `99a715c` feat(gateway): ActiveReadEngine (scheduled + read_now) (Task 14) +- `1798b69` feat(gateway): gRPC proto + generated stubs (C4) (Task 15) +- `52223e6` feat(gateway): gRPC server ReadTag (C4) (Task 16) +- Carry-forward to Task 18 (wiring): + - Wire the gRPC `slot_lookup` seam (address→names from the current slot) so on-demand reads persist; if the lookup becomes I/O-bound, move it inside the servicer's try/except. + - gRPC server is internal-only, no auth (no consumer yet — likely frontend later, per feature owner). + - Edge case (minor): `read_now` with `SourceTimestamp=None` persists `timestamp=None`; add a guard if it matters in practice. **F4 — Wiring, health, deploy, hardening:** _(fill at F4 close-out)_ ### Final: open per-phase PRs (only after F4 is complete, tested, working) @@ -839,9 +846,9 @@ async def test_readtag_maps_response(): --- ### Task 16.end: F3 close-out (see [Phase Delivery](#phase-delivery--commit-ledger)) -- [ ] Full `pytest tests` green + `ruff` clean. -- [ ] Record F3 commit SHAs (base + range) in the [Commit Ledger](#commit-ledger). -- [ ] Do NOT open the PR yet. +- [x] Full `pytest tests` green (206) + `ruff` clean. +- [x] F3 commit SHAs recorded in the [Commit Ledger](#commit-ledger) (base `4855205`). +- [ ] PR intentionally NOT opened yet — opened only after F4 closes. --- From 6f4b04e8c0e1af815f153abc3c575483862dda41 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 22 Sep 2026 11:27:55 -0300 Subject: [PATCH 13/17] style: ruff format (new quality gate check picked up from main) --- ingestor/engines/active_read.py | 14 +++++++++--- ingestor/engines/write.py | 31 +++++++++++++++++++------- ingestor/io/data_sink.py | 6 ++++- tests/unit/engines/test_active_read.py | 4 +++- 4 files changed, 42 insertions(+), 13 deletions(-) diff --git a/ingestor/engines/active_read.py b/ingestor/engines/active_read.py index 9748517..f42b4f7 100644 --- a/ingestor/engines/active_read.py +++ b/ingestor/engines/active_read.py @@ -40,7 +40,11 @@ class ActiveReadEngine(SientiaMonitoring): ) async def read_now( - self, server_id: str, addr: str, names: list[dict] | None = None, server_cfg: dict | None = None + self, + server_id: str, + addr: str, + names: list[dict] | None = None, + server_cfg: dict | None = None, ) -> dict: """Read `addr` on `server_id` right now; persist via sink only if `names` given.""" tags = {'pod_id': self.pod_id, 'server_name': server_id} @@ -72,7 +76,9 @@ class ActiveReadEngine(SientiaMonitoring): source_str = self._format_ts(source_ts) persisted = bool(names) if persisted: - await self.sink.publish_reading(tag=addr, value=value, timestamp=source_str, names=names) + await self.sink.publish_reading( + tag=addr, value=value, timestamp=source_str, names=names + ) return { 'value': value, 'opc_status_code': status_name, @@ -96,7 +102,9 @@ class ActiveReadEngine(SientiaMonitoring): if source_ts is None: return lag = MonotonicClock.now_wall() - source_ts.replace(tzinfo=OPC_TIMEZONE).timestamp() - await self.emit_metric(metric_object=metrics.OPC_READ_LAG, method='set', value=lag, tags=tags) + await self.emit_metric( + metric_object=metrics.OPC_READ_LAG, method='set', value=lag, tags=tags + ) async def scheduled_tick(self, slot: dict) -> None: """Sweep every server/tag in a parsed slot, reading tags due for an active read.""" diff --git a/ingestor/engines/write.py b/ingestor/engines/write.py index 2f90d64..3b297d6 100644 --- a/ingestor/engines/write.py +++ b/ingestor/engines/write.py @@ -37,13 +37,24 @@ _VARIANT_TYPES = { # ponytail: same reconnect-worthy Bad*-status subset as laborious's # opc_repository.py (different repo, not importable here); keep in sync by # hand or promote to a shared lib if it drifts. -_SESSION_BAD_NAMES = frozenset({ - 'BadSessionIdInvalid', 'BadSessionClosed', 'BadSessionNotActivated', - 'BadSecureChannelIdInvalid', 'BadSecureChannelClosed', 'BadSecureChannelTokenUnknown', - 'BadTcpSecureChannelUnknown', 'BadServerNotConnected', 'BadConnectionClosed', - 'BadConnectionRejected', 'BadNotConnected', 'BadTimeout', 'BadRequestTimeout', - 'BadTooManySessions', -}) +_SESSION_BAD_NAMES = frozenset( + { + 'BadSessionIdInvalid', + 'BadSessionClosed', + 'BadSessionNotActivated', + 'BadSecureChannelIdInvalid', + 'BadSecureChannelClosed', + 'BadSecureChannelTokenUnknown', + 'BadTcpSecureChannelUnknown', + 'BadServerNotConnected', + 'BadConnectionClosed', + 'BadConnectionRejected', + 'BadNotConnected', + 'BadTimeout', + 'BadRequestTimeout', + 'BadTooManySessions', + } +) @dataclass @@ -174,7 +185,11 @@ class WriteEngine(SientiaMonitoring): metric_object=metrics.OPC_WRITE_STATUS_BY_MODEL, method='set', value=1, - tags={'pod_id': self.pod_id, 'model_id': cmd.get('model_id', 'unknown'), 'status': 'ok'}, + tags={ + 'pod_id': self.pod_id, + 'model_id': cmd.get('model_id', 'unknown'), + 'status': 'ok', + }, ) result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='') await self.sink.update_write_status(cmd, result) diff --git a/ingestor/io/data_sink.py b/ingestor/io/data_sink.py index eef8a50..290f0e4 100644 --- a/ingestor/io/data_sink.py +++ b/ingestor/io/data_sink.py @@ -57,7 +57,11 @@ class DataSink: f'UPDATE {schema}.{table} SET write_status = :status ' # noqa: S608 'WHERE timestamp = :timestamp AND model_id = :model_id' ) - params = {'status': result.status, 'timestamp': cmd['timestamp'], 'model_id': cmd['model_id']} + params = { + 'status': result.status, + 'timestamp': cmd['timestamp'], + 'model_id': cmd['model_id'], + } # ponytail: sync engine call inside async def — same pattern sientia_do's own # Postgres activities use (e.g. export_data_to_postgres); no async pg driver here. with self.postgres.engine.begin() as connection: diff --git a/tests/unit/engines/test_active_read.py b/tests/unit/engines/test_active_read.py index d880a9e..2a7640b 100644 --- a/tests/unit/engines/test_active_read.py +++ b/tests/unit/engines/test_active_read.py @@ -74,7 +74,9 @@ async def test_read_now_persists_and_returns_fields_when_names_given(): assert latency_call.kwargs['method'] == 'observe' assert latency_call.kwargs['value'] >= 0 lag_call = next( - c for c in eng.emit_metric.await_args_list if c.kwargs['metric_object'] is metrics.OPC_READ_LAG + c + for c in eng.emit_metric.await_args_list + if c.kwargs['metric_object'] is metrics.OPC_READ_LAG ) assert lag_call.kwargs['method'] == 'set' From 4f4037b4ae7443ca005578724afcde69d529f830 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 22 Sep 2026 11:42:24 -0300 Subject: [PATCH 14/17] fix(gateway): satisfy mypy on nats/grpc adapters and generated proto code - nats_consumer/grpc_server: annotate nc/sub/server attrs (were inferred as None-only from __init__, breaking on the real assignment later). - active_read: narrow on 'if names:' instead of the derived 'persisted' bool so mypy sees names as non-Optional inside the publish_reading call. - pyproject: exclude ingestor/proto/ (protoc/grpcio-tools output) from mypy's file walk and skip following it as an import - it's generated, not hand-maintained, and mypy can't see its runtime-built message classes. --- ingestor/adapters/grpc_server.py | 2 +- ingestor/adapters/nats_consumer.py | 6 ++++-- ingestor/engines/active_read.py | 2 +- pyproject.toml | 11 +++++++++++ 4 files changed, 17 insertions(+), 4 deletions(-) diff --git a/ingestor/adapters/grpc_server.py b/ingestor/adapters/grpc_server.py index 05cf6e8..ca2d5c2 100644 --- a/ingestor/adapters/grpc_server.py +++ b/ingestor/adapters/grpc_server.py @@ -51,7 +51,7 @@ class GrpcServer: self.logger = logger self.port = port self.slot_lookup = slot_lookup - self.server = None + self.server: grpc.aio.Server | None = None async def start(self) -> None: self.server = grpc.aio.server() diff --git a/ingestor/adapters/nats_consumer.py b/ingestor/adapters/nats_consumer.py index 189220d..97ed2fe 100644 --- a/ingestor/adapters/nats_consumer.py +++ b/ingestor/adapters/nats_consumer.py @@ -9,6 +9,8 @@ here. No reconnect logic — nats-py handles reconnection internally. import json import nats +from nats.aio.client import Client +from nats.aio.subscription import Subscription from sientia_do.observability.logger import Logger from ingestor.engines.write import WriteEngine @@ -28,8 +30,8 @@ class NatsWriteConsumer: self.logger = logger self.subject = subject self.queue = queue - self.nc = None - self.sub = None + self.nc: Client | None = None + self.sub: Subscription | None = None async def start(self) -> None: self.nc = await nats.connect(servers=self.nats_url) diff --git a/ingestor/engines/active_read.py b/ingestor/engines/active_read.py index f42b4f7..6aedf7e 100644 --- a/ingestor/engines/active_read.py +++ b/ingestor/engines/active_read.py @@ -75,7 +75,7 @@ class ActiveReadEngine(SientiaMonitoring): await self._record_read_lag(source_ts, tags) source_str = self._format_ts(source_ts) persisted = bool(names) - if persisted: + if names: await self.sink.publish_reading( tag=addr, value=value, timestamp=source_str, names=names ) diff --git a/pyproject.toml b/pyproject.toml index 9b58646..fd3e168 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -81,6 +81,9 @@ warn_unused_ignores = false warn_no_return = true strict_equality = true ignore_missing_imports = true +# Generated protoc/grpcio-tools output - not a checking target, dynamic attrs +# mypy can't see (message classes built at runtime from DESCRIPTOR). +exclude = ["^ingestor/proto/"] # Ignore missing imports for external packages [[tool.mypy.overrides]] @@ -107,6 +110,14 @@ ignore_missing_imports = true module = "pandas.*" ignore_missing_imports = true +# Generated protoc/grpcio-tools output - not hand-maintained, dynamic attrs +# mypy can't see (message classes built at runtime from DESCRIPTOR). Skip +# following it entirely so importers see Any instead of false attr-defined errors. +[[tool.mypy.overrides]] +module = "ingestor.proto.*" +ignore_errors = true +follow_imports = "skip" + [tool.pytest.ini_options] testpaths = ["tests"] python_files = ["test_*.py"] From 673b13cc6e780e7329d22540fe9ac35bf3a57c0f Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 22 Sep 2026 12:02:29 -0300 Subject: [PATCH 15/17] fix(gateway): adapt DataValue construction to asyncua 2.x main's requirements.txt unpins asyncua (PR #30), so CI installs 2.0.1, where DataValue's field was renamed StatusCode_ -> StatusCode. The old kwarg made every OPC write raise TypeError at runtime - masked locally by a stale asyncua 1.0.6 in the dev conda env. Also uses ua.DateTime (a datetime subclass) and ua.UInt32 to satisfy the new stricter types, with no behavior change (verified via binary encode/decode round-trip). --- ingestor/engines/active_read.py | 2 +- ingestor/engines/write.py | 11 +++++------ tests/unit/engines/test_active_read.py | 2 +- tests/unit/engines/test_write.py | 8 ++++---- 4 files changed, 11 insertions(+), 12 deletions(-) diff --git a/ingestor/engines/active_read.py b/ingestor/engines/active_read.py index 6aedf7e..cb2cacd 100644 --- a/ingestor/engines/active_read.py +++ b/ingestor/engines/active_read.py @@ -89,7 +89,7 @@ class ActiveReadEngine(SientiaMonitoring): @staticmethod def _extract(dv): - status_name = dv.StatusCode_.name if dv.StatusCode_ is not None else None + status_name = dv.StatusCode.name if dv.StatusCode is not None else None return dv.Value.Value, status_name, dv.SourceTimestamp, dv.ServerTimestamp @staticmethod diff --git a/ingestor/engines/write.py b/ingestor/engines/write.py index 3b297d6..675cb74 100644 --- a/ingestor/engines/write.py +++ b/ingestor/engines/write.py @@ -7,7 +7,6 @@ to the server, a rate-limit floor between writes, and OPC error typing. import asyncio from dataclasses import dataclass -from datetime import datetime from asyncua import ua from asyncua.ua.uaerrors import UaStatusCodeError @@ -146,12 +145,12 @@ class WriteEngine(SientiaMonitoring): variant = ua.Variant(typed_value, self._variant_type(cmd['data_type'])) write_quality = cmd.get('write_quality', self.write_quality) write_timestamp = cmd.get('write_timestamp', self.write_timestamp) - # NOTE: DataValue.StatusCode_ default_factory populates a Good status - # even when omitted, so "off" must pass StatusCode_=None explicitly + # NOTE: DataValue.StatusCode default_factory populates a Good status + # even when omitted, so "off" must pass StatusCode=None explicitly # or the flag silently no-ops. - status_code = ua.StatusCode(ua.StatusCodes.Good) if write_quality else None - source_ts = datetime.fromisoformat(cmd['timestamp']) if write_timestamp else None - return ua.DataValue(variant, StatusCode_=status_code, SourceTimestamp=source_ts) + status_code = ua.StatusCode(ua.UInt32(ua.StatusCodes.Good)) if write_quality else None + source_ts = ua.DateTime.fromisoformat(cmd['timestamp']) if write_timestamp else None + return ua.DataValue(variant, StatusCode=status_code, SourceTimestamp=source_ts) def _is_session_bad(self, exc: Exception) -> bool: return isinstance(exc, UaStatusCodeError) and type(exc).__name__ in _SESSION_BAD_NAMES diff --git a/tests/unit/engines/test_active_read.py b/tests/unit/engines/test_active_read.py index 2a7640b..c6bb5b3 100644 --- a/tests/unit/engines/test_active_read.py +++ b/tests/unit/engines/test_active_read.py @@ -25,7 +25,7 @@ def make_dv(value=42.0, status='Good', source_ts=None, server_ts=None): server_ts = server_ts or source_ts return ua.DataValue( ua.Variant(value), - StatusCode_=ua.StatusCode(getattr(ua.StatusCodes, status)), + StatusCode=ua.StatusCode(getattr(ua.StatusCodes, status)), SourceTimestamp=source_ts, ServerTimestamp=server_ts, ) diff --git a/tests/unit/engines/test_write.py b/tests/unit/engines/test_write.py index 6fc1963..ad54c30 100644 --- a/tests/unit/engines/test_write.py +++ b/tests/unit/engines/test_write.py @@ -108,8 +108,8 @@ async def test_write_defaults_carry_statuscode_and_timestamp(): await eng.write(base_cmd()) dv = node.write_value.await_args.args[0] - assert dv.StatusCode_ is not None - assert dv.StatusCode_.is_good() + assert dv.StatusCode is not None + assert dv.StatusCode.is_good() assert dv.SourceTimestamp == datetime.fromisoformat('2026-07-13T12:00:00Z') @@ -122,7 +122,7 @@ async def test_write_engine_flags_off_omit_statuscode_and_timestamp(): await eng.write(base_cmd()) dv = node.write_value.await_args.args[0] - assert dv.StatusCode_ is None + assert dv.StatusCode is None assert dv.SourceTimestamp is None @@ -137,7 +137,7 @@ async def test_per_message_flag_override_disables_timestamp(): dv = node.write_value.await_args.args[0] assert dv.SourceTimestamp is None - assert dv.StatusCode_ is not None # write_quality untouched, stays on + assert dv.StatusCode is not None # write_quality untouched, stays on @pytest.mark.asyncio From 54408d616064a16b97b2e4521d86f9c367d0b3c0 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 22 Sep 2026 15:59:36 -0300 Subject: [PATCH 16/17] fix(gateway): suppress bandit B608 false positive in update_write_status schema/table are allowlist-validated via _validate_identifier() before interpolation (see docstring above), not a real SQL injection vector. The existing '# noqa: S608' only silences ruff; bandit runs as its own CI step and needs its own '# nosec B608' suppression syntax. --- ingestor/io/data_sink.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ingestor/io/data_sink.py b/ingestor/io/data_sink.py index 290f0e4..c5806bc 100644 --- a/ingestor/io/data_sink.py +++ b/ingestor/io/data_sink.py @@ -54,7 +54,7 @@ class DataSink: # schema/table are allowlist-validated identifiers above; status/timestamp/model_id # are bound params below, never interpolated — not a real injection vector. stmt = text( - f'UPDATE {schema}.{table} SET write_status = :status ' # noqa: S608 + f'UPDATE {schema}.{table} SET write_status = :status ' # noqa: S608 # nosec B608 'WHERE timestamp = :timestamp AND model_id = :model_id' ) params = { From ff9339d408d874f73af02c5bc94efeaf68a80205 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 22 Sep 2026 16:57:29 -0300 Subject: [PATCH 17/17] fix(gateway): address SonarQube PR-mode findings on F2/F3 code - data_sink.py: _IDENTIFIER_RE uses \w+/re.ASCII instead of the equivalent-but-verbose [A-Za-z0-9_]+ character class (verified same match behavior). NOSONAR on update_write_status's 'async with no await inside' - deliberate, per the ponytail comment already in the method body (uniform async interface across DataSink, no async pg driver). - test_grpc_server.py: move make_request() out of the pytest.raises block so exactly one call can raise the expected exception. - sonar-project.properties: exclude ingestor/proto/ (generated grpc code) from analysis - its ReadTag/etc. names are dictated by grpcio-tools, not a style choice, and renaming them breaks the wire. --- ingestor/io/data_sink.py | 4 ++-- sonar-project.properties | 4 ++++ tests/unit/adapters/test_grpc_server.py | 3 ++- 3 files changed, 8 insertions(+), 3 deletions(-) diff --git a/ingestor/io/data_sink.py b/ingestor/io/data_sink.py index c5806bc..633e1df 100644 --- a/ingestor/io/data_sink.py +++ b/ingestor/io/data_sink.py @@ -6,7 +6,7 @@ import ingestor.metrics as metrics # fullmatch (no ^/$ anchors) so a trailing newline can't slip through — `$` would # match before a trailing '\n', which is not what we want for a SQL identifier. -_IDENTIFIER_RE = re.compile(r'[A-Za-z0-9_]+') +_IDENTIFIER_RE = re.compile(r'\w+', re.ASCII) class DataSink: @@ -42,7 +42,7 @@ class DataSink: }, ) - async def update_write_status(self, cmd: dict, result) -> None: + async def update_write_status(self, cmd: dict, result) -> None: # NOSONAR """UPDATE the per-model prediction row's write_status (C3), by timestamp+model_id. schema/table_name are per-model SQL identifiers, not values — they can't be diff --git a/sonar-project.properties b/sonar-project.properties index fc43878..5c832c5 100644 --- a/sonar-project.properties +++ b/sonar-project.properties @@ -8,3 +8,7 @@ sonar.python.coverage.reportPaths=coverage.xml sonar.python.xunit.reportPath=pytest.xml sonar.python.version=3.11 sonar.coverage.exclusions=ingestor/app.py +# Generated protoc/grpcio-tools output - not hand-maintained; renaming its +# ReadTag/etc. methods to satisfy naming-convention rules would break gRPC +# wire compatibility (grpcio-tools dictates these names, not style choice). +sonar.exclusions=ingestor/proto/** diff --git a/tests/unit/adapters/test_grpc_server.py b/tests/unit/adapters/test_grpc_server.py index 2e542a2..a5115f6 100644 --- a/tests/unit/adapters/test_grpc_server.py +++ b/tests/unit/adapters/test_grpc_server.py @@ -60,8 +60,9 @@ async def test_read_tag_aborts_with_internal_on_engine_failure(): context = AsyncMock() context.abort.side_effect = grpc.RpcError('aborted') + request = make_request() with pytest.raises(grpc.RpcError): - await servicer.ReadTag(make_request(), context) + await servicer.ReadTag(request, context) context.abort.assert_awaited_once() args = context.abort.call_args.args