Upload files to "certs"

.github/workflows/deploy.yml

@@ -20,7 +20,7 @@ jobs:
       branch_name: ${{ github.event.pull_request.head.ref }}
       project_name: 'ingestor'
       chart_name: 'opc-ingestor'
-      env_list_path: '[".env"]'
+      env_list_path: '.env'
       update_version_in: '["pyproject"]'
       create_tarball: ${{ github.event_name == 'pull_request' }}
       mirror: true

.gitignore

@@ -181,10 +181,4 @@ git_log
 
 .git/
 .ruff_cache/
-.mypy_cache/
+.mypy_cache/
-
-# Local planning/validation notes (not committed, not team docs)
-docs/superpowers/
-docs/e2e-test-guide.md
-docs/deploy-notes.md
-docs/HANDOFF.md

certs/opc-ingestor-cert.pem

@@ -0,0 +1,23 @@
+-----BEGIN CERTIFICATE-----
+MIID0zCCArugAwIBAgIUW3zWE3LXrXXSwmWnqDZWlx1GIrowDQYJKoZIhvcNAQEL
+BQAwQzEdMBsGA1UEAwwUc2llbnRpYS1vcGMtaW5nZXN0b3IxEDAOBgNVBAoMB0Fp
+Z25vc2kxEDAOBgNVBAsMB1NpZW50aWEwHhcNMjYwNzE1MjAwMDI3WhcNMzYwNzEy
+MjAwMDI3WjBDMR0wGwYDVQQDDBRzaWVudGlhLW9wYy1pbmdlc3RvcjEQMA4GA1UE
+CgwHQWlnbm9zaTEQMA4GA1UECwwHU2llbnRpYTCCASIwDQYJKoZIhvcNAQEBBQAD
+ggEPADCCAQoCggEBALUoKgKJTlfpWqraCoi4f0GgAtmgQcIu+VD+7Y89PSNKx+bk
+HCoe67nEHnFHXEhtKVxprQ1/001bQjyrKjZOIjd6G2WvYfYP0FROa67P+3zzQ1sY
+/mkQBvV6LdZOy6tk9Amk8QYVbVOXmacg8+0elAxeM2swl+02W9yuTcOpTGNLA7bW
+vTZrFL9DqPcghdeA6ZlD9AuUGaYvj7C/ShR+ELTHblUQO+f4xxpEXDv6w/8UaUaW
+xZMJtWbBO4fX0DIvMtCBi28oxGD2obbTVXHbgKDzuwqPh5qP4IPJJQEg4M8z6ttf
+2/OfFOc1O8XcFRK9vji4TBJFbs5wTDE6cPogFJcCAwEAAaOBvjCBuzAdBgNVHQ4E
+FgQU64naOPKzlGirtomkUtclQOIjtMAwHwYDVR0jBBgwFoAU64naOPKzlGirtomk
+UtclQOIjtMAwOQYDVR0RBDIwMIYYdXJuOnNpZW50aWE6b3BjLWluZ2VzdG9yghRz
+aWVudGlhLW9wYy1pbmdlc3RvcjAOBgNVHQ8BAf8EBAMCBPAwIAYDVR0lAQH/BBYw
+FAYIKwYBBQUHAwIGCCsGAQUFBwMBMAwGA1UdEwEB/wQCMAAwDQYJKoZIhvcNAQEL
+BQADggEBAGM4GXMc0EzsF2D02HOV0ZW1Igr5ya+561K0Cpg1SmMyQMFdCSk3s53I
+OrB/hcUNQ/xtg3MeVNQSGeif/dijNWCVKN8o7dGvZU0gjm7j/ue2gIj53TjONw8K
+QdF8vxvSDWgRHYFtPB3DlmWrHZaIeYD7Q60tnG1aaaM4pyRF0vH90PJL+XA9Tz4f
+9Q5+TzyqMPs87+fPpWtI5YwAhpLYfRU/nbpiTEqYUn8L97oFCyn4fjOa7cN5Wil3
+kiJgexHb0GimUiiKJeKmXDGgwZdw/xol//oB2DjhYPNcaPXTI4cqVtnTjuPwSkSS
+ZnxDPAwl0ssqzil895bYcq3AHG/jrro=
+-----END CERTIFICATE-----

certs/opc-ingestor-key.pem

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQC1KCoCiU5X6Vqq
+2gqIuH9BoALZoEHCLvlQ/u2PPT0jSsfm5BwqHuu5xB5xR1xIbSlcaa0Nf9NNW0I8
+qyo2TiI3ehtlr2H2D9BUTmuuz/t880NbGP5pEAb1ei3WTsurZPQJpPEGFW1Tl5mn
+IPPtHpQMXjNrMJftNlvcrk3DqUxjSwO21r02axS/Q6j3IIXXgOmZQ/QLlBmmL4+w
+v0oUfhC0x25VEDvn+McaRFw7+sP/FGlGlsWTCbVmwTuH19AyLzLQgYtvKMRg9qG2
+01Vx24Cg87sKj4eaj+CDySUBIODPM+rbX9vznxTnNTvF3BUSvb44uEwSRW7OcEwx
+OnD6IBSXAgMBAAECggEAJUF7UkY2rHG7ScPJwtBeq3pi26rPm0XZ1ER5SqoHdsic
+T+m9uRPe2rdwo09PtT1t8ilxg4M0S4ZnGt8AsWZkNxCPrdvhw58A8zWlg7AWQl4t
+DgEX4TeAlt9nqdlOnZMHdSxMvTmO4FAfM3XddHiDbf+5h2PY+tnhDoNyTrj0kr6z
+8JCD35FuOdU1zOBmum7T4eJDK1ol7XZ+UdrPraeM/jqDKaAnutUkRaDCpxUdaR7m
+RKG9XoqC0mw6t/esPbSbA4gUDpojgle69QXx98M/EhTzE3Q4bSGsjT9G5q3/6rHB
+RWgC2dWsTJOOI2y8Eym2MFLtf0MksZ4cgBtI0zW9AQKBgQDjjnVHSXC4LZ/Lbzdd
+bAE0lzTjpoNMKLyL+spWzbnY6kYhq79XC2lb8C1aor9P9Gkq+uUSqyt0yZ0nEjB4
+baCiKHEwe9qmBqyLEc84d6rt0aYzOxzzG/UGzMyvl2lhpJmofAxjgw4vaG2IVMog
+qaQymfIPNCXIo0UdUNzDWnaTQQKBgQDLzPks/MtMcCVCReJT3LYGD4D4NDAxv0/q
++V0fLngKP3DmeLRbvqWyTQ6bCM6SH6HfOJYLLyK39DBnf0uPbolPH/iBuRqkl7Ez
+YxcM4EJMR3U+aQH8hH24RJupr16OT3T+bZXK1pdjzM9SqwXiVI1rq/kRokfXT3hC
+GBkhZrUp1wKBgQCJv5rO5G4Pj87DXzesncy9AStLA2lHmQFJ2y7RDQgLfwWlLEpP
+buEAfot16j7VL6baFbbFbkiIEePH2wLsY7BxePnL162sAy4/oTBHhmk6PUbWjFoh
+8DKLQz5ErOtl6DX7/9ugAZ18pUJo3KreytJSr04QVgjpQ6x66pA40XImAQKBgHLF
+4J5BT4ygq5sDGE76QVhwasVdPrcFctB11se0t82tpLtmxsvUNiSP2OXes9b/EX5Q
+s89goEDQuGHRApCoQ5PYAW+s+yAufZDfASAzO6MRqymWoZPAWbzohnGtXabvgIe4
+118AMwn1UdzQi+PGkk5M71vK9IKBMsymHaD94pDJAoGBANAbRSmWgJfZNsYiVV7u
+kQ6L5Y5aN2/LfHlqNbWFYot00olsUWPz2WUEvKXI2e3L6lmkGJFNREOlAsOlFwSH
+ZBOSX+9Fq6vg8CUHi8obDgWAnIelTCCzHWC3Ep/ATLZQct82wgYZxIYTt4yCtMQo
+R5c2DqKOYl8ZAjr3Zah5uwFI
+-----END PRIVATE KEY-----

ingestor/adapters/grpc_server.py

@@ -1,68 +0,0 @@
-"""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: grpc.aio.Server | None = 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)

ingestor/adapters/nats_consumer.py

@@ -1,55 +0,0 @@
-"""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 nats.aio.client import Client
-from nats.aio.subscription import Subscription
-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: Client | None = None
-        self.sub: Subscription | None = 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 as e:  # includes JSONDecodeError and UnicodeDecodeError
-            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()

ingestor/connection/certs.py

@@ -1,29 +0,0 @@
-from pathlib import Path
-
-COLLECTION_NAME = 'platform_certificates'
-
-
-class CertProvider:
-    """Downloads the platform's client cert + private key (Mongo metadata,
-    MinIO bytes) into a writable cache dir — pod root FS is read-only (§12),
-    so `cache_dir` must be an emptyDir/tmpfs mount."""
-
-    def __init__(self, mongo, minio, cache_dir: str, bucket: str, metadata: dict | None = None):
-        self.mongo = mongo
-        self.minio = minio
-        self.cache_dir = cache_dir
-        self.bucket = bucket
-        self.metadata = metadata or {}
-
-    async def fetch(self) -> tuple[str, str]:
-        doc = await self.mongo.find_one(COLLECTION_NAME, {}, self.metadata)
-        Path(self.cache_dir).mkdir(parents=True, exist_ok=True)
-        cert_path = await self._download(doc['client_cert']['minio_object_name'])
-        key_path = await self._download(doc['private_key']['minio_object_name'])
-        return cert_path, key_path
-
-    async def _download(self, object_name: str) -> str:
-        data = await self.minio.download_file(object_name, self.bucket, self.metadata)
-        local_path = str(Path(self.cache_dir) / Path(object_name).name)
-        Path(local_path).write_bytes(data)
-        return local_path

ingestor/connection/clock.py

@@ -1,19 +0,0 @@
-import time
-
-
-class MonotonicClock:
-    """Single monotonic reference for latency/timeout math (1879 §2).
-
-    Wall-clock (`now_wall`) is only for human-facing timestamps; all elapsed/
-    timeout math must use `elapsed()` so NTP re-sync can't produce negatives.
-    """
-
-    def __init__(self):
-        self._start = time.monotonic()
-
-    def elapsed(self) -> float:
-        return time.monotonic() - self._start
-
-    @staticmethod
-    def now_wall() -> float:
-        return time.time()

ingestor/connection/pool.py

@@ -1,134 +0,0 @@
-import asyncio
-
-from asyncua import Client
-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.certs import CertProvider
-from ingestor.connection.clock import MonotonicClock
-from ingestor.connection.security import apply_security
-
-
-class OpcConnectionPool(SientiaMonitoring):
-    """Per-server OPC UA client cache (F0 capstone): reuses live connections,
-    evicts idle ones, and reconnects after a failed attempt (never caches it)."""
-
-    def __init__(
-        self,
-        cert_provider: CertProvider,
-        logger: Logger,
-        notification_handler: NotificationHandler,
-        metrics_controller: MetricsController,
-    ):
-        self.cert_provider = cert_provider
-        self.clock = MonotonicClock()
-        self._connections: dict[str, tuple[Client, float]] = {}
-        # ponytail: one Lock per server_id, never cleaned up — server set is
-        # small and bounded. Add eviction of stale locks only if it grows.
-        self._locks: dict[str, asyncio.Lock] = {}
-
-        SientiaMonitoring.__init__(
-            self,
-            logger=logger,
-            notification_handler=notification_handler,
-            metrics_controller=metrics_controller,
-        )
-
-    async def get(self, server_id: str, server_cfg: dict) -> Client:
-        """Return the cached client for `server_id`, connecting one if needed.
-
-        Double-checked under a per-server lock so concurrent callers for the
-        same new server share one connection (no leaked, untracked sessions)."""
-        client = self._cached(server_id)
-        if client is not None:
-            return client
-        async with self._locks.setdefault(server_id, asyncio.Lock()):
-            client = self._cached(server_id)  # re-check: a racer may have connected
-            if client is not None:
-                return client
-            return await self._open(server_id, server_cfg)
-
-    def _cached(self, server_id: str) -> Client | None:
-        entry = self._connections.get(server_id)
-        if entry is None:
-            return None
-        client, _ = entry
-        self._connections[server_id] = (client, self.clock.elapsed())
-        return client
-
-    async def _open(self, server_id: str, server_cfg: dict) -> Client:
-        try:
-            client = await self._connect(server_cfg)
-        except Exception:
-            await self.emit_metric(
-                metric_object=metrics.OPC_CONNECTIONS_FAILED,
-                tags={'pod_id': self.pod_id, 'server_name': server_id},
-            )
-            raise
-
-        self._connections[server_id] = (client, self.clock.elapsed())
-        await self._record_connected(server_id, server_cfg['url'])
-        return client
-
-    async def _connect(self, server_cfg: dict) -> Client:
-        client = Client(server_cfg['url'], timeout=10)
-        server_uri = server_cfg.get('server_uri')
-        if server_uri:
-            client.application_uri = server_uri
-        if server_cfg.get('use_security'):
-            cert, key = await self.cert_provider.fetch()
-            await apply_security(client, server_cfg.get('server_uri'), cert, key, None)
-        await client.connect()
-        return client
-
-    async def _record_connected(self, server_id: str, url: str):
-        await self.emit_metric(
-            metric_object=metrics.OPC_CONNECTIONS_TOTAL,
-            tags={'pod_id': self.pod_id, 'server_name': server_id},
-        )
-        await self.emit_metric(
-            metric_object=metrics.OPC_CONNECTION_STATUS,
-            method='set',
-            value=1,
-            tags={'pod_id': self.pod_id, 'server_name': server_id, 'server_url': url},
-        )
-
-    async def release_idle(self, max_idle_s: float):
-        """Disconnect and drop clients unused for longer than `max_idle_s`."""
-        now = self.clock.elapsed()
-        idle_ids = [
-            sid for sid, (_, last_used) in self._connections.items() if now - last_used > max_idle_s
-        ]
-        for server_id in idle_ids:
-            await self._evict(server_id)
-
-    async def close_all(self):
-        """Disconnect every pooled client (e.g. on shutdown)."""
-        # list() snapshots the keys: _evict() pops from self._connections,
-        # so iterating the dict directly would raise "changed size during iteration".
-        for server_id in list(self._connections):  # NOSONAR
-            await self._evict(server_id)
-
-    async def _evict(self, server_id: str):
-        client, _ = self._connections.pop(server_id)
-        await self._disconnect_with_retries(client, server_id)
-        await self.emit_metric(
-            metric_object=metrics.OPC_DISCONNECTIONS_TOTAL,
-            tags={'pod_id': self.pod_id, 'server_name': server_id},
-        )
-
-    async def _disconnect_with_retries(self, client: Client, server_id: str) -> None:
-        """Tries 5 times to disconnect, with a delay of 100ms x try (mirrors
-        OpcManager.disconnection_fallback)."""
-        for i in range(5):
-            try:
-                await client.disconnect()
-                return
-            except Exception as e:
-                self.logger.error(
-                    f'Failed to disconnect from {server_id}, attempt {i + 1} of 5: {e}'
-                )
-                await asyncio.sleep(0.1 * i)

ingestor/connection/security.py

@@ -1,17 +0,0 @@
-from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256
-
-SESSION_TIMEOUT_MS = 600_000  # 10 min; tune per server (Samarco: long sessions degrade)
-
-
-async def apply_security(client, server_uri, cert_path, private_key_path, server_cert_path):
-    """Apply Basic256Sha256 + certificate auth (1879 §1) and set channel > session
-    lifetime (1879 §3: secure_channel_timeout = session_timeout * 1.25)."""
-    client.application_uri = server_uri
-    await client.set_security(
-        SecurityPolicyBasic256Sha256,
-        certificate=cert_path,
-        private_key=private_key_path,
-        server_certificate=server_cert_path,
-    )
-    client.session_timeout = SESSION_TIMEOUT_MS
-    client.secure_channel_timeout = int(SESSION_TIMEOUT_MS * 1.25)

ingestor/converters.py

@@ -1,44 +0,0 @@
-"""Safe coercion of JSON-transported values into typed OPC write values.
-
-Fixes 1879 BUG-1 (bool('false') is True in Python) and BUG-2 (uncaught
-TypeError/ValueError from int/float on bad input).
-"""
-
-_FALSE_STRINGS = {'false', '0', 'no', 'off', ''}
-
-
-class ConversionError(ValueError):
-    """Raised when a raw value cannot be coerced to the requested type."""
-
-
-def _to_bool(raw) -> bool:
-    if isinstance(raw, bool):
-        return raw
-    if isinstance(raw, (int, float)):
-        return bool(raw)
-    return str(raw).strip().lower() not in _FALSE_STRINGS
-
-
-def to_typed_value(raw, data_type: str):
-    """Coerce raw (often a JSON string) to the given data_type.
-
-    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'):
-            return _to_bool(raw)
-        if dtype in ('int', 'integer'):
-            if isinstance(raw, str) and ('.' in raw or raw.strip() == ''):
-                raise ConversionError(f'invalid int: {raw!r}')
-            return int(raw)
-        if dtype in ('float', 'double', 'real'):
-            return float(raw)
-        if dtype in ('str', 'string'):
-            return str(raw)
-    except (TypeError, ValueError) as e:
-        raise ConversionError(f'cannot convert {raw!r} to {data_type}: {e}') from e
-    raise ConversionError(f'unknown data_type: {data_type!r}')

ingestor/engines/active_read.py

@@ -1,132 +0,0 @@
-"""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 names:
-            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

ingestor/engines/subscription.py

@@ -1,192 +0,0 @@
-import json
-
-from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
-from sientia_do.notifications.models import NotificationLevel
-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 SubscriptionEngine(SientiaMonitoring):
-    """Adapts OpcManager's subscription/datachange path (F1 capstone) to the
-    shared OpcConnectionPool and the multi-name slot v2 structure: on data
-    change it fans one reading out to every configured name via DataSink,
-    instead of the old flat per-`topics` publish loop."""
-
-    def __init__(
-        self,
-        pool: OpcConnectionPool,
-        sink: DataSink,
-        logger: Logger,
-        notification_handler: NotificationHandler,
-        metrics_controller: MetricsController,
-        metadata: dict | None = None,
-    ):
-        self.pool = pool
-        self.sink = sink
-        self.metadata = metadata or {}
-        self.nodes: dict = {}
-        self.subscriptions: dict = {}
-        self._clients: dict = {}
-        self._servers: dict = {}
-        self.non_receive_count = 0
-
-        SientiaMonitoring.__init__(
-            self,
-            logger=logger,
-            notification_handler=notification_handler,
-            metrics_controller=metrics_controller,
-        )
-
-    async def create_subscription(self, server_id: str, server_cfg: dict, name: str):
-        """Get a client for `server_id` from the pool and create a subscription
-        named `name` on it, with the engine itself as the datachange handler."""
-        client = await self.pool.get(server_id, server_cfg)
-        try:
-            self.subscriptions[name] = await client.create_subscription(
-                server_cfg['subscription_period_ms'], self
-            )
-        except Exception as e:
-            self.logger.error(f'Failed to create subscription {name} on {server_id}: {e}')
-            raise
-        self._clients[name] = client
-        self._servers[name] = server_id
-        self.logger.info(f'Subscription {name} created on {server_id}.')
-        await self.emit_metric(
-            metric_object=metrics.OPC_SUBSCRIPTIONS_CREATED,
-            method='inc',
-            value=1,
-            tags={'pod_id': self.pod_id, 'server_name': server_id, 'slot_name': name},
-        )
-
-    async def subscribe(self, subscription: str, nodes: dict, collect_period: int):
-        """Register `nodes` (slot v2 tags: frequency + names + active_read) under
-        `subscription` and start monitoring them for data changes."""
-        if not self.subscriptions.get(subscription):
-            raise ValueError('Subscription not created. Call create_subscription first.')
-        client = self._clients[subscription]
-        server_id = self._servers[subscription]
-        addr_nodes = [client.get_node(n) for n in nodes]
-        self.nodes.update(nodes)
-        self._init_cycle_rules(nodes, collect_period, server_id)
-
-        await self.subscriptions[subscription].subscribe_data_change(addr_nodes)
-        await self.emit_metric(
-            metric_object=metrics.OPC_TAGS_SUBSCRIBED,
-            method='set',
-            value=len(self.nodes),
-            tags={'pod_id': self.pod_id, 'server_name': server_id},
-        )
-
-    def _init_cycle_rules(self, nodes: dict, collect_period: int, server_id: str):
-        for node in nodes:
-            self.nodes[node]['server_name'] = server_id
-            self.nodes[node]['cycle_rule'] = {
-                'cycle_increment': collect_period * 1000 / float(self.nodes[node]['frequency']),
-                'cycle_count': 0,
-            }
-
-    async def unsubscribe(self, subscription: str):
-        if not self.subscriptions.get(subscription):
-            self.logger.warning(f"Subscription '{subscription}' not found. Cannot unsubscribe.")
-            return
-        await self.subscriptions[subscription].delete()
-        del self.subscriptions[subscription]
-        self._clients.pop(subscription, None)
-        self._servers.pop(subscription, None)
-        self.logger.info(f'Unsubscribed from {subscription}.')
-
-    async def datachange_notification(self, node, _val, data):
-        """asyncua subscription callback: fan the new value out to every name
-        configured for this address, then reset cycle/lag bookkeeping."""
-        value = data.monitored_item.Value.Value.Value
-        source_timestamp = data.monitored_item.Value.SourceTimestamp.replace(tzinfo=OPC_TIMEZONE)
-        tag = str(node)
-        config = self.nodes[tag]
-
-        await self.sink.publish_reading(
-            tag=tag,
-            value=value,
-            timestamp=source_timestamp.strftime(DATETIME_FORMAT_WITH_TZ),
-            names=config['names'],
-        )
-        await self._record_read_lag(config, source_timestamp)
-        await self._reset_cycle_state(config)
-
-    async def _record_read_lag(self, config: dict, source_timestamp):
-        lag_seconds = MonotonicClock.now_wall() - source_timestamp.timestamp()
-        await self.emit_metric(
-            metric_object=metrics.OPC_READ_LAG,
-            method='set',
-            value=lag_seconds,
-            tags={'pod_id': self.pod_id, 'server_name': config.get('server_name', 'unknown')},
-        )
-
-    async def _reset_cycle_state(self, config: dict):
-        config['cycle_rule']['cycle_count'] = 0
-        self.non_receive_count = 0
-        await self.emit_metric(
-            metric_object=metrics.OPC_CYCLES_WITHOUT_DATA,
-            method='set',
-            value=0,
-            tags={'pod_id': self.pod_id, 'server_name': config.get('server_name', 'unknown')},
-        )
-
-    async def check_cycles(self):
-        """Increment every node's cycle count and warn on nodes stuck >= 5 cycles."""
-        for node, config in self.nodes.items():
-            config['cycle_rule']['cycle_count'] += config['cycle_rule']['cycle_increment']
-            if config['cycle_rule']['cycle_count'] >= 5:
-                await self._notify_node_stopped(node, config)
-
-    async def _notify_node_stopped(self, node: str, config: dict):
-        cycles = config['cycle_rule']['cycle_count']
-        names = ','.join(n['tag_name'] for n in config['names'])
-        await self.send_notification_async(
-            metadata=self.metadata,
-            notification_id=f'TAG_{node}:{names}_LISTENNING_STOPPED',
-            message=f'{cycles} cycles without receive from {node}:{names}',
-            block='subscription_engine',
-            level=NotificationLevel.WARNING,
-        )
-
-    async def check_opc_listenning(self, server_name: str) -> bool:
-        """Track cycles without any datachange for `server_name`; warn at 5,
-        error + signal reconnect at 15. `server_name` is explicit (unlike
-        OpcManager's `self.name`) since one engine can serve several servers."""
-        self.non_receive_count += 1
-        await self.emit_metric(
-            metric_object=metrics.OPC_CYCLES_WITHOUT_DATA,
-            method='set',
-            value=self.non_receive_count,
-            tags={'pod_id': self.pod_id, 'server_name': server_name},
-        )
-        if self.non_receive_count >= 5:
-            await self.send_notification_async(
-                metadata=self.metadata,
-                notification_id=f'OPC_LISTENNING_STOPPED__{server_name}',
-                message=f'{self.non_receive_count} cycles without receive from OPC '
-                f'{server_name}. Tags: {json.dumps(self.nodes, default=str)}',
-                block='subscription_engine',
-                level=NotificationLevel.ERROR,
-            )
-        if self.non_receive_count >= 15:
-            await self.emit_metric(
-                metric_object=metrics.OPC_RECONNECTIONS_TOTAL,
-                tags={'pod_id': self.pod_id, 'server_name': server_name},
-            )
-            await self.send_notification_async(
-                metadata=self.metadata,
-                notification_id=f'OPC_CONNECTION_RETRY__{server_name}',
-                message=f'Retrying to connect to server {server_name}',
-                block='subscription_engine',
-                level=NotificationLevel.ERROR,
-            )
-            return True
-        return False

ingestor/engines/write.py

@@ -1,195 +0,0 @@
-"""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 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,
-        write_timestamp: bool = True,
-        write_quality: bool = True,
-        min_pause_s: float = 0.0,
-    ):
-        self.pool = pool
-        self.sink = sink
-        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()
-
-        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:
-            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:
-        # 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']))
-        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.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
-
-    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.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'},
-        )
-        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

ingestor/io/data_sink.py

@@ -1,74 +0,0 @@
-import re
-
-from sqlalchemy import text
-
-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'\w+', re.ASCII)
-
-
-class DataSink:
-    """Gateway egress wrapper: fans a reading out to every configured name/collection."""
-
-    def __init__(self, data_manager, postgres) -> None:
-        self.data_manager = data_manager
-        self.postgres = postgres
-
-    async def emit_metric(self, **kwargs) -> None:
-        # ponytail: delegate to data_manager, it's already a SientiaMonitoring
-        await self.data_manager.emit_metric(**kwargs)
-
-    async def publish_reading(self, tag: str, value, timestamp, names: list[dict]) -> None:
-        """Publish one MongoDB document per name entry and record its tag value metric."""
-        for name in names:
-            doc = {
-                'tag': tag,
-                'name': name['tag_name'],
-                'model_id': name['model_id'],
-                'timestamp': timestamp,
-                'value': value,
-            }
-            await self.data_manager.publish(name['collection'], doc)
-            await self.emit_metric(
-                metric_object=metrics.OPC_TAG_VALUE,
-                method='set',
-                value=value,
-                tags={
-                    'pod_id': self.data_manager.pod_id,
-                    'tag_name': name['tag_name'],
-                    'model_id': name['model_id'],
-                },
-            )
-
-    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
-        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 # nosec B608
-            '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 isinstance(name, str) or not _IDENTIFIER_RE.fullmatch(name):
-            raise ValueError(f'invalid SQL identifier: {name!r}')
-        return name

ingestor/io/slots.py

@@ -1,25 +0,0 @@
-"""Parse the slot v2 structure written by the Orchestrator to Redis (contract C1)."""
-
-SERVER_FIELDS = ('server_id', 'url', 'server_uri', 'use_security', 'subscription_period_ms')
-
-
-def _parse_tag(address: str, tag: dict) -> dict:
-    names = tag.get('names')
-    if not names:
-        raise ValueError(f'tag {address!r} has no names')
-    return {
-        'frequency': tag['frequency'],
-        'names': names,
-        'active_read': tag.get('active_read', {'enabled': False}),
-    }
-
-
-def _parse_server(server: dict) -> dict:
-    parsed = {field: server[field] for field in SERVER_FIELDS if field in server}
-    parsed['tags'] = {addr: _parse_tag(addr, tag) for addr, tag in server['tags'].items()}
-    return parsed
-
-
-def parse_slot(raw: dict) -> dict:
-    """Normalize a raw slot v2 dict into `{server_name: {..., tags: {addr: tag}}}`."""
-    return {name: _parse_server(server) for name, server in raw.items()}

ingestor/metrics.py

@@ -139,52 +139,3 @@ NOTIFICATIONS_SENT = Counter(
     'Total number of notifications sent',
     NOTIFICATION_LABELS,
 )
-
-# --- Gateway: Write metrics ---
-OPC_WRITE_TOTAL = Counter(
-    'opc_write_total',
-    'OPC active write attempts',
-    ['pod_id', 'server_name', 'model_id', 'result'],  # result: success|error
-)
-OPC_WRITE_LATENCY = Histogram(
-    'opc_write_latency_seconds',
-    'OPC active write latency',
-    ['pod_id', 'server_name'],
-)
-OPC_WRITE_STATUS_BY_MODEL = Gauge(
-    'opc_write_status_by_model',
-    'Last write status per model (1=ok,0=error)',
-    ['pod_id', 'model_id', 'status'],
-)
-# --- Gateway: Active read metrics ---
-OPC_ACTIVE_READ_TOTAL = Counter(
-    'opc_active_read_total',
-    'OPC active read attempts',
-    ['pod_id', 'server_name', 'result'],
-)
-OPC_ACTIVE_READ_LATENCY = Histogram(
-    'opc_active_read_latency_seconds',
-    'OPC active read latency',
-    ['pod_id', 'server_name'],
-)
-OPC_READ_LAG = Gauge(
-    'opc_read_lag_seconds',
-    'Lag between OPC source timestamp and machine time',
-    ['pod_id', 'server_name'],
-)
-OPC_TAG_VALUE = Gauge(
-    'opc_tag_value',
-    'Last value read per tag/model',
-    ['pod_id', 'tag_name', 'model_id'],
-)
-# --- Gateway: connection/error metrics ---
-OPC_DISCONNECTIONS_TOTAL = Counter(
-    'opc_disconnections_total',
-    'OPC disconnections',
-    ['pod_id', 'server_name'],
-)
-OPC_ERROR_BY_TYPE = Counter(
-    'opc_error_by_type_total',
-    'OPC errors by type',
-    ['pod_id', 'server_name', 'error_type'],
-)

ingestor/proto/opc_gateway.proto

@@ -1,18 +0,0 @@
-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;
-}

ingestor/proto/opc_gateway_pb2.py

@@ -1,40 +0,0 @@
-# -*- 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)

ingestor/proto/opc_gateway_pb2_grpc.py

@@ -1,97 +0,0 @@
-# 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)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "ingestor"
-version = "2.0.0-rc1"
+version = "1.4.4"
 description = "Sientia DataOps Ingestor - OPC Tag Ingestor"
 readme = "README.md"
 requires-python = ">=3.11"
@@ -23,7 +23,6 @@ exclude = [
     "*.pyc",
     ".pytest_cache",
     "htmlcov",
-    "ingestor/proto/*_pb2*.py",
 ]
 
 [tool.ruff.lint]
@@ -81,9 +80,6 @@ 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]]
@@ -110,14 +106,6 @@ 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"]

requirements.txt

@@ -1,11 +1,5 @@
 asyncua
 redis
-# TODO(SIENTIAPDE-1972): bump once sientia-dataops-library releases SIENTIAPDE-1590
-# (generic Mongo find_one/distinct/aggregate methods) — not in 1.12.2 yet.
-# Local dev uses an editable install of the local checkout (feature/SIENTIAPDE-1590).
 sientia_do==1.12.2
 prometheus_client
-pymongo
+pymongo
-nats-py>=2.6
-grpcio>=1.60
-grpcio-tools>=1.60

sonar-project.properties

@@ -8,7 +8,3 @@ 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/**

tests/unit/adapters/test_grpc_server.py

@@ -1,97 +0,0 @@
-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')
-
-    request = make_request()
-    with pytest.raises(grpc.RpcError):
-        await servicer.ReadTag(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

tests/unit/adapters/test_nats_consumer.py

@@ -1,76 +0,0 @@
-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()

tests/unit/connection/test_certs.py

@@ -1,28 +0,0 @@
-from unittest.mock import AsyncMock
-
-import pytest
-
-from ingestor.connection.certs import CertProvider
-
-
-@pytest.mark.asyncio
-async def test_fetch_returns_local_paths(tmp_path):
-    mongo = AsyncMock()
-    mongo.find_one.return_value = {
-        'client_cert': {'minio_object_name': 'platform/certificates/client_cert_c.pem'},
-        'private_key': {'minio_object_name': 'platform/certificates/private_key_k.pem'},
-    }
-    minio = AsyncMock()
-    minio.download_file.side_effect = [b'cert-bytes', b'key-bytes']
-
-    provider = CertProvider(mongo, minio, cache_dir=str(tmp_path), bucket='sientia-dataops')
-    cert, key = await provider.fetch()
-
-    assert cert.endswith('c.pem') and key.endswith('k.pem')
-    assert minio.download_file.await_count == 2
-    mongo.find_one.assert_awaited_once_with('platform_certificates', {}, {})
-    minio.download_file.assert_any_await(
-        'platform/certificates/client_cert_c.pem', 'sientia-dataops', {}
-    )
-    assert (tmp_path / 'client_cert_c.pem').read_bytes() == b'cert-bytes'
-    assert (tmp_path / 'private_key_k.pem').read_bytes() == b'key-bytes'

tests/unit/connection/test_clock.py

@@ -1,8 +0,0 @@
-from ingestor.connection.clock import MonotonicClock
-
-
-def test_elapsed_is_non_negative_and_monotonic():
-    clk = MonotonicClock()
-    a = clk.elapsed()
-    b = clk.elapsed()
-    assert a >= 0 and b >= a  # never negative even if wall clock jumps back

tests/unit/connection/test_pool.py

@@ -1,184 +0,0 @@
-import asyncio
-from unittest.mock import AsyncMock, MagicMock, patch
-
-import pytest
-
-import ingestor.metrics as metrics
-from ingestor.connection.pool import OpcConnectionPool
-
-
-def make_pool(cert_provider=None):
-    pool = OpcConnectionPool(
-        cert_provider=cert_provider or AsyncMock(),
-        logger=MagicMock(),
-        notification_handler=MagicMock(),
-        metrics_controller=AsyncMock(),
-    )
-    pool.emit_metric = AsyncMock()
-    return pool
-
-
-@pytest.mark.asyncio
-async def test_get_reuses_connection_per_server():
-    pool = make_pool()
-    fake_client = AsyncMock()
-    with patch('ingestor.connection.pool.Client', return_value=fake_client):
-        c1 = await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
-        c2 = await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
-
-    assert c1 is c2
-    assert fake_client.connect.await_count == 1
-
-
-@pytest.mark.asyncio
-async def test_get_failure_records_metric_and_raises():
-    pool = make_pool()
-    fake_client = AsyncMock()
-    fake_client.connect.side_effect = RuntimeError('boom')
-    with patch('ingestor.connection.pool.Client', return_value=fake_client):
-        with pytest.raises(RuntimeError):
-            await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
-
-    pool.emit_metric.assert_any_await(
-        metric_object=metrics.OPC_CONNECTIONS_FAILED,
-        tags={'pod_id': pool.pod_id, 'server_name': 'srv'},
-    )
-    assert 'srv' not in pool._connections
-
-    # a later successful get must reconnect (failed attempt was not cached)
-    fake_client2 = AsyncMock()
-    with patch('ingestor.connection.pool.Client', return_value=fake_client2):
-        client = await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
-    assert client is fake_client2
-    assert fake_client2.connect.await_count == 1
-
-
-@pytest.mark.asyncio
-async def test_get_applies_security_when_requested():
-    cert_provider = AsyncMock()
-    cert_provider.fetch.return_value = ('cert.pem', 'key.pem')
-    pool = make_pool(cert_provider=cert_provider)
-    fake_client = AsyncMock()
-    with (
-        patch('ingestor.connection.pool.Client', return_value=fake_client),
-        patch('ingestor.connection.pool.apply_security', new=AsyncMock()) as mock_apply,
-    ):
-        await pool.get(
-            'srv',
-            {'url': 'opc.tcp://h:4840', 'server_uri': 'urn:test', 'use_security': True},
-        )
-
-    cert_provider.fetch.assert_awaited_once()
-    mock_apply.assert_awaited_once_with(fake_client, 'urn:test', 'cert.pem', 'key.pem', None)
-
-
-@pytest.mark.asyncio
-async def test_release_idle_evicts_and_disconnects():
-    pool = make_pool()
-    fake_client = AsyncMock()
-    with patch('ingestor.connection.pool.Client', return_value=fake_client):
-        await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
-
-    client, _ = pool._connections['srv']
-    pool._connections['srv'] = (client, -1000.0)  # force idle
-
-    await pool.release_idle(max_idle_s=0)
-
-    fake_client.disconnect.assert_awaited_once()
-    assert 'srv' not in pool._connections
-    pool.emit_metric.assert_any_await(
-        metric_object=metrics.OPC_DISCONNECTIONS_TOTAL,
-        tags={'pod_id': pool.pod_id, 'server_name': 'srv'},
-    )
-
-
-@pytest.mark.asyncio
-async def test_release_idle_keeps_recently_used():
-    pool = make_pool()
-    fake_client = AsyncMock()
-    with patch('ingestor.connection.pool.Client', return_value=fake_client):
-        await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
-
-    await pool.release_idle(max_idle_s=3600)
-
-    fake_client.disconnect.assert_not_awaited()
-    assert 'srv' in pool._connections
-
-
-@pytest.mark.asyncio
-async def test_close_all_disconnects_everything():
-    pool = make_pool()
-    fake_a, fake_b = AsyncMock(), AsyncMock()
-    with patch('ingestor.connection.pool.Client', side_effect=[fake_a, fake_b]):
-        await pool.get('a', {'url': 'opc.tcp://h:4840', 'use_security': False})
-        await pool.get('b', {'url': 'opc.tcp://h:4840', 'use_security': False})
-
-    await pool.close_all()
-
-    fake_a.disconnect.assert_awaited_once()
-    fake_b.disconnect.assert_awaited_once()
-    assert pool._connections == {}
-
-
-@pytest.mark.asyncio
-async def test_release_idle_disconnect_retries_then_gives_up():
-    pool = make_pool()
-    fake_client = AsyncMock()
-    fake_client.disconnect.side_effect = RuntimeError('nope')
-    with patch('ingestor.connection.pool.Client', return_value=fake_client):
-        await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
-
-    client, _ = pool._connections['srv']
-    pool._connections['srv'] = (client, -1000.0)
-
-    await pool.release_idle(max_idle_s=0)
-
-    assert fake_client.disconnect.await_count == 5
-    assert 'srv' not in pool._connections
-
-
-@pytest.mark.asyncio
-async def test_success_path_emits_connection_metrics():
-    pool = make_pool()
-    fake_client = AsyncMock()
-    with patch('ingestor.connection.pool.Client', return_value=fake_client):
-        await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
-
-    pool.emit_metric.assert_any_await(
-        metric_object=metrics.OPC_CONNECTIONS_TOTAL,
-        tags={'pod_id': pool.pod_id, 'server_name': 'srv'},
-    )
-    pool.emit_metric.assert_any_await(
-        metric_object=metrics.OPC_CONNECTION_STATUS,
-        method='set',
-        value=1,
-        tags={'pod_id': pool.pod_id, 'server_name': 'srv', 'server_url': 'opc.tcp://h:4840'},
-    )
-
-
-@pytest.mark.asyncio
-async def test_concurrent_get_same_server_connects_once():
-    pool = make_pool()
-    started = asyncio.Event()
-    proceed = asyncio.Event()
-
-    clients = [AsyncMock(), AsyncMock()]
-
-    async def slow_connect():
-        started.set()
-        await proceed.wait()
-
-    clients[0].connect.side_effect = slow_connect
-
-    cfg = {'url': 'opc.tcp://h:4840', 'use_security': False}
-    with patch('ingestor.connection.pool.Client', side_effect=clients) as ctor:
-        t1 = asyncio.create_task(pool.get('srv', cfg))
-        await started.wait()  # t1 is mid-connect, holding the per-server lock
-        t2 = asyncio.create_task(pool.get('srv', cfg))
-        await asyncio.sleep(0)  # let t2 block on the lock (not build a 2nd Client)
-        proceed.set()
-        c1, c2 = await asyncio.gather(t1, t2)
-
-    assert c1 is c2 is clients[0]
-    assert ctor.call_count == 1
-    assert clients[0].connect.await_count == 1

tests/unit/connection/test_security.py

@@ -1,18 +0,0 @@
-from unittest.mock import AsyncMock, MagicMock
-
-import pytest
-
-from ingestor.connection.security import SESSION_TIMEOUT_MS, apply_security
-
-
-@pytest.mark.asyncio
-async def test_apply_security_sets_sha256_and_channel_lifetime():
-    client = MagicMock()
-    client.set_security = AsyncMock()
-    await apply_security(client, 'urn:x', 'c.pem', 'k.pem', None)
-    assert client.session_timeout == SESSION_TIMEOUT_MS
-    assert client.secure_channel_timeout == int(SESSION_TIMEOUT_MS * 1.25)
-    args, kwargs = client.set_security.call_args
-    from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256
-
-    assert args[0] is SecurityPolicyBasic256Sha256

tests/unit/engines/test_active_read.py

@@ -1,179 +0,0 @@
-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()

tests/unit/engines/test_subscription.py

@@ -1,203 +0,0 @@
-from datetime import datetime
-from unittest.mock import AsyncMock, MagicMock
-
-import pytest
-from sientia_do.temporal.constants import OPC_TIMEZONE
-
-import ingestor.metrics as metrics
-from ingestor.engines.subscription import SubscriptionEngine
-
-
-def make_engine():
-    eng = SubscriptionEngine(
-        pool=AsyncMock(),
-        sink=AsyncMock(),
-        logger=MagicMock(),
-        notification_handler=MagicMock(),
-        metrics_controller=AsyncMock(),
-        metadata={},
-    )
-    eng.emit_metric = AsyncMock()
-    eng.send_notification_async = AsyncMock()
-    return eng
-
-
-def make_node(name='ns=2;i=2'):
-    node = MagicMock()
-    node.__str__ = lambda self: name
-    return node
-
-
-def make_data(value=42, timestamp=None):
-    data = MagicMock()
-    data.monitored_item.Value.Value.Value = value
-    if timestamp is not None:
-        data.monitored_item.Value.SourceTimestamp = timestamp
-    return data
-
-
-@pytest.mark.asyncio
-async def test_datachange_fans_out():
-    eng = make_engine()
-    eng.nodes['ns=2;i=2'] = {
-        'names': [{'tag_name': 'a'}, {'tag_name': 'b'}],
-        'cycle_rule': {'cycle_increment': 1, 'cycle_count': 3},
-    }
-    node = make_node()
-    data = make_data(42)
-
-    await eng.datachange_notification(node, 42, data)
-
-    assert eng.sink.publish_reading.await_count == 1
-    _, kwargs = eng.sink.publish_reading.call_args
-    assert len(kwargs['names']) == 2
-    assert kwargs['tag'] == 'ns=2;i=2'
-    assert kwargs['value'] == 42
-
-
-@pytest.mark.asyncio
-async def test_datachange_resets_cycle_count_and_non_receive_count():
-    eng = make_engine()
-    eng.non_receive_count = 7
-    eng.nodes['ns=2;i=2'] = {
-        'names': [{'tag_name': 'a'}],
-        'cycle_rule': {'cycle_increment': 1, 'cycle_count': 3},
-    }
-
-    await eng.datachange_notification(make_node(), 42, make_data(42))
-
-    assert eng.nodes['ns=2;i=2']['cycle_rule']['cycle_count'] == 0
-    assert eng.non_receive_count == 0
-    eng.emit_metric.assert_any_await(
-        metric_object=metrics.OPC_CYCLES_WITHOUT_DATA,
-        method='set',
-        value=0,
-        tags={'pod_id': eng.pod_id, 'server_name': 'unknown'},
-    )
-
-
-@pytest.mark.asyncio
-async def test_datachange_emits_read_lag():
-    eng = make_engine()
-    eng.nodes['ns=2;i=2'] = {
-        'names': [{'tag_name': 'a'}],
-        'cycle_rule': {'cycle_increment': 1, 'cycle_count': 0},
-        'server_name': 'srv1',
-    }
-    source_ts = datetime.now(tz=OPC_TIMEZONE)
-
-    await eng.datachange_notification(make_node(), 42, make_data(42, timestamp=source_ts))
-
-    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['tags'] == {'pod_id': eng.pod_id, 'server_name': 'srv1'}
-    assert abs(lag_call.kwargs['value']) < 5
-
-
-@pytest.mark.asyncio
-async def test_create_subscription_uses_pool_client():
-    eng = make_engine()
-    client = AsyncMock()
-    eng.pool.get.return_value = client
-
-    await eng.create_subscription('srv1', {'subscription_period_ms': 500}, 'sub1')
-
-    eng.pool.get.assert_awaited_once_with('srv1', {'subscription_period_ms': 500})
-    client.create_subscription.assert_awaited_once_with(500, eng)
-    assert eng.subscriptions['sub1'] is not None
-    eng.emit_metric.assert_any_await(
-        metric_object=metrics.OPC_SUBSCRIPTIONS_CREATED,
-        method='inc',
-        value=1,
-        tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'slot_name': 'sub1'},
-    )
-
-
-@pytest.mark.asyncio
-async def test_subscribe_registers_nodes_with_cycle_rule():
-    eng = make_engine()
-    client = AsyncMock()
-    eng.pool.get.return_value = client
-    await eng.create_subscription('srv1', {'subscription_period_ms': 500}, 'sub1')
-
-    nodes = {
-        'ns=2;i=2': {
-            'frequency': 1000,
-            'names': [{'tag_name': 'a'}, {'tag_name': 'b'}],
-            'active_read': {'enabled': False},
-        }
-    }
-
-    await eng.subscribe('sub1', nodes, collect_period=1)
-
-    assert eng.nodes['ns=2;i=2']['server_name'] == 'srv1'
-    assert eng.nodes['ns=2;i=2']['cycle_rule']['cycle_count'] == 0
-    assert eng.nodes['ns=2;i=2']['cycle_rule']['cycle_increment'] == pytest.approx(1.0)
-    eng.subscriptions['sub1'].subscribe_data_change.assert_awaited_once()
-
-
-@pytest.mark.asyncio
-async def test_subscribe_without_subscription_raises():
-    eng = make_engine()
-    with pytest.raises(ValueError, match='Subscription not created'):
-        await eng.subscribe('sub1', {}, collect_period=1)
-
-
-@pytest.mark.asyncio
-async def test_unsubscribe_missing_logs_warning():
-    eng = make_engine()
-    await eng.unsubscribe('sub1')
-    eng.logger.warning.assert_called_once_with("Subscription 'sub1' not found. Cannot unsubscribe.")
-
-
-@pytest.mark.asyncio
-async def test_unsubscribe_deletes_subscription():
-    eng = make_engine()
-    client = AsyncMock()
-    eng.pool.get.return_value = client
-    await eng.create_subscription('srv1', {'subscription_period_ms': 500}, 'sub1')
-    sub = eng.subscriptions['sub1']
-
-    await eng.unsubscribe('sub1')
-
-    sub.delete.assert_awaited_once()
-    assert 'sub1' not in eng.subscriptions
-
-
-@pytest.mark.asyncio
-async def test_check_cycles_triggers_notification_over_threshold():
-    eng = make_engine()
-    eng.nodes['ns=2;i=2'] = {
-        'names': [{'tag_name': 'a'}],
-        'cycle_rule': {'cycle_increment': 5.5, 'cycle_count': 0},
-    }
-
-    await eng.check_cycles()
-
-    eng.send_notification_async.assert_called_once()
-
-
-@pytest.mark.asyncio
-async def test_check_opc_listenning_below_threshold():
-    eng = make_engine()
-    eng.non_receive_count = 2
-
-    result = await eng.check_opc_listenning('srv1')
-
-    assert eng.non_receive_count == 3
-    assert result is False
-    eng.send_notification_async.assert_not_called()
-
-
-@pytest.mark.asyncio
-async def test_check_opc_listenning_triggers_reconnect_at_15():
-    eng = make_engine()
-    eng.non_receive_count = 14
-
-    result = await eng.check_opc_listenning('srv1')
-
-    assert result is True
-    assert eng.send_notification_async.call_count == 2

tests/unit/engines/test_write.py

@@ -1,259 +0,0 @@
-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(),
-        **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_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()
-    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()

tests/unit/io/test_data_sink_publish.py

@@ -1,69 +0,0 @@
-from unittest.mock import AsyncMock
-
-import pytest
-
-import ingestor.metrics as metrics
-from ingestor.io.data_sink import DataSink
-
-
-def make_sink(data_manager=None):
-    dm = data_manager or AsyncMock()
-    sink = DataSink(data_manager=dm, postgres=AsyncMock())
-    sink.emit_metric = AsyncMock()
-    return sink, dm
-
-
-@pytest.mark.asyncio
-async def test_publish_reading_fans_out_per_name():
-    sink, dm = make_sink()
-    names = [
-        {'collection': 'raw_x', 'tag_name': 'Counter', 'model_id': '1', 'model_name': 'm1'},
-        {'collection': 'raw_y', 'tag_name': 'Contador', 'model_id': '2', 'model_name': 'm2'},
-    ]
-
-    await sink.publish_reading(
-        tag='ns=2;i=2', value=42, timestamp='2026-07-13T00:00:00Z', names=names
-    )
-
-    assert dm.publish.await_count == 2
-
-
-@pytest.mark.asyncio
-async def test_publish_reading_builds_doc_and_routes_collection():
-    sink, dm = make_sink()
-    names = [{'collection': 'raw_x', 'tag_name': 'Counter', 'model_id': '1', 'model_name': 'm1'}]
-
-    await sink.publish_reading(
-        tag='ns=2;i=2', value=42, timestamp='2026-07-13T00:00:00Z', names=names
-    )
-
-    collection, doc = dm.publish.await_args_list[0].args
-    assert collection == 'raw_x'
-    assert doc == {
-        'tag': 'ns=2;i=2',
-        'name': 'Counter',
-        'model_id': '1',
-        'timestamp': '2026-07-13T00:00:00Z',
-        'value': 42,
-    }
-
-
-@pytest.mark.asyncio
-async def test_publish_reading_emits_tag_value_metric_per_name():
-    sink, dm = make_sink()
-    names = [
-        {'collection': 'raw_x', 'tag_name': 'Counter', 'model_id': '1', 'model_name': 'm1'},
-        {'collection': 'raw_y', 'tag_name': 'Contador', 'model_id': '2', 'model_name': 'm2'},
-    ]
-
-    await sink.publish_reading(
-        tag='ns=2;i=2', value=42, timestamp='2026-07-13T00:00:00Z', names=names
-    )
-
-    assert sink.emit_metric.await_count == 2
-    sink.emit_metric.assert_any_await(
-        metric_object=metrics.OPC_TAG_VALUE,
-        method='set',
-        value=42,
-        tags={'pod_id': dm.pod_id, 'tag_name': 'Counter', 'model_id': '1'},
-    )

tests/unit/io/test_data_sink_status.py

@@ -1,89 +0,0 @@
-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_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()
-    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()

tests/unit/io/test_slots.py

@@ -1,64 +0,0 @@
-from ingestor.io.slots import parse_slot
-
-RAW = {
-    'server1': {
-        'server_id': '1',
-        'url': 'opc.tcp://h:4840',
-        'use_security': True,
-        'subscription_period_ms': 1000,
-        'tags': {
-            'ns=2;i=2': {
-                'frequency': '15000',
-                'active_read': {'enabled': True, 'cycle_multiple': 3},
-                'names': [
-                    {
-                        'collection': 'raw_x',
-                        'tag_name': 'Counter',
-                        'model_name': 'm1',
-                        'model_id': '1',
-                    },
-                    {
-                        'collection': 'raw_y',
-                        'tag_name': 'Contador',
-                        'model_name': 'm2',
-                        'model_id': '2',
-                    },
-                ],
-            }
-        },
-    }
-}
-
-
-def test_parse_slot_multi_name():
-    slot = parse_slot(RAW)
-    tag = slot['server1']['tags']['ns=2;i=2']
-    assert len(tag['names']) == 2
-    assert tag['active_read']['cycle_multiple'] == 3
-
-
-def test_parse_slot_defaults_active_read_disabled():
-    raw = {
-        's': {
-            'server_id': '1',
-            'url': 'u',
-            'tags': {
-                'ns=2;i=1': {
-                    'frequency': '1000',
-                    'names': [
-                        {'collection': 'c', 'tag_name': 't', 'model_name': 'm', 'model_id': '1'}
-                    ],
-                }
-            },
-        }
-    }
-    tag = parse_slot(raw)['s']['tags']['ns=2;i=1']
-    assert tag['active_read'] == {'enabled': False}
-
-
-def test_parse_slot_rejects_tag_without_names():
-    raw = {'s': {'server_id': '1', 'url': 'u', 'tags': {'ns=2;i=1': {'frequency': '1000'}}}}
-    import pytest
-
-    with pytest.raises((KeyError, ValueError)):
-        parse_slot(raw)

tests/unit/managers/test_ingestor_manager.py

@@ -1,4 +1,3 @@
-import gc
 from unittest.mock import AsyncMock, MagicMock, call, patch
 
 from pytest import fixture, mark
@@ -16,18 +15,6 @@ metadata = {
 }
 
 
-@fixture(autouse=True)
-def _collect_garbage_after_test():
-    # ponytail: IngestorManager.__del__ calls asyncio.run(self.shutdown()); if a
-    # cyclic-referenced instance from a previous test is swept mid-test by the
-    # generational GC, it can double-call whatever asyncio mock happens to be
-    # patched at that moment (e.g. test___del__). Force a deterministic sweep
-    # after each test so finalizers run in their own test's scope. Upgrade path:
-    # only needed if IngestorManager stops relying on __del__ for cleanup.
-    yield
-    gc.collect()
-
-
 @fixture
 @patch('ingestor.managers.ingestor_manager.DataManager')
 @patch('ingestor.managers.ingestor_manager.ResourceManager')

tests/unit/proto/test_proto.py

@@ -1,36 +0,0 @@
-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')

tests/unit/test_converters.py

@@ -1,47 +0,0 @@
-import pytest
-
-from ingestor.converters import ConversionError, to_typed_value
-
-
-@pytest.mark.parametrize(
-    'raw,expected',
-    [
-        ('false', False),
-        ('False', False),
-        ('0', False),
-        ('no', False),
-        ('off', False),
-        ('', False),
-        ('true', True),
-        ('1', True),
-        (True, True),
-    ],
-)
-def test_bool_from_json_string(raw, expected):
-    assert to_typed_value(raw, 'boolean') is expected
-
-
-def test_float_and_int():
-    assert to_typed_value('42.5', 'float') == 42.5
-    assert to_typed_value('7', 'int') == 7
-
-
-@pytest.mark.parametrize(
-    'raw,dtype',
-    [
-        (None, 'int'),
-        ('abc', 'float'),
-        ('123.123', 'int'),
-        (None, 'float'),
-    ],
-)
-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)

tests/unit/test_metrics.py

@@ -189,16 +189,3 @@ def test_notifications_sent():
     assert isinstance(metrics.NOTIFICATIONS_SENT, Counter)
     assert metrics.NOTIFICATIONS_SENT._name == 'notifications_sent'  # REMOVED _total
     assert set(metrics.NOTIFICATIONS_SENT._labelnames) == {'pod_id', 'level', 'block'}
-
-
-def test_gateway_metrics_exist():
-    """Verify the new OPC Gateway metrics (write, active read, error) are defined."""
-    metrics.OPC_WRITE_TOTAL.labels(pod_id='p', server_name='s', model_id='1', result='success')
-    metrics.OPC_WRITE_LATENCY.labels(pod_id='p', server_name='s')
-    metrics.OPC_ACTIVE_READ_TOTAL.labels(pod_id='p', server_name='s', result='success')
-    metrics.OPC_ACTIVE_READ_LATENCY.labels(pod_id='p', server_name='s')
-    metrics.OPC_WRITE_STATUS_BY_MODEL.labels(pod_id='p', model_id='1', status='ok')
-    metrics.OPC_TAG_VALUE.labels(pod_id='p', tag_name='t', model_id='1')
-    metrics.OPC_READ_LAG.labels(pod_id='p', server_name='s')
-    metrics.OPC_ERROR_BY_TYPE.labels(pod_id='p', server_name='s', error_type='BadTimeout')
-    metrics.OPC_DISCONNECTIONS_TOTAL.labels(pod_id='p', server_name='s')