Merge remote-tracking branch 'origin/release/SIENTIAPDE-1973' into rc/ingestor2

Merge feito para que a branch rc/ingestor2 passe a enxergar o rc 2
corretamente

.env.example

@@ -15,3 +15,32 @@ MONGODB_USERNAME="mongo_user"
 MONGODB_PASSWORD="mongo_password"
 MONGODB_DATABASE="sientia"
 
+# OPC Gateway
+NATS_URL="nats://localhost:4222"
+NATS_WRITE_SUBJECT="opc.write"
+NATS_QUEUE="gateway-writers"
+GRPC_ENABLED="false"
+GRPC_PORT="50051"
+CERT_BUCKET="sientia-dataops"
+CERT_CACHE_DIR="/tmp/opc-certs"
+WRITE_TIMESTAMP="true"
+WRITE_QUALITY="true"
+WRITE_MIN_PAUSE_S="0.0"
+IDLE_CONN_MAX_S="300.0"
+
+# OPC Gateway - MinIO (client cert storage, read by GatewayRuntime/CertProvider)
+MINIO_ENDPOINT="localhost:9000"
+MINIO_ACCESS_KEY="minioadmin"
+MINIO_SECRET_KEY="minioadmin"
+MINIO_SECURE="false"
+
+# OPC Gateway - Postgres (write_status updates, read by GatewayRuntime/DataSink
+# via sientia_do.utils.connectors_config.build_postgres_config)
+POSTGRES_HOST="localhost"
+POSTGRES_PORT="5432"
+POSTGRES_USER="sientia"
+POSTGRES_PASSWORD="sientia"
+POSTGRES_DBNAME="sientia"
+POSTGRES_MIN_CONNECTIONS="5"
+POSTGRES_MAX_CONNECTIONS="20"
+

.github/workflows/deploy.yml

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

.github/workflows/release.yml

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

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

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

.gitignore

@@ -181,4 +181,10 @@ 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

README.md

@@ -49,6 +49,27 @@ The OPC Ingestor uses a modular, manager-based architecture designed for scalabi
 - **Data Manager**: Manages data persistence (MongoDB)
 - **Resource Manager**: Coordinates resource allocation and instance coordination via Redis
 
+## 🚪 OPC Gateway (SIENTIAPDE-1972)
+
+Alongside the passive ingestion pipeline above, this service also runs an OPC
+Gateway for on-demand and command-driven interactions with OPC UA servers:
+
+- **NATS write consumer**: consumes write commands from `opc.write` (see
+  `NATS_WRITE_SUBJECT`) and executes them against the target OPC server
+  (`ingestor/adapters/nats_consumer.py`, `ingestor/engines/write.py`).
+- **On-demand gRPC `ReadTag`**: a gRPC server (`ingestor/adapters/grpc_server.py`,
+  `ingestor/proto/opc_gateway.proto`) for synchronous, single-tag reads.
+- **Scheduled active reads**: a polling read engine (`ingestor/engines/active_read.py`)
+  for tags that need periodic reads outside the subscription pipeline;
+  orchestrator-side wiring lands in SIENTIAPDE-1974.
+- **Certificate management**: client certs for secure OPC connections are
+  fetched from MinIO + a Mongo `platform_certificates` collection
+  (`ingestor/connection/certs.py`) rather than shipped in the image.
+
+See `docs/deploy-notes.md` for the deployment-facing details (env vars,
+health probe, NetworkPolicy, cert cache) and the `# OPC Gateway` block in
+`.env.example` for the full list of new environment variables.
+
 ## 📋 Prerequisites
 
 - Python 3.11+
@@ -253,6 +274,12 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring:
 - `ingestor_tag_written_count`: Data write operations
 - `redis_operations_total`: Redis operation count
 
+### OPC Gateway Metrics
+- `opc_write_total`: Active write attempts (NATS-driven), by result
+- `opc_write_latency_seconds`: Active write latency
+- `opc_active_read_total`: Scheduled/on-demand read attempts, by result
+- `opc_error_by_type_total`: OPC errors by type (connection, conversion, session, ...)
+
 ## ⚙️ Configuration
 
 ### Environment Variables
@@ -273,6 +300,10 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring:
 | `MONGODB_PASSWORD` | MongoDB password | `sientia` | No |
 | `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No |
 
+See the `# OPC Gateway` block in `.env.example` (`NATS_URL`, `GRPC_ENABLED`,
+`CERT_BUCKET`, `WRITE_TIMESTAMP`, ...) and `docs/deploy-notes.md` for the
+gateway-specific environment variables.
+
 ### OPC Server Configuration
 
 OPC servers are configured through Redis with the following structure. `cert_path` / `private_key_path` are only required for secured connections (see [Certificate Generation](#-certificate-generation)):

ingestor/adapters/grpc_server.py

@@ -0,0 +1,70 @@
+"""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 at the SIENTIAPDE-1974 integration -- `names` only exist in slot v2,
+        # which the Orchestrator produces after 1974. Until then ReadTag never
+        # persists (persisted=False); 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

@@ -0,0 +1,55 @@
+"""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/app.py

@@ -7,7 +7,16 @@ from time import time
 
 from prometheus_client import start_http_server
 
+try:  # optional: local dev reads .env; in-cluster env comes from Helm (no .env file)
+    from dotenv import load_dotenv
+
+    load_dotenv()
+except ImportError:
+    pass
+
 import ingestor.metrics as metrics
+from ingestor.config import GatewayConfig
+from ingestor.gateway_runtime import GatewayRuntime
 from ingestor.ingestor import Ingestor
 
 exit_signal = Event()
@@ -48,10 +57,36 @@ async def main():  # NOSONAR
         exit_signal.set()
     ingestor.logger.info('Ingestor prepared. Starting main loop.')
 
+    # Gateway write/active-read/gRPC wiring (Task 18). Only built if prepare_ingestor
+    # got far enough to have a data_manager (needed by CertProvider/DataSink) --
+    # skipped entirely if prepare failed before that point. Construction can hit
+    # the network (MinIO bucket check, Postgres engine): a failure there must not
+    # kill the read path, so the app continues without the gateway runtime.
+    runtime: GatewayRuntime | None = None
+    if ingestor.ingestor_manager is not None:
+        try:
+            runtime = GatewayRuntime(
+                config=GatewayConfig.from_env(),
+                data_manager=ingestor.ingestor_manager.data_manager,
+                logger=ingestor.logger,
+                notification_handler=ingestor.notification_handler,
+                metrics_controller=ingestor.metrics_controller,
+                metadata=ingestor.metadata,
+            )
+            await runtime.start()
+        except Exception as e:
+            runtime = None
+            metrics.APP_ERRORS_TOTAL.labels(pod_id=POD_ID).inc()
+            ingestor.logger.error(
+                f'GatewayRuntime failed to initialize, continuing without gateway: {e}'
+            )
+
     while not exit_signal.is_set():  # NOSONAR
         start_time = time()  # Start loop timer
         try:
             await ingestor.loop()
+            if runtime is not None:
+                await runtime.idle_sweep()
             metrics.APP_LOOP_COUNT.labels(pod_id=POD_ID).inc()  # Increment loop counter
 
             # Use asyncio.sleep instead of exit_signal.wait for better async compatibility
@@ -70,6 +105,8 @@ async def main():  # NOSONAR
             duration = time() - start_time
             metrics.APP_LOOP_DURATION.labels(pod_id=POD_ID).observe(duration)
 
+    if runtime is not None:
+        await runtime.stop()
     await ingestor.shutdown()
     metrics.APP_UP.labels(pod_id=POD_ID).set(0)  # Mark app as DOWN
 

ingestor/config.py

@@ -0,0 +1,46 @@
+"""Gateway-specific env config (NATS/gRPC/certs/write flags).
+
+Separate from ingestor.py's inline os.getenv (Redis/Mongo/TTLs) — this module
+covers only the new gateway wiring consumed by app.py in Task 18.
+"""
+
+from dataclasses import dataclass
+from os import getenv
+
+
+def _env_bool(name: str, default: bool) -> bool:
+    value = getenv(name)
+    if value is None:
+        return default
+    return value.strip().lower() in ('true', '1', 'yes')
+
+
+@dataclass
+class GatewayConfig:
+    nats_url: str = 'nats://localhost:4222'
+    nats_write_subject: str = 'opc.write'
+    nats_queue: str = 'gateway-writers'
+    grpc_enabled: bool = False
+    grpc_port: int = 50051
+    cert_bucket: str = 'sientia-dataops'
+    cert_cache_dir: str = '/tmp/opc-certs'  # noqa: S108 # nosec B108 - default is spec'd, overridable via env
+    write_timestamp: bool = True
+    write_quality: bool = True
+    write_min_pause_s: float = 0.0
+    idle_conn_max_s: float = 300.0
+
+    @classmethod
+    def from_env(cls) -> 'GatewayConfig':
+        return cls(
+            nats_url=getenv('NATS_URL', cls.nats_url),
+            nats_write_subject=getenv('NATS_WRITE_SUBJECT', cls.nats_write_subject),
+            nats_queue=getenv('NATS_QUEUE', cls.nats_queue),
+            grpc_enabled=_env_bool('GRPC_ENABLED', cls.grpc_enabled),
+            grpc_port=int(getenv('GRPC_PORT', cls.grpc_port)),
+            cert_bucket=getenv('CERT_BUCKET', cls.cert_bucket),
+            cert_cache_dir=getenv('CERT_CACHE_DIR', cls.cert_cache_dir),
+            write_timestamp=_env_bool('WRITE_TIMESTAMP', cls.write_timestamp),
+            write_quality=_env_bool('WRITE_QUALITY', cls.write_quality),
+            write_min_pause_s=float(getenv('WRITE_MIN_PAUSE_S', cls.write_min_pause_s)),
+            idle_conn_max_s=float(getenv('IDLE_CONN_MAX_S', cls.idle_conn_max_s)),
+        )

ingestor/connection/certs.py

@@ -0,0 +1,29 @@
+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

@@ -0,0 +1,19 @@
+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

@@ -0,0 +1,134 @@
+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

@@ -0,0 +1,17 @@
+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

@@ -0,0 +1,44 @@
+"""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

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

@@ -0,0 +1,192 @@
+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

@@ -0,0 +1,195 @@
+"""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/gateway_runtime.py

@@ -0,0 +1,156 @@
+"""GatewayRuntime (Task 18 F4 capstone): builds and owns the write/active-read/gRPC
+object graph and wires it into the app's lifecycle.
+
+Reuses the existing ingestor's logger/notification_handler/metrics_controller and
+its already-connected DataManager (so CertProvider shares the existing Mongo
+connection instead of opening a second one -- MongoDBRepository.find_one has the
+exact (collection_name, filters, metadata) signature CertProvider.fetch calls).
+
+# ponytail: subscription v2 swap + scheduled_tick wiring happen at 1974
+# integration; when swapping, build ONE SubscriptionEngine per OPC server (its
+# non_receive_count is engine-wide -- see plan ledger F1 carry-forward).
+"""
+
+from os import getenv
+
+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.repository.minio_repository import MinioRepository
+from sientia_do.temporal.activities.postgres import Postgres
+from sientia_do.utils.connectors_config import build_postgres_config
+
+from ingestor.adapters.grpc_server import GrpcServer
+from ingestor.adapters.nats_consumer import NatsWriteConsumer
+from ingestor.config import GatewayConfig, _env_bool
+from ingestor.connection.certs import CertProvider
+from ingestor.connection.pool import OpcConnectionPool
+from ingestor.engines.active_read import ActiveReadEngine
+from ingestor.engines.write import WriteEngine
+from ingestor.io.data_sink import DataSink
+from ingestor.managers.data_manager import DataManager
+
+
+class GatewayRuntime(SientiaMonitoring):
+    """Owns the gateway write/active-read/gRPC object graph. Resilient to NATS
+    being unreachable at startup: the read path (existing subscription loop,
+    gRPC on-demand reads) must keep working even if the write path can't
+    connect yet -- so a NATS connect failure is logged/notified, not raised,
+    and retried from `idle_sweep()` on each main-loop iteration."""
+
+    def __init__(
+        self,
+        config: GatewayConfig,
+        data_manager: DataManager,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+        metadata: dict | None = None,
+    ):
+        SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
+        self.config = config
+        self.metadata = metadata or {}
+        self._nats_started = False
+        self._nats_notified = False  # notify once per outage, not per 5s retry
+
+        minio = MinioRepository(
+            endpoint=getenv('MINIO_ENDPOINT', 'localhost:9000'),
+            access_key=getenv('MINIO_ACCESS_KEY', 'minioadmin'),
+            secret_key=getenv('MINIO_SECRET_KEY', 'minioadmin'),
+            secure=_env_bool('MINIO_SECURE', True),
+            bucket=config.cert_bucket,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+        cert_provider = CertProvider(
+            mongo=data_manager.mongo_repository,
+            minio=minio,
+            cache_dir=config.cert_cache_dir,
+            bucket=config.cert_bucket,
+            metadata=self.metadata,
+        )
+        self.pool = OpcConnectionPool(
+            cert_provider, logger, notification_handler, metrics_controller
+        )
+
+        postgres = Postgres(
+            **build_postgres_config(),
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+        self.sink = DataSink(data_manager, postgres)
+
+        self.write_engine = WriteEngine(
+            self.pool,
+            self.sink,
+            logger,
+            notification_handler,
+            metrics_controller,
+            write_timestamp=config.write_timestamp,
+            write_quality=config.write_quality,
+            min_pause_s=config.write_min_pause_s,
+        )
+        self.nats_consumer = NatsWriteConsumer(
+            config.nats_url,
+            self.write_engine,
+            logger,
+            subject=config.nats_write_subject,
+            queue=config.nats_queue,
+        )
+        self.active_read = ActiveReadEngine(
+            self.pool, self.sink, logger, notification_handler, metrics_controller
+        )
+
+        # slot_lookup=None for now: address -> names resolution needs slot v2
+        # (1974); until then ReadTag responses just aren't persisted.
+        self.grpc_server = (
+            GrpcServer(self.active_read, logger, port=config.grpc_port, slot_lookup=None)
+            if config.grpc_enabled
+            else None
+        )
+
+    async def start(self) -> None:
+        await self._start_nats()
+        if self.grpc_server is not None:
+            await self.grpc_server.start()
+        self.logger.info(
+            f'GatewayRuntime started: nats_started={self._nats_started} '
+            f'grpc_enabled={self.grpc_server is not None}'
+        )
+
+    async def stop(self) -> None:
+        if self.grpc_server is not None:
+            await self.grpc_server.stop()
+        if self._nats_started:
+            await self.nats_consumer.stop()
+            self._nats_started = False
+        await self.pool.close_all()
+
+    async def idle_sweep(self) -> None:
+        await self.pool.release_idle(self.config.idle_conn_max_s)
+        if not self._nats_started:
+            await self._start_nats()
+
+    async def _start_nats(self) -> None:
+        if self._nats_started:
+            return
+        try:
+            await self.nats_consumer.start()
+            self._nats_started = True
+            self._nats_notified = False
+        except Exception as e:
+            self.logger.error(f'GatewayRuntime: failed to start NATS write consumer: {e}')
+            if self._nats_notified:
+                return
+            self._nats_notified = True
+            await self.send_notification_async(
+                metadata=self.metadata,
+                notification_id='GATEWAY_NATS_CONSUMER_START_FAILED',
+                message=f'Failed to start NATS write consumer: {e}',
+                block='gateway_runtime',
+                level=NotificationLevel.ERROR,
+            )

ingestor/io/data_sink.py

@@ -0,0 +1,74 @@
+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

@@ -0,0 +1,25 @@
+"""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,3 +139,52 @@ 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

@@ -0,0 +1,18 @@
+syntax = "proto3";
+package opcgateway;
+
+service OpcGateway {
+  rpc ReadTag(ReadTagRequest) returns (ReadTagResponse);
+}
+message ReadTagRequest {
+  string server_id = 1;
+  string addr = 2;
+  string data_type = 3;
+}
+message ReadTagResponse {
+  string value = 1;
+  string opc_status_code = 2;
+  string source_timestamp = 3;
+  string server_timestamp = 4;
+  bool persisted = 5;
+}

ingestor/proto/opc_gateway_pb2.py

@@ -0,0 +1,40 @@
+# -*- coding: utf-8 -*-
+# Generated by the protocol buffer compiler.  DO NOT EDIT!
+# NO CHECKED-IN PROTOBUF GENCODE
+# source: opc_gateway.proto
+# Protobuf Python Version: 7.35.0
+"""Generated protocol buffer code."""
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import descriptor_pool as _descriptor_pool
+from google.protobuf import runtime_version as _runtime_version
+from google.protobuf import symbol_database as _symbol_database
+from google.protobuf.internal import builder as _builder
+_runtime_version.ValidateProtobufRuntimeVersion(
+    _runtime_version.Domain.PUBLIC,
+    7,
+    35,
+    0,
+    '',
+    'opc_gateway.proto'
+)
+# @@protoc_insertion_point(imports)
+
+_sym_db = _symbol_database.Default()
+
+
+
+
+DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x11opc_gateway.proto\x12\nopcgateway\"D\n\x0eReadTagRequest\x12\x11\n\tserver_id\x18\x01 \x01(\t\x12\x0c\n\x04\x61\x64\x64r\x18\x02 \x01(\t\x12\x11\n\tdata_type\x18\x03 \x01(\t\"\x80\x01\n\x0fReadTagResponse\x12\r\n\x05value\x18\x01 \x01(\t\x12\x17\n\x0fopc_status_code\x18\x02 \x01(\t\x12\x18\n\x10source_timestamp\x18\x03 \x01(\t\x12\x18\n\x10server_timestamp\x18\x04 \x01(\t\x12\x11\n\tpersisted\x18\x05 \x01(\x08\x32P\n\nOpcGateway\x12\x42\n\x07ReadTag\x12\x1a.opcgateway.ReadTagRequest\x1a\x1b.opcgateway.ReadTagResponseb\x06proto3')
+
+_globals = globals()
+_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
+_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'opc_gateway_pb2', _globals)
+if not _descriptor._USE_C_DESCRIPTORS:
+  DESCRIPTOR._loaded_options = None
+  _globals['_READTAGREQUEST']._serialized_start=33
+  _globals['_READTAGREQUEST']._serialized_end=101
+  _globals['_READTAGRESPONSE']._serialized_start=104
+  _globals['_READTAGRESPONSE']._serialized_end=232
+  _globals['_OPCGATEWAY']._serialized_start=234
+  _globals['_OPCGATEWAY']._serialized_end=314
+# @@protoc_insertion_point(module_scope)

ingestor/proto/opc_gateway_pb2_grpc.py

@@ -0,0 +1,97 @@
+# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
+"""Client and server classes corresponding to protobuf-defined services."""
+import grpc
+import warnings
+
+from . import opc_gateway_pb2 as opc__gateway__pb2
+
+GRPC_GENERATED_VERSION = '1.82.1'
+GRPC_VERSION = grpc.__version__
+_version_not_supported = False
+
+try:
+    from grpc._utilities import first_version_is_lower
+    _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
+except ImportError:
+    _version_not_supported = True
+
+if _version_not_supported:
+    raise RuntimeError(
+        f'The grpc package installed is at version {GRPC_VERSION},'
+        + ' but the generated code in opc_gateway_pb2_grpc.py depends on'
+        + f' grpcio>={GRPC_GENERATED_VERSION}.'
+        + f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}'
+        + f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.'
+    )
+
+
+class OpcGatewayStub:
+    """Missing associated documentation comment in .proto file."""
+
+    def __init__(self, channel):
+        """Constructor.
+
+        Args:
+            channel: A grpc.Channel.
+        """
+        self.ReadTag = channel.unary_unary(
+                '/opcgateway.OpcGateway/ReadTag',
+                request_serializer=opc__gateway__pb2.ReadTagRequest.SerializeToString,
+                response_deserializer=opc__gateway__pb2.ReadTagResponse.FromString,
+                _registered_method=True)
+
+
+class OpcGatewayServicer:
+    """Missing associated documentation comment in .proto file."""
+
+    def ReadTag(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+
+def add_OpcGatewayServicer_to_server(servicer, server):
+    rpc_method_handlers = {
+            'ReadTag': grpc.unary_unary_rpc_method_handler(
+                    servicer.ReadTag,
+                    request_deserializer=opc__gateway__pb2.ReadTagRequest.FromString,
+                    response_serializer=opc__gateway__pb2.ReadTagResponse.SerializeToString,
+            ),
+    }
+    generic_handler = grpc.method_handlers_generic_handler(
+            'opcgateway.OpcGateway', rpc_method_handlers)
+    server.add_generic_rpc_handlers((generic_handler,))
+    server.add_registered_method_handlers('opcgateway.OpcGateway', rpc_method_handlers)
+
+
+ # This class is part of an EXPERIMENTAL API.
+class OpcGateway:
+    """Missing associated documentation comment in .proto file."""
+
+    @staticmethod
+    def ReadTag(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(
+            request,
+            target,
+            '/opcgateway.OpcGateway/ReadTag',
+            opc__gateway__pb2.ReadTagRequest.SerializeToString,
+            opc__gateway__pb2.ReadTagResponse.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+            _registered_method=True)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "ingestor"
-version = "0.0.0"
+version = "2.0.0-rc1"
 description = "Sientia DataOps Ingestor - OPC Tag Ingestor"
 readme = "README.md"
 requires-python = ">=3.11"
@@ -23,6 +23,7 @@ exclude = [
     "*.pyc",
     ".pytest_cache",
     "htmlcov",
+    "ingestor/proto/*_pb2*.py",
 ]
 
 [tool.ruff.lint]
@@ -80,6 +81,9 @@ warn_unused_ignores = false
 warn_no_return = true
 strict_equality = true
 ignore_missing_imports = true
+# Generated protoc/grpcio-tools output - not a checking target, dynamic attrs
+# mypy can't see (message classes built at runtime from DESCRIPTOR).
+exclude = ["^ingestor/proto/"]
 
 # Ignore missing imports for external packages
 [[tool.mypy.overrides]]
@@ -106,6 +110,14 @@ ignore_missing_imports = true
 module = "pandas.*"
 ignore_missing_imports = true
 
+# Generated protoc/grpcio-tools output - not hand-maintained, dynamic attrs
+# mypy can't see (message classes built at runtime from DESCRIPTOR). Skip
+# following it entirely so importers see Any instead of false attr-defined errors.
+[[tool.mypy.overrides]]
+module = "ingestor.proto.*"
+ignore_errors = true
+follow_imports = "skip"
+
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 python_files = ["test_*.py"]

requirements-dev.txt

@@ -16,4 +16,4 @@ pytest-asyncio>=0.21.0   # Async test support (already in main requirements)
 
 # Development Tools
 ipython>=8.12.0          # Enhanced Python shell
-ipdb>=0.13.13            # IPython debugger
+ipdb>=0.13.13            # IPython debuggerpython-dotenv>=1.0    # local dev only: app/seed read .env if present (in-cluster env comes from Helm)

requirements.txt

@@ -1,5 +1,11 @@
 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

scripts/seed_e2e.py

@@ -0,0 +1,161 @@
+"""Seed test data for the OPC Gateway e2e validation session (SIENTIAPDE-1972).
+
+Populates:
+  1. Mongo `connectors` + `connector_tags` (each tag with `metadata: {}` — required
+     for documentation, per the feature owner).
+  2. Redis slot v1 (`slot:opc_tags:1`) so the legacy subscription path collects them.
+  3. Postgres test predictions table (with the `write_status` column) + one row the
+     NATS write test can UPDATE (correlated by timestamp + model_id).
+
+Idempotent: connectors/tags are upserted, the slot is overwritten, the table uses
+CREATE IF NOT EXISTS + ON CONFLICT DO NOTHING.
+
+Usage (from the repo root, sientia-dataops conda env):
+  python scripts/seed_e2e.py --print-only          # dry-run: show docs/SQL, touch nothing
+  python scripts/seed_e2e.py                       # seed Mongo + Redis + Postgres
+
+Env (same names the gateway/app use):
+  MONGODB_URL/MONGODB_USERNAME/MONGODB_PASSWORD/MONGODB_DATABASE
+  REDIS_HOST/REDIS_PORT  POSTGRES_HOST/PORT/USER/PASSWORD/DBNAME
+  OPC_SERVER_URL (default opc.tcp://localhost:49320 — KepServer default port)
+
+# ponytail: plain sync pymongo/redis/sqlalchemy, no sientia_do async plumbing —
+# this is a test fixture, not production code.
+"""
+
+import argparse
+import json
+import os
+import sys
+
+try:  # optional: local dev reads .env; in-cluster env comes from Helm
+    from dotenv import load_dotenv
+
+    load_dotenv()
+except ImportError:
+    pass
+
+OPC_URL = os.getenv('OPC_SERVER_URL', 'opc.tcp://localhost:49320')
+
+CONNECTOR = {
+    'connector_id': 9001,
+    'type': 'opc_ua',
+    'name': 'e2e-kepserver',
+    'url': OPC_URL,
+    'server_uri': 'urn:e2e:gateway:test',
+    'subscription_period_ms': 1000,
+    'use_security': False,  # round 1; flip after certs are in sientia-dataops
+    'status': 'Active',
+    'description': 'OPC Gateway e2e validation connector (SIENTIAPDE-1972)',
+}
+
+# metadata: {} on every tag — feature owner asked test tags to carry the field.
+TAGS = [
+    {'tag_id': 9101, 'connector_id': 9001, 'address': 'ns=2;s=Simulation Examples.Functions.Ramp1',
+     'tag_name': 'e2e_ramp', 'category': 'Read', 'data_type': 'float', 'unit': '-',
+     'status': 'Active', 'description': 'e2e read tag', 'metadata': {}},
+    {'tag_id': 9102, 'connector_id': 9001, 'address': 'ns=2;s=Simulation Examples.Functions.User1',
+     'tag_name': 'e2e_write_target', 'category': 'Write', 'data_type': 'float', 'unit': '-',
+     'status': 'Active', 'description': 'e2e write tag', 'metadata': {}},
+]
+
+# Slot v1 (legacy path the gateway still runs until SIENTIAPDE-1974 delivers slot v2).
+SLOT_V1 = {
+    'e2e-kepserver': {
+        'name': 'e2e-kepserver',
+        'server_id': '9001',
+        'url': OPC_URL,
+        'server_uri': 'urn:e2e:gateway:test',
+        'subscription_period_ms': 1000,
+        'tags': {
+            TAGS[0]['address']: {
+                'tag_name': TAGS[0]['tag_name'],
+                'frequency': '5000',
+                'topics': ['raw_e2e-gateway-test'],
+            }
+        },
+    }
+}
+
+PREDICTIONS_SQL = """
+CREATE SCHEMA IF NOT EXISTS sientia_data;
+CREATE TABLE IF NOT EXISTS sientia_data.e2e_predictions (
+    model_id text NOT NULL,
+    prediction double precision,
+    prediction_confidence double precision,
+    timestamp text NOT NULL,
+    write_status text,
+    PRIMARY KEY (model_id, timestamp)
+);
+INSERT INTO sientia_data.e2e_predictions (model_id, prediction, prediction_confidence, timestamp)
+VALUES ('9001', 42.5, 0.97, '2026-07-15T12:00:00Z')
+ON CONFLICT DO NOTHING;
+"""
+
+# The matching NATS write command for the row above (paste into `nats pub opc.write '...'`).
+NATS_CMD = {
+    'msg_id': 'e2e-0001', 'server_id': '9001', 'addr': TAGS[1]['address'],
+    'value': 42.5, 'data_type': 'float', 'tag_type': 'prediction',
+    'model_id': '9001', 'model_name': 'e2e-model',
+    'schema': 'sientia_data', 'table_name': 'e2e_predictions',
+    'timestamp': '2026-07-15T12:00:00Z',
+    'url': OPC_URL, 'use_security': False, 'server_uri': 'urn:e2e:gateway:test',
+}
+
+
+def seed_mongo():
+    from pymongo import MongoClient
+
+    url = os.getenv('MONGODB_URL', 'localhost:27017')
+    user, pwd = os.getenv('MONGODB_USERNAME', 'sientia'), os.getenv('MONGODB_PASSWORD', 'sientia')
+    db = MongoClient(f'mongodb://{user}:{pwd}@{url}')[os.getenv('MONGODB_DATABASE', 'sientia')]
+    db['connectors'].replace_one({'connector_id': CONNECTOR['connector_id']}, CONNECTOR, upsert=True)
+    for tag in TAGS:
+        db['connector_tags'].replace_one({'tag_id': tag['tag_id']}, tag, upsert=True)
+    print(f'mongo: upserted 1 connector + {len(TAGS)} tags (metadata: {{}})')
+
+
+def seed_redis():
+    import redis
+
+    r = redis.Redis(host=os.getenv('REDIS_HOST', 'localhost'), port=int(os.getenv('REDIS_PORT', '6379')))
+    r.set('slot:opc_tags:1', json.dumps(SLOT_V1))
+    print('redis: wrote slot:opc_tags:1 (v1)')
+
+
+def seed_postgres():
+    from sqlalchemy import create_engine, text
+
+    host, port = os.getenv('POSTGRES_HOST', 'localhost'), os.getenv('POSTGRES_PORT', '5432')
+    user, pwd = os.getenv('POSTGRES_USER', 'sientia'), os.getenv('POSTGRES_PASSWORD', 'sientia')
+    dbname = os.getenv('POSTGRES_DBNAME', 'sientia')
+    engine = create_engine(f'postgresql://{user}:{pwd}@{host}:{port}/{dbname}')
+    with engine.begin() as conn:
+        for stmt in filter(str.strip, PREDICTIONS_SQL.split(';')):
+            conn.execute(text(stmt))
+    print('postgres: sientia_data.e2e_predictions ready (write_status column + 1 row)')
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--print-only', action='store_true', help='show docs/SQL, touch nothing')
+    args = parser.parse_args()
+
+    if args.print_only:
+        print('connector:', json.dumps(CONNECTOR, indent=2))
+        print('tags:', json.dumps(TAGS, indent=2))
+        print('slot v1:', json.dumps(SLOT_V1, indent=2))
+        print('sql:', PREDICTIONS_SQL)
+        print('nats write cmd:', json.dumps(NATS_CMD, indent=2))
+        return 0
+
+    seed_mongo()
+    seed_redis()
+    seed_postgres()
+    print('\nNATS write command for the session:')
+    print(f"nats pub opc.write '{json.dumps(NATS_CMD)}'")
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

sonar-project.properties

@@ -8,3 +8,7 @@ sonar.python.coverage.reportPaths=coverage.xml
 sonar.python.xunit.reportPath=pytest.xml
 sonar.python.version=3.11
 sonar.coverage.exclusions=ingestor/app.py
+# Generated protoc/grpcio-tools output - not hand-maintained; renaming its
+# ReadTag/etc. methods to satisfy naming-convention rules would break gRPC
+# wire compatibility (grpcio-tools dictates these names, not style choice).
+sonar.exclusions=ingestor/proto/**

tests/functional/conftest.py

@@ -1,8 +1,5 @@
 # import subprocess
 # from time import sleep
-# from typing import Generator
-# import uuid
-# from kafka import KafkaConsumer
 # import pytest
 # from redis import Redis
 
@@ -32,16 +29,3 @@
 #     # Limpa o banco de dados após os testes
 #     redis.flushdb()
 #     redis.close()
-
-
-# def kafka_searcher(topic) -> Generator[KafkaConsumer, None, None]:
-#     consumer = KafkaConsumer(
-#         topic,
-#         bootstrap_servers="localhost:9092",
-#         group_id=f"test-group-{uuid.uuid4()}",
-#         auto_offset_reset="earliest",   # Começa a consumir apenas mensagens novas
-#         enable_auto_commit=True,
-#     )
-
-#     yield consumer
-#     consumer.close()

tests/functional/test_single_node.py

@@ -2,8 +2,6 @@
 # import subprocess
 # from time import sleep
 
-# from tests.functional.conftest import kafka_searcher
-
 
 # new_data = {
 #     "slot:opc_tags:1": {
@@ -69,14 +67,6 @@
 #     assert redis_client.get("lease:opc_tags:1") == 'ingestor'
 #     assert redis_client.get("heartbeat:ingestor:ingestor") == '1'
 
-#     # Check if data is in Kafka
-
-#     kafka = next(kafka_searcher('test_topic_1'))
-#     sleep(1)
-#     messages = kafka.poll(timeout_ms=10000)
-
-#     assert messages, "Expected messages in Kafka, but got none."
-
 
 # def test_simple_double_slot(redis_client):
 
@@ -90,12 +80,6 @@
 #     assert redis_client.get("lease:opc_tags:1") == 'ingestor'
 #     assert redis_client.get("heartbeat:ingestor:ingestor") == '1'
 
-#     # Check if data is in Kafka
-#     kafka1 = next(kafka_searcher('test_topic_double_slot1'))
-#     messages = kafka1.poll(timeout_ms=10000)
-
-#     assert messages, "Expected messages in test_topic_double_slot1, but got none."
-
 #     new_data['slot:opc_tags:2']['server2']['tags']['ns=2;i=2']['topics'] = [
 #         'test_topic_double_slot2']
 #     redis_client.set("slot:opc_tags:2",
@@ -107,12 +91,3 @@
 #     assert redis_client.get("lease:opc_tags:2") == 'ingestor'
 #     assert redis_client.get("lease:opc_tags:1") == 'ingestor'
 #     assert redis_client.get("heartbeat:ingestor:ingestor") == '1'
-
-#     # Check if data is in Kafka
-#     kafka2 = next(kafka_searcher('test_topic_double_slot2'))
-#     messages = kafka2.poll(timeout_ms=10000)
-
-#     assert messages, "Expected messages in test_topic_double_slot2, but got none."
-
-#     messages = kafka1.poll(timeout_ms=10000)
-#     assert messages, "Expected messages in test_topic_double_slot1, but got none."

tests/unit/adapters/test_grpc_server.py

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

@@ -0,0 +1,76 @@
+import json
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from ingestor.adapters.nats_consumer import NatsWriteConsumer
+
+
+def make_consumer(**kwargs):
+    return NatsWriteConsumer(
+        nats_url='nats://localhost:4222',
+        write_engine=AsyncMock(),
+        logger=MagicMock(),
+        **kwargs,
+    )
+
+
+@pytest.mark.asyncio
+async def test_handle_valid_json_dispatches_to_write_engine():
+    consumer = make_consumer()
+    payload = {'server_id': 'srv1', 'addr': 'ns=2;i=5', 'value': 1.0, 'data_type': 'float'}
+    msg = MagicMock(data=json.dumps(payload).encode())
+
+    await consumer._handle(msg)
+
+    consumer.write_engine.write.assert_awaited_once_with(payload)
+
+
+@pytest.mark.asyncio
+async def test_handle_invalid_json_drops_message_and_logs_error():
+    consumer = make_consumer()
+    msg = MagicMock(data=b'not-json')
+
+    await consumer._handle(msg)
+
+    consumer.write_engine.write.assert_not_awaited()
+    consumer.logger.error.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_start_subscribes_with_subject_and_queue():
+    consumer = make_consumer(subject='opc.write', queue='gateway-writers')
+    mock_nc = AsyncMock()
+    mock_nc.subscribe = AsyncMock()
+
+    with patch('ingestor.adapters.nats_consumer.nats') as mock_nats:
+        mock_nats.connect = AsyncMock(return_value=mock_nc)
+        await consumer.start()
+
+    mock_nats.connect.assert_awaited_once_with(servers='nats://localhost:4222')
+    mock_nc.subscribe.assert_awaited_once_with(
+        'opc.write', queue='gateway-writers', cb=consumer._handle
+    )
+
+
+@pytest.mark.asyncio
+async def test_stop_before_start_does_not_raise():
+    consumer = make_consumer()
+
+    await consumer.stop()  # no raise
+
+
+@pytest.mark.asyncio
+async def test_stop_after_start_drains_and_unsubscribes():
+    consumer = make_consumer()
+    mock_nc = AsyncMock()
+    mock_sub = AsyncMock()
+    mock_nc.subscribe = AsyncMock(return_value=mock_sub)
+
+    with patch('ingestor.adapters.nats_consumer.nats') as mock_nats:
+        mock_nats.connect = AsyncMock(return_value=mock_nc)
+        await consumer.start()
+        await consumer.stop()
+
+    mock_sub.unsubscribe.assert_awaited_once()
+    mock_nc.drain.assert_awaited_once()

tests/unit/connection/test_certs.py

@@ -0,0 +1,28 @@
+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

@@ -0,0 +1,8 @@
+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

@@ -0,0 +1,184 @@
+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

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

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

@@ -0,0 +1,203 @@
+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

@@ -0,0 +1,259 @@
+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

@@ -0,0 +1,69 @@
+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

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

@@ -0,0 +1,64 @@
+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,3 +1,4 @@
+import gc
 from unittest.mock import AsyncMock, MagicMock, call, patch
 
 from pytest import fixture, mark
@@ -15,6 +16,18 @@ 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

@@ -0,0 +1,36 @@
+from ingestor.proto import opc_gateway_pb2, opc_gateway_pb2_grpc
+
+
+def test_read_tag_request_roundtrip():
+    req = opc_gateway_pb2.ReadTagRequest(server_id='s1', addr='ns=2;s=Tag1', data_type='Float')
+
+    data = req.SerializeToString()
+    parsed = opc_gateway_pb2.ReadTagRequest()
+    parsed.ParseFromString(data)
+
+    assert parsed.server_id == 's1'
+    assert parsed.addr == 'ns=2;s=Tag1'
+    assert parsed.data_type == 'Float'
+
+
+def test_read_tag_response_roundtrip():
+    resp = opc_gateway_pb2.ReadTagResponse(
+        value='42.0',
+        opc_status_code='Good',
+        source_timestamp='2026-07-14T00:00:00Z',
+        server_timestamp='2026-07-14T00:00:01Z',
+        persisted=True,
+    )
+
+    data = resp.SerializeToString()
+    parsed = opc_gateway_pb2.ReadTagResponse()
+    parsed.ParseFromString(data)
+
+    assert parsed.value == '42.0'
+    assert parsed.opc_status_code == 'Good'
+    assert parsed.persisted is True
+
+
+def test_grpc_stub_module_loads():
+    assert hasattr(opc_gateway_pb2_grpc, 'OpcGatewayStub')
+    assert hasattr(opc_gateway_pb2_grpc, 'OpcGatewayServicer')

tests/unit/test_app.py

@@ -43,6 +43,15 @@ def mock_app_env(monkeypatch):
     monkeypatch.setattr(app, 'Ingestor', mocks['Ingestor'])
     monkeypatch.setattr(asyncio, 'sleep', mocks['asyncio_sleep'])
 
+    mocks['GatewayConfig'] = MagicMock()
+    mocks['GatewayRuntime'] = MagicMock()
+    monkeypatch.setattr(app, 'GatewayConfig', mocks['GatewayConfig'])
+    monkeypatch.setattr(app, 'GatewayRuntime', mocks['GatewayRuntime'])
+    mock_runtime_instance = mocks['GatewayRuntime'].return_value
+    mock_runtime_instance.start = AsyncMock()
+    mock_runtime_instance.idle_sweep = AsyncMock()
+    mock_runtime_instance.stop = AsyncMock()
+
     monkeypatch.setattr(
         app.metrics.APP_UP,
         'labels',
@@ -77,6 +86,7 @@ def mock_app_env(monkeypatch):
     mock_ingestor_instance = mocks['Ingestor'].return_value
     mock_ingestor_instance.poll_interval = 0.01
     mock_ingestor_instance.logger = MagicMock()
+    mock_ingestor_instance.ingestor_manager = MagicMock()
 
     # Make async methods async mocks
     mock_ingestor_instance.prepare_ingestor = AsyncMock()
@@ -124,6 +134,21 @@ async def test_main_successful_run_one_loop(mock_app_env, capsys):
     mock_ingestor_instance.shutdown.assert_called_once()
     mock_ingestor_instance.logger.info.assert_any_call('Main loop exit_signaled.')
 
+    # GatewayRuntime (Task 18): built once prepare_ingestor succeeds, started before
+    # the loop, idle-swept each iteration, stopped on shutdown.
+    mock_app_env['GatewayRuntime'].assert_called_once_with(
+        config=mock_app_env['GatewayConfig'].from_env.return_value,
+        data_manager=mock_ingestor_instance.ingestor_manager.data_manager,
+        logger=mock_ingestor_instance.logger,
+        notification_handler=mock_ingestor_instance.notification_handler,
+        metrics_controller=mock_ingestor_instance.metrics_controller,
+        metadata=mock_ingestor_instance.metadata,
+    )
+    mock_runtime_instance = mock_app_env['GatewayRuntime'].return_value
+    mock_runtime_instance.start.assert_awaited_once()
+    mock_runtime_instance.idle_sweep.assert_awaited_once()
+    mock_runtime_instance.stop.assert_awaited_once()
+
     captured = capsys.readouterr()
     assert 'Prometheus server started on port 9090.' in captured.out
 
@@ -297,12 +322,16 @@ async def test_main_prepare_ingestor_failure(mock_app_env, capsys):
     mock_exit_signal = mock_app_env['mock_exit_signal']
 
     mock_ingestor_instance.prepare_ingestor.side_effect = Exception('Preparation failed')
+    mock_ingestor_instance.ingestor_manager = None  # prepare failed before assignment
     mock_exit_signal.is_set.side_effect = [False, True]
 
     with pytest.raises(OsExitCalledError) as excinfo:
         await app.main()
     assert excinfo.value.code == 0
 
+    # GatewayRuntime must not be built when prepare_ingestor never got a data_manager
+    mock_app_env['GatewayRuntime'].assert_not_called()
+
     # Verify error metrics were incremented
     app.metrics.APP_ERRORS_TOTAL.labels.assert_called_with(pod_id='test_pod')
     mock_app_env['metrics_APP_ERRORS_TOTAL_labels_inc'].assert_called_once()
@@ -317,3 +346,34 @@ async def test_main_prepare_ingestor_failure(mock_app_env, capsys):
     mock_ingestor_instance.logger.error.assert_called_once_with(
         'Failed to prepare ingestor: Preparation failed'
     )
+
+
+@pytest.mark.asyncio
+async def test_main_gateway_runtime_init_failure_keeps_read_path_alive(mock_app_env):
+    """GatewayRuntime construction hitting the network (MinIO/Postgres) and failing
+    must not kill the app: the main loop (read path) still runs and shuts down cleanly."""
+    mock_ingestor_instance = mock_app_env['Ingestor'].return_value
+    mock_exit_signal = mock_app_env['mock_exit_signal']
+
+    mock_app_env['GatewayRuntime'].side_effect = Exception('MinIO unreachable')
+    mock_exit_signal.is_set.side_effect = [False, True]
+    mock_app_env['time_time'].side_effect = [10.0, 11.0]
+
+    with pytest.raises(OsExitCalledError) as excinfo:
+        await app.main()
+    assert excinfo.value.code == 0
+
+    # Loop still ran and shutdown was clean, without any runtime calls
+    mock_ingestor_instance.loop.assert_called_once()
+    mock_ingestor_instance.shutdown.assert_called_once()
+    mock_runtime_instance = mock_app_env['GatewayRuntime'].return_value
+    mock_runtime_instance.idle_sweep.assert_not_called()
+    mock_runtime_instance.stop.assert_not_called()
+
+    # Failure was logged and counted, but did not set exit_signal
+    assert any(
+        'GatewayRuntime failed to initialize' in str(c)
+        for c in mock_ingestor_instance.logger.error.call_args_list
+    )
+    mock_app_env['metrics_APP_ERRORS_TOTAL_labels_inc'].assert_called_once()
+    mock_exit_signal.set.assert_not_called()

tests/unit/test_config.py

@@ -0,0 +1,71 @@
+import pytest
+
+from ingestor.config import GatewayConfig
+
+_ENV_VARS = (
+    'NATS_URL',
+    'NATS_WRITE_SUBJECT',
+    'NATS_QUEUE',
+    'GRPC_ENABLED',
+    'GRPC_PORT',
+    'CERT_BUCKET',
+    'CERT_CACHE_DIR',
+    'WRITE_TIMESTAMP',
+    'WRITE_QUALITY',
+    'WRITE_MIN_PAUSE_S',
+    'IDLE_CONN_MAX_S',
+)
+
+
+@pytest.fixture(autouse=True)
+def clean_env(monkeypatch):
+    for var in _ENV_VARS:
+        monkeypatch.delenv(var, raising=False)
+
+
+def test_from_env_defaults():
+    cfg = GatewayConfig.from_env()
+
+    assert cfg.nats_url == 'nats://localhost:4222'
+    assert cfg.nats_write_subject == 'opc.write'
+    assert cfg.nats_queue == 'gateway-writers'
+    assert cfg.grpc_enabled is False
+    assert cfg.grpc_port == 50051
+    assert cfg.cert_bucket == 'sientia-dataops'
+    assert cfg.cert_cache_dir == '/tmp/opc-certs'  # noqa: S108
+    assert cfg.write_timestamp is True
+    assert cfg.write_quality is True
+    assert cfg.write_min_pause_s == 0.0
+    assert cfg.idle_conn_max_s == 300.0
+
+
+def test_from_env_overrides(monkeypatch):
+    monkeypatch.setenv('GRPC_ENABLED', 'true')
+    monkeypatch.setenv('GRPC_PORT', '6000')
+    monkeypatch.setenv('WRITE_MIN_PAUSE_S', '0.5')
+    monkeypatch.setenv('WRITE_QUALITY', 'false')
+
+    cfg = GatewayConfig.from_env()
+
+    assert cfg.grpc_enabled is True
+    assert cfg.grpc_port == 6000
+    assert cfg.write_min_pause_s == 0.5
+    assert cfg.write_quality is False
+
+
+@pytest.mark.parametrize(
+    'value,expected',
+    [
+        ('0', False),
+        ('no', False),
+        ('FALSE', False),
+        ('1', True),
+        ('YES', True),
+    ],
+)
+def test_bool_parsing_edge_cases(monkeypatch, value, expected):
+    monkeypatch.setenv('GRPC_ENABLED', value)
+
+    cfg = GatewayConfig.from_env()
+
+    assert cfg.grpc_enabled is expected

tests/unit/test_converters.py

@@ -0,0 +1,47 @@
+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_gateway_runtime.py

@@ -0,0 +1,172 @@
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from ingestor import gateway_runtime as gr
+from ingestor.config import GatewayConfig
+
+_PATCHED_CLASSES = (
+    'MinioRepository',
+    'CertProvider',
+    'OpcConnectionPool',
+    'Postgres',
+    'build_postgres_config',
+    'DataSink',
+    'WriteEngine',
+    'NatsWriteConsumer',
+    'ActiveReadEngine',
+    'GrpcServer',
+)
+
+
+@pytest.fixture
+def mocks(monkeypatch):
+    patched = {}
+    for name in _PATCHED_CLASSES:
+        m = MagicMock()
+        monkeypatch.setattr(gr, name, m)
+        patched[name] = m
+
+    patched['build_postgres_config'].return_value = {
+        'host': 'localhost',
+        'port': 5432,
+        'user': 'u',
+        'password': 'p',
+        'dbname': 'd',
+        'min_connections': 5,
+        'max_connections': 20,
+    }
+    patched['OpcConnectionPool'].return_value.close_all = AsyncMock()
+    patched['OpcConnectionPool'].return_value.release_idle = AsyncMock()
+    patched['NatsWriteConsumer'].return_value.start = AsyncMock()
+    patched['NatsWriteConsumer'].return_value.stop = AsyncMock()
+    patched['GrpcServer'].return_value.start = AsyncMock()
+    patched['GrpcServer'].return_value.stop = AsyncMock()
+    return patched
+
+
+def make_runtime(**config_overrides):
+    return gr.GatewayRuntime(
+        config=GatewayConfig(**config_overrides),
+        data_manager=MagicMock(),
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(),
+    )
+
+
+def test_builds_object_graph_with_config_flags(mocks):
+    runtime = make_runtime(
+        write_timestamp=False,
+        write_quality=False,
+        write_min_pause_s=0.25,
+        cert_bucket='my-bucket',
+        cert_cache_dir='certs-cache',
+        grpc_enabled=False,
+    )
+
+    mocks['OpcConnectionPool'].assert_called_once()
+    mocks['DataSink'].assert_called_once()
+
+    mocks['WriteEngine'].assert_called_once()
+    _, kwargs = mocks['WriteEngine'].call_args
+    assert kwargs['write_timestamp'] is False
+    assert kwargs['write_quality'] is False
+    assert kwargs['min_pause_s'] == 0.25
+
+    _, cert_kwargs = mocks['CertProvider'].call_args
+    assert cert_kwargs['bucket'] == 'my-bucket'
+    assert cert_kwargs['cache_dir'] == 'certs-cache'
+
+    mocks['GrpcServer'].assert_not_called()
+    assert runtime.grpc_server is None
+
+
+def test_builds_grpc_server_when_enabled(mocks):
+    runtime = make_runtime(grpc_enabled=True, grpc_port=12345)
+
+    mocks['GrpcServer'].assert_called_once()
+    assert runtime.grpc_server is mocks['GrpcServer'].return_value
+
+
+@pytest.mark.asyncio
+async def test_start_starts_nats_consumer(mocks):
+    runtime = make_runtime(grpc_enabled=False)
+
+    await runtime.start()
+
+    mocks['NatsWriteConsumer'].return_value.start.assert_awaited_once()
+    mocks['GrpcServer'].return_value.start.assert_not_called()
+    assert runtime._nats_started is True
+
+
+@pytest.mark.asyncio
+async def test_start_also_starts_grpc_when_enabled(mocks):
+    runtime = make_runtime(grpc_enabled=True)
+
+    await runtime.start()
+
+    mocks['NatsWriteConsumer'].return_value.start.assert_awaited_once()
+    mocks['GrpcServer'].return_value.start.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_nats_start_failure_does_not_raise_and_idle_sweep_retries(mocks):
+    runtime = make_runtime()
+    runtime.send_notification_async = AsyncMock()
+    nats_mock = mocks['NatsWriteConsumer'].return_value
+    nats_mock.start = AsyncMock(side_effect=[Exception('boom'), Exception('boom'), None])
+
+    await runtime.start()  # must not raise
+
+    assert runtime._nats_started is False
+    runtime.logger.error.assert_called_once()
+    runtime.send_notification_async.assert_awaited_once()
+
+    await runtime.idle_sweep()  # second failure: retry connects, but no re-notify
+
+    assert runtime._nats_started is False
+    runtime.send_notification_async.assert_awaited_once()  # still just one notification
+
+    await runtime.idle_sweep()
+
+    assert runtime._nats_started is True
+    assert nats_mock.start.await_count == 3
+    # flag reset on success: a future outage notifies again
+    assert runtime._nats_notified is False
+
+
+@pytest.mark.asyncio
+async def test_stop_stops_started_components_and_closes_pool(mocks):
+    runtime = make_runtime(grpc_enabled=True)
+    await runtime.start()
+
+    await runtime.stop()
+
+    mocks['GrpcServer'].return_value.stop.assert_awaited_once()
+    mocks['NatsWriteConsumer'].return_value.stop.assert_awaited_once()
+    mocks['OpcConnectionPool'].return_value.close_all.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_stop_skips_nats_stop_if_never_started(mocks):
+    runtime = make_runtime()
+    runtime.send_notification_async = AsyncMock()
+    nats_mock = mocks['NatsWriteConsumer'].return_value
+    nats_mock.start = AsyncMock(side_effect=Exception('boom'))
+
+    await runtime.start()  # nats fails to start
+    await runtime.stop()
+
+    nats_mock.stop.assert_not_called()
+    mocks['OpcConnectionPool'].return_value.close_all.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_idle_sweep_releases_idle_connections(mocks):
+    runtime = make_runtime(idle_conn_max_s=42.0)
+    runtime._nats_started = True  # isolate from the retry path
+
+    await runtime.idle_sweep()
+
+    mocks['OpcConnectionPool'].return_value.release_idle.assert_awaited_once_with(42.0)

tests/unit/test_metrics.py

@@ -189,3 +189,16 @@ 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')