chore: bump version to 2.1.0-rc1

SIENTIAPDE-1974: leitura v2 — slot v2 via SubscriptionEngine, certificado OPC do MinIO

.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"
+

.gitignore

@@ -187,4 +187,4 @@ git_log
 docs/superpowers/
 docs/e2e-test-guide.md
 docs/deploy-notes.md
-docs/HANDOFF.md
+docs/HANDOFF.md

CLAUDE.md

@@ -40,7 +40,9 @@ app.py (asyncio loop, signals, Prometheus server)
   -> Ingestor (ingestor.py) — lifecycle: prepare_ingestor() -> loop() -> shutdown()
        -> IngestorManager (managers/ingestor_manager.py) — central coordinator
             -> ResourceManager (managers/resource_manager.py) — Redis slot leases, heartbeats, active-ingestor registry
-            -> OpcManager (managers/opc_manager.py), one instance per OPC server — connection, subscriptions, datachange callbacks
+            -> ServerReader (managers/server_reader.py), one per OPC server — SubscriptionEngine over the shared OpcConnectionPool
+               (READ_ENGINE=legacy: OpcManager (managers/opc_manager.py), the rc3 reader kept as plan B until the end of SIENTIAPDE-1974)
+            -> OpcConnectionPool (connection/pool.py) + CertProvider (connection/certs.py) — pooled connections, cert from MinIO
             -> DataManager (managers/data_manager.py) — MongoDB persistence + optional Kafka publish
 ```
 
@@ -56,7 +58,7 @@ app.py (asyncio loop, signals, Prometheus server)
 
 Any change to slot/lease counts should be traced through `manage_leases`/`manage_no_slots` — that's where the rebalancing math lives, not in `ResourceManager` itself (which just does Redis primitives).
 
-**OpcManager** owns one OPC UA `Client` connection (asyncua), handles security policy setup (cert/key-based, `SecurityPolicyBasic256`), subscriptions per tag group, and `datachange_notification` callbacks that push into `data_queue` for `DataManager` to consume. Tracks `non_receive_count` to detect silently-dead subscriptions (`check_cycles`).
+**ServerReader** (SIENTIAPDE-1974, default `READ_ENGINE=v2`; the legacy `OpcManager` stays as plan B with `READ_ENGINE=legacy` — slot v1 only, certificate files from the Gitea clone — until it is unused and the tests pass the same without it) reads one OPC server through a `SubscriptionEngine`: the connection comes from the shared `OpcConnectionPool` (also used by NATS writes and gRPC), which fetches the platform certificate from MinIO via `CertProvider` and uses `SecurityPolicyBasic256Sha256` — no certificate files in the repo/image. Slots are parsed by `io/slots.py` (`parse_slot`), which accepts slot v2 (`names[]`) and converts the legacy v1 (`topics`). `datachange_notification` fans each value out to every `names` entry via `DataSink`. Subscribed connections are pinned in the pool so the idle sweep never drops them. Tracks `non_receive_count` to detect silently-dead subscriptions (`check_cycles` / `check_opc_listenning`); a lost server is shut down (pool connection evicted) and re-initialized on the next loop.
 
 **Metrics** (`ingestor/metrics.py`): all Prometheus metrics defined here, always labeled by `pod_id`. Emitted via `SientiaMonitoring.emit_metric(...)` (from `sientia_do`), not the prometheus client directly — new metrics should follow that pattern for consistency with logging/notifications.
 

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+
@@ -102,9 +123,9 @@ The OPC Ingestor uses a modular, manager-based architecture designed for scalabi
    # Ensure services are accessible on localhost with appropriate ports
    ```
 
-6. **Generate an OPC UA client certificate** (only if connecting to a secured OPC UA server)
+6. **Register the OPC UA client certificate** (only if connecting to a secured OPC UA server)
 
-   See [Certificate Generation](#-certificate-generation) below.
+   See [OPC Client Certificate](#-opc-client-certificate) below.
 
 ##  Usage
 
@@ -172,37 +193,50 @@ pytest --cov=ingestor --cov-report=html
 
 The feeder creates sample OPC tag configurations in Redis that the ingestor can discover and manage.
 
-## 🔐 Certificate Generation
+## 🔐 OPC Client Certificate
 
-OPC UA servers that require secure connections (e.g. KEPServer with `Basic256` policy) validate the ingestor via an X.509 client certificate. Use `scripts/generate-and-push-opc-cert.sh` to generate that certificate and push it to the Gitea repo the pod clones on boot.
+OPC UA servers that require secure connections validate the gateway with a single,
+shared platform X.509 client certificate. Every OPC connection — subscription reads,
+NATS writes and gRPC reads — uses it with `Basic256Sha256` (SIENTIAPDE-1974).
 
-```bash
-GITEA_PASSWORD='<senha-do-gitea>' ./scripts/generate-and-push-opc-cert.sh
-```
+The certificate is **never** read from the repository or the image. On connect, the
+`CertProvider` (`ingestor/connection/certs.py`):
+1. reads the single document of the Mongo collection `platform_certificates`:
+   `{"client_cert": {"minio_object_name": "..."}, "private_key": {"minio_object_name": "..."}}`;
+2. downloads both objects from the MinIO bucket `CERT_BUCKET` (default
+   `streamlit-connectors`) into `CERT_CACHE_DIR` (default `/tmp/opc-certs`).
 
-Environment variables (all optional except `GITEA_PASSWORD`):
+Requirements for a server to accept the connection:
+- the server's slot config has `use_security: true` (a slot v1 server with a
+  `cert_path` is treated the same way; the path itself is ignored);
+- the certificate's SAN URI equals the server's `server_uri` in the slot — it is sent
+  as the client `application_uri` (KEPServer rejects a mismatch with
+  `BadCertificateUriInvalid`);
+- the server exposes a `Basic256Sha256` endpoint and trusts the certificate (e.g.
+  KEPServer: *OPC UA Configuration Manager → Trusted Clients → Trust*, then restart
+  its runtime).
 
-| Variable | Description | Default |
-|----------|-------------|---------|
-| `GITEA_URL` | Gitea in-cluster URL | `https://git.sientia.ai` |
-| `GITEA_USER` | Gitea user | `gitea_admin` |
-| `GITEA_PASSWORD` | Gitea password (**required**) | — |
-| `GITEA_REPO` | Repo the ingestor pod clones | `gitea_admin/sientia-dataops-opc-ingestor` |
-| `BRANCH` | Branch to push the cert to | `main` |
-| `APP_URI` | SAN URI of the cert — **must match** the `uri` field of the corresponding document in the `OPC_servers` Mongo collection, since the ingestor uses it as `application_uri` (KEPServer rejects with `BadCertificateUriInvalid` on mismatch) | `urn:sientia:opc-ingestor` |
-| `CN` | Certificate common name | `sientia-opc-ingestor` |
-| `DAYS` | Certificate validity (days) | `3650` |
-| `FORCE` | Set to `1` to overwrite an existing cert already pushed to the repo | unset |
+MinIO is only contacted when a certificate is first needed, so a MinIO outage only
+affects secured servers, which are retried on the next main-loop iteration.
 
-What the script does:
-1. Generates a self-signed RSA-2048 X.509 cert + key with `openssl`, with the SAN URI, `keyUsage`, and `extendedKeyUsage` extensions OPC UA / KEPServer require.
-2. Clones the ingestor's Gitea repo, copies the cert/key (`.pem`) and a `.der` copy (for manual import) into `certs/`, commits, and pushes.
-3. Prints the paths the pod will see after cloning (`/app/certs/opc-ingestor-cert.pem`, `/app/certs/opc-ingestor-key.pem`) and the next manual steps.
+### Plan B: legacy read engine (`READ_ENGINE=legacy`)
 
-After pushing, you still need to:
-- Set `cert_path` / `private_key_path` (and `uri`) on the server's document in the `OPC_servers` collection so `OpcManager` picks them up (see [OPC Server Configuration](#opc-server-configuration)).
-- Restart the ingestor deployment so it re-clones the repo: `kubectl -n sientia rollout restart deployment/opc-ingestor`.
-- On first connection, the server rejects the new cert — trust it in the OPC UA server's console (e.g. KEPServer: *OPC UA Configuration Manager → Trusted Clients → Trust*) and restart its runtime.
+Until the end of SIENTIAPDE-1974 the legacy reader is kept for validation and as a
+fallback. `READ_ENGINE` selects the read path (default `v2`; an invalid value falls
+back to `v2` with a warning):
+
+| `READ_ENGINE` | Reader | Slot | Certificate |
+|---|---|---|---|
+| `v2` (default) | `ServerReader` + `SubscriptionEngine` over the shared pool | v1 or v2 | MinIO + Mongo (`CertProvider`), `Basic256Sha256` |
+| `legacy` | the rc3 `OpcManager`, unchanged, with its own connection | **v1 only** (a v2 slot is rejected and notified) | files from the repository the pod clones (`cert_path` / `private_key_path` in the slot), `Basic256` |
+
+Writes and gRPC always use the pool (MinIO certificate), whatever the read engine.
+The legacy engine needs the certificate files in the Gitea branch the pod clones;
+`scripts/generate-and-push-opc-cert.sh` generates them and pushes them to
+`certs/` (`GITEA_PASSWORD='<senha>' BRANCH=<branch> ./scripts/generate-and-push-opc-cert.sh`).
+Each version deploy creates a new Gitea branch without `certs/`, so they have to be
+copied again for the legacy engine. Remove the legacy engine (OpcManager, its tests
+and the script) once it is unused and the test suite passes the same without it.
 
 ## 🧪 Testing
 
@@ -253,6 +287,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,9 +313,18 @@ 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)):
+OPC servers are configured by the Orchestrator through Redis, one key per slot
+(`slot:opc_tags:<n>`). The gateway reads the **slot v2** structure (contract C1):
+one entry per address, with one `names` item per collection / tag name / model, so
+the same address can feed several models under different names. For secured
+connections set `use_security: true` (the certificate comes from MinIO, see
+[OPC Client Certificate](#-opc-client-certificate)):
 
 ```json
 {
@@ -284,30 +333,30 @@ OPC servers are configured through Redis with the following structure. `cert_pat
       "name": "server1",
       "server_id": "1",
       "url": "opc.tcp://localhost:4841",
-      "server_uri": "http://opcua-server.simulator",
-      "cert_path": "/app/certs/opc-ingestor-cert.pem",
-      "private_key_path": "/app/certs/opc-ingestor-key.pem",
+      "server_uri": "urn:sientia:opc-ingestor",
+      "use_security": true,
+      "subscription_period_ms": 7500,
       "tags": {
         "ns=2;i=2": {
-            "aggr_func": "avg",
-            "data_range": [
-               -100,
-               100
-            ],
-            "frequency": "15000",
-            "server_id": "1",
-            "tag_address": "ns=2;i=2",
-            "tag_name": "Counter",
-            "topics": [
-               "raw_scouter-opcua-orchestrated-pipeline",
-               "raw_scouter-basic-sum-model"
-            ]
+          "frequency": "15000",
+          "active_read": {"enabled": false},
+          "names": [
+            {"collection": "raw_scouter-opcua-orchestrated-pipeline", "tag_name": "Counter", "model_name": "m1", "model_id": "1"},
+            {"collection": "raw_scouter-basic-sum-model", "tag_name": "Contador", "model_name": "m2", "model_id": "2"}
+          ]
+        }
       }
     }
   }
 }
 ```
 
+The legacy **slot v1** (each tag with `tag_name` + `topics`, the server with
+`cert_path`) is still accepted and converted on read (`ingestor/io/slots.py`): one
+`names` item per topic, and `use_security` is true when `cert_path` is set (the path
+itself is ignored). This lets the gateway and the Orchestrator be deployed in any
+order. Documents written from a v1 slot have no `model_id`, exactly as before.
+
 ## 🔧 Development
 
 ### Project Structure
@@ -317,13 +366,14 @@ sientia-dataops-opc-ingestor/
 │   ├── managers/            # Component managers
 │   │   ├── data_manager.py # Data persistence and export
 │   │   ├── ingestor_manager.py # Main coordination
-│   │   ├── opc_manager.py  # OPC UA server management
-│   │   └── resource_manager.py # Resource coordination
+│   │   ├── opc_manager.py  # Legacy reader, plan B (READ_ENGINE=legacy)
+│   │   ├── resource_manager.py # Resource coordination
+│   │   └── server_reader.py # OPC UA server reads (SubscriptionEngine + pool)
 │   ├── app.py              # Main application entry point
 │   ├── ingestor.py         # Core ingestor logic
 │   └── metrics.py          # Prometheus metrics definitions
 ├── simulator/               # OPC simulation and testing tools
-├── scripts/                 # Ops scripts (e.g. OPC UA cert generation)
+├── scripts/                 # Ops scripts (e2e seed; legacy OPC cert generation)
 ├── tests/                   # Test suite
 └── requirements.txt         # Python dependencies
 ```

ingestor/adapters/grpc_server.py

@@ -17,7 +17,9 @@ class ReadTagServicer(opc_gateway_pb2_grpc.OpcGatewayServicer):
         self.active_read = active_read
         self.logger = logger
         # ponytail: slot_lookup wiring (address -> names from the current slot)
-        # lands in Task 18 (app wiring); optional callable keeps this servicer usable now.
+        # 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):

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,37 @@ 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 an ingestor_manager (its data_manager and connection
+    # pool are shared) -- skipped entirely if prepare failed before that point. The
+    # read path (subscriptions) lives in the ingestor_manager and is already running:
+    # a failure here only disables writes/gRPC, the app continues without the 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,
+                pool=ingestor.ingestor_manager.pool,
+                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 +106,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,58 @@
+"""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
+    # Legacy frontend bucket the platform already provisions (SIENTIAPDE-1974 D7);
+    # the design's `sientia-dataops` bucket does not exist in the platform MinIO.
+    cert_bucket: str = 'streamlit-connectors'
+    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 certs_from_env(cls) -> 'GatewayConfig':
+        """Only the certificate settings — what the read path needs. Plain strings,
+        no number parsing, so a malformed write/gRPC env var (GRPC_PORT,
+        WRITE_MIN_PAUSE_S, ...) can only disable the runtime, never the reads."""
+        return cls(
+            cert_bucket=getenv('CERT_BUCKET', cls.cert_bucket),
+            cert_cache_dir=getenv('CERT_CACHE_DIR', cls.cert_cache_dir),
+        )
+
+    @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

@@ -1,12 +1,29 @@
+import asyncio
+import os
+from os import getenv
 from pathlib import Path
 
+from sientia_do.repository.minio_repository import MinioRepository
+
+from ingestor.config import _env_bool
+
 COLLECTION_NAME = 'platform_certificates'
 
 
+class CertificateNotConfiguredError(RuntimeError):
+    """No usable platform OPC client certificate is registered (Mongo + MinIO)."""
+
+
 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."""
+    so `cache_dir` must be an emptyDir/tmpfs mount.
+
+    The certificate is single and shared by every connection (reads, writes,
+    gRPC), so it is downloaded once and cached; `invalidate()` forces a fresh
+    download (the pool calls it when a secured connection fails, e.g. after the
+    certificate was replaced in MinIO). The lock serializes downloads so two
+    connections never rewrite the cache files at the same time."""
 
     def __init__(self, mongo, minio, cache_dir: str, bucket: str, metadata: dict | None = None):
         self.mongo = mongo
@@ -14,16 +31,84 @@ class CertProvider:
         self.cache_dir = cache_dir
         self.bucket = bucket
         self.metadata = metadata or {}
+        self._lock = asyncio.Lock()
+        self._paths: tuple[str, str] | None = None
 
     async def fetch(self) -> tuple[str, str]:
+        async with self._lock:
+            if self._paths is None:
+                self._paths = await self._download_all()
+            return self._paths
+
+    def invalidate(self) -> None:
+        self._paths = None
+
+    async def _download_all(self) -> tuple[str, str]:
         doc = await self.mongo.find_one(COLLECTION_NAME, {}, self.metadata)
+        try:
+            cert_object = doc['client_cert']['minio_object_name']
+            key_object = doc['private_key']['minio_object_name']
+        except (KeyError, TypeError) as e:
+            raise CertificateNotConfiguredError(
+                f'No OPC client certificate configured: the Mongo collection '
+                f"'{COLLECTION_NAME}' needs a document with client_cert.minio_object_name "
+                f'and private_key.minio_object_name (objects in the MinIO bucket '
+                f"'{self.bucket}'). Register the platform certificate before using "
+                f'secured OPC servers.'
+            ) from e
         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'])
+        cert_path = await self._download(cert_object)
+        key_path = await self._download(key_object)
         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
+        local_path = Path(self.cache_dir) / Path(object_name).name
+        # Atomic replace: a connection loading the previous file never sees it half-written.
+        tmp_path = local_path.with_name(local_path.name + '.tmp')
+        tmp_path.write_bytes(data)
+        os.replace(tmp_path, local_path)
+        return str(local_path)
+
+
+class LazyMinio:
+    """Builds the MinIO client on first use. MinioRepository's constructor hits the
+    network (bucket check), so building it eagerly would let a MinIO outage at pod
+    start take the whole read path down. Built lazily, only the servers that need a
+    certificate fail — and they are retried on the next loop."""
+
+    def __init__(self, factory):
+        self._factory = factory
+        self._client = None
+
+    async def download_file(self, object_name: str, bucket: str, metadata: dict):
+        if self._client is None:
+            self._client = self._factory()
+        return await self._client.download_file(object_name, bucket, metadata)
+
+
+def build_cert_provider(
+    mongo, config, logger, notification_handler, metrics_controller, metadata=None
+) -> CertProvider:
+    """CertProvider shared by the read path (IngestorManager) and the write path
+    (GatewayRuntime): one certificate source — MinIO + Mongo — for every connection."""
+
+    def minio_factory() -> MinioRepository:
+        return 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,
+        )
+
+    return CertProvider(
+        mongo=mongo,
+        minio=LazyMinio(minio_factory),
+        cache_dir=config.cert_cache_dir,
+        bucket=config.cert_bucket,
+        metadata=metadata,
+    )

ingestor/connection/pool.py

@@ -26,6 +26,12 @@ class OpcConnectionPool(SientiaMonitoring):
         self.cert_provider = cert_provider
         self.clock = MonotonicClock()
         self._connections: dict[str, tuple[Client, float]] = {}
+        # Active subscriptions per server. A subscribed connection only receives
+        # callbacks and never calls get(), so without a pin release_idle() would
+        # evict it as idle and silently stop the reads (SIENTIAPDE-1974 G4).
+        self._pins: dict[str, int] = {}
+        # Connection settings each pooled client was opened with (see _check_cfg).
+        self._configs: dict[str, dict] = {}
         # 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] = {}
@@ -41,16 +47,35 @@ class OpcConnectionPool(SientiaMonitoring):
         """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)."""
+        same new server share one connection (no leaked, untracked sessions).
+        `server_id` is normalized to str, so 1 and '1' share the same session."""
+        server_id = str(server_id)
         client = self._cached(server_id)
         if client is not None:
+            self._check_cfg(server_id, server_cfg)
             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:
+                self._check_cfg(server_id, server_cfg)
                 return client
             return await self._open(server_id, server_cfg)
 
+    def _check_cfg(self, server_id: str, server_cfg: dict) -> None:
+        """Warn when a caller asks for a server with other connection settings than
+        the pooled client was opened with. The client is reused as is: reconnecting
+        on every mismatch would let one malformed write command drop the reads.
+        Config changes from the slot evict the client first (ServerReader.shutdown)."""
+        opened = self._configs.get(server_id, {})
+        diff = [
+            k for k in ('url', 'server_uri', 'use_security') if opened.get(k) != server_cfg.get(k)
+        ]
+        if diff:
+            self.logger.warning(
+                f'OPC server {server_id}: reusing the pooled connection, but {diff} differ '
+                f'from the settings it was opened with'
+            )
+
     def _cached(self, server_id: str) -> Client | None:
         entry = self._connections.get(server_id)
         if entry is None:
@@ -63,6 +88,10 @@ class OpcConnectionPool(SientiaMonitoring):
         try:
             client = await self._connect(server_cfg)
         except Exception:
+            if server_cfg.get('use_security'):
+                # The cached certificate may be the cause (replaced in MinIO,
+                # rejected by the server): download it again on the next attempt.
+                self.cert_provider.invalidate()
             await self.emit_metric(
                 metric_object=metrics.OPC_CONNECTIONS_FAILED,
                 tags={'pod_id': self.pod_id, 'server_name': server_id},
@@ -70,13 +99,23 @@ class OpcConnectionPool(SientiaMonitoring):
             raise
 
         self._connections[server_id] = (client, self.clock.elapsed())
+        self._configs[server_id] = dict(server_cfg)
         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)
+        # Identify the pod on the OPC server, as the legacy OpcManager did: asyncua
+        # sends `name` as ApplicationName and "<description> Session<n>" as the
+        # session name, so each pod's session is recognizable in the server console.
+        client.name = self.pod_id
+        client.description = self.pod_id
+        pod_uri = self.pod_id.replace('-', ':')
+        client.product_uri = pod_uri
+        client.application_uri = pod_uri
         server_uri = server_cfg.get('server_uri')
         if server_uri:
+            # Must match the certificate's SAN URI (KEPServer: BadCertificateUriInvalid).
             client.application_uri = server_uri
         if server_cfg.get('use_security'):
             cert, key = await self.cert_provider.fetch()
@@ -96,15 +135,39 @@ class OpcConnectionPool(SientiaMonitoring):
             tags={'pod_id': self.pod_id, 'server_name': server_id, 'server_url': url},
         )
 
+    def pin(self, server_id: str) -> None:
+        """Mark one more active subscription on `server_id` (never evicted as idle)."""
+        server_id = str(server_id)
+        self._pins[server_id] = self._pins.get(server_id, 0) + 1
+
+    def unpin(self, server_id: str) -> None:
+        """Drop one active subscription mark from `server_id`."""
+        server_id = str(server_id)
+        count = self._pins.get(server_id, 0) - 1
+        if count > 0:
+            self._pins[server_id] = count
+        else:
+            self._pins.pop(server_id, None)
+
     async def release_idle(self, max_idle_s: float):
-        """Disconnect and drop clients unused for longer than `max_idle_s`."""
+        """Disconnect and drop clients unused for longer than `max_idle_s`,
+        except the ones pinned by an active subscription."""
         now = self.clock.elapsed()
         idle_ids = [
-            sid for sid, (_, last_used) in self._connections.items() if now - last_used > max_idle_s
+            sid
+            for sid, (_, last_used) in self._connections.items()
+            if now - last_used > max_idle_s and not self._pins.get(sid)
         ]
         for server_id in idle_ids:
             await self._evict(server_id)
 
+    async def evict(self, server_id: str):
+        """Disconnect `server_id` now (lost server or changed config); the next
+        get() reconnects. No-op when the server has no pooled client."""
+        server_id = str(server_id)
+        if server_id in self._connections:
+            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,
@@ -114,6 +177,7 @@ class OpcConnectionPool(SientiaMonitoring):
 
     async def _evict(self, server_id: str):
         client, _ = self._connections.pop(server_id)
+        self._configs.pop(server_id, None)
         await self._disconnect_with_retries(client, server_id)
         await self.emit_metric(
             metric_object=metrics.OPC_DISCONNECTIONS_TOTAL,
@@ -122,7 +186,7 @@ class OpcConnectionPool(SientiaMonitoring):
 
     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)."""
+        the legacy OpcManager.disconnection_fallback)."""
         for i in range(5):
             try:
                 await client.disconnect()

ingestor/engines/subscription.py

@@ -35,6 +35,9 @@ class SubscriptionEngine(SientiaMonitoring):
         self.subscriptions: dict = {}
         self._clients: dict = {}
         self._servers: dict = {}
+        # Addresses each subscription monitors, so unsubscribe can forget them
+        # (otherwise check_cycles keeps alarming on tags nobody reads anymore).
+        self._sub_nodes: dict[str, set[str]] = {}
         self.non_receive_count = 0
 
         SientiaMonitoring.__init__(
@@ -57,6 +60,7 @@ class SubscriptionEngine(SientiaMonitoring):
             raise
         self._clients[name] = client
         self._servers[name] = server_id
+        self.pool.pin(server_id)
         self.logger.info(f'Subscription {name} created on {server_id}.')
         await self.emit_metric(
             metric_object=metrics.OPC_SUBSCRIPTIONS_CREATED,
@@ -74,6 +78,7 @@ class SubscriptionEngine(SientiaMonitoring):
         server_id = self._servers[subscription]
         addr_nodes = [client.get_node(n) for n in nodes]
         self.nodes.update(nodes)
+        self._sub_nodes.setdefault(subscription, set()).update(nodes)
         self._init_cycle_rules(nodes, collect_period, server_id)
 
         await self.subscriptions[subscription].subscribe_data_change(addr_nodes)
@@ -96,11 +101,35 @@ class SubscriptionEngine(SientiaMonitoring):
         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]
+        sub = self.subscriptions.pop(subscription)
         self._clients.pop(subscription, None)
-        self._servers.pop(subscription, None)
+        server_id = self._servers.pop(subscription, None)
+        if server_id is not None:
+            self.pool.unpin(server_id)
+        self._forget_nodes(subscription)
+        try:
+            await sub.delete()
+        except Exception as e:
+            # The connection may already be gone (lost server); the subscription is
+            # dropped from the books either way so the slot can be re-subscribed.
+            self.logger.warning(f'Failed to delete subscription {subscription}: {e}')
         self.logger.info(f'Unsubscribed from {subscription}.')
+        if server_id is not None:
+            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 _forget_nodes(self, subscription: str) -> None:
+        """Drop the subscription's addresses, except those another subscription
+        of this engine still monitors."""
+        nodes = self._sub_nodes.pop(subscription, set())
+        still_monitored = set().union(*self._sub_nodes.values())
+        for node in nodes:
+            if node not in still_monitored:
+                self.nodes.pop(node, None)
 
     async def datachange_notification(self, node, _val, data):
         """asyncua subscription callback: fan the new value out to every name

ingestor/gateway_runtime.py

@@ -0,0 +1,134 @@
+"""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. The OpcConnectionPool (and its CertProvider,
+MinIO + Mongo) belongs to the read path in IngestorManager (SIENTIAPDE-1974):
+reads start in prepare_ingestor, before this runtime exists, and must keep running
+if anything here fails. The runtime only borrows the pool for writes and gRPC.
+
+# ponytail: scheduled_tick + gRPC slot_lookup wiring land in the next 1974 RC.
+"""
+
+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.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
+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,
+        pool: OpcConnectionPool,
+        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
+
+        self.pool = pool  # owned by IngestorManager (read path); closed there
+
+        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
+        # The pool is not closed here: IngestorManager.shutdown() owns it and runs
+        # after this, once its readers have dropped their subscriptions.
+
+    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

@@ -21,15 +21,16 @@ class DataSink:
         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."""
+        """Publish one MongoDB document per name entry and record its tag value metric.
+
+        `model_id` is only written when the name carries one: names converted from a
+        slot v1 have none, and their documents must stay identical to the legacy ones."""
         for name in names:
-            doc = {
-                'tag': tag,
-                'name': name['tag_name'],
-                'model_id': name['model_id'],
-                'timestamp': timestamp,
-                'value': value,
-            }
+            doc = {'tag': tag, 'name': name['tag_name']}
+            if 'model_id' in name:
+                doc['model_id'] = name['model_id']
+            doc['timestamp'] = timestamp
+            doc['value'] = value
             await self.data_manager.publish(name['collection'], doc)
             await self.emit_metric(
                 metric_object=metrics.OPC_TAG_VALUE,
@@ -38,7 +39,7 @@ class DataSink:
                 tags={
                     'pod_id': self.data_manager.pod_id,
                     'tag_name': name['tag_name'],
-                    'model_id': name['model_id'],
+                    'model_id': name.get('model_id', ''),
                 },
             )
 

ingestor/io/slots.py

@@ -1,25 +1,47 @@
-"""Parse the slot v2 structure written by the Orchestrator to Redis (contract C1)."""
+"""Parse the slot written by the Orchestrator to Redis into the slot v2 structure (contract C1).
+
+Both shapes are accepted so the gateway and the Orchestrator can be deployed in
+any order (SIENTIAPDE-1974):
+
+- v2: each tag carries `names` (one entry per collection / tag name / model);
+- v1 (legacy): each tag carries `tag_name` + `topics`, and the server carries the
+  file-based `cert_path`. v1 is converted to v2: one `names` entry per topic, and
+  `use_security` is true when the server had a certificate path (the certificate
+  itself now always comes from MinIO via CertProvider, never from that path).
+"""
 
 SERVER_FIELDS = ('server_id', 'url', 'server_uri', 'use_security', 'subscription_period_ms')
+# What a connection/subscription needs (the pool keys clients by server_id itself).
+CONNECTION_FIELDS = tuple(f for f in SERVER_FIELDS if f != 'server_id')
+
+
+def _parse_names(address: str, tag: dict) -> list[dict]:
+    names = tag.get('names')
+    if names:
+        return names
+    topics = tag.get('topics')
+    if topics and tag.get('tag_name'):
+        return [{'collection': topic, 'tag_name': tag['tag_name']} for topic in topics]
+    raise ValueError(f'tag {address!r} has no names')
 
 
 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,
+        'names': _parse_names(address, tag),
         'active_read': tag.get('active_read', {'enabled': False}),
     }
 
 
-def _parse_server(server: dict) -> dict:
+def _parse_server(name: str, server: dict) -> dict:
     parsed = {field: server[field] for field in SERVER_FIELDS if field in server}
+    parsed['name'] = server.get('name', name)
+    if 'use_security' not in parsed:
+        parsed['use_security'] = bool(server.get('cert_path'))
     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()}
+    """Normalize a raw slot (v1 or v2) into `{server_name: {..., tags: {addr: tag}}}`."""
+    return {name: _parse_server(name, server) for name, server in raw.items()}

ingestor/managers/ingestor_manager.py

@@ -1,5 +1,6 @@
 import traceback
 from copy import deepcopy
+from os import getenv
 
 from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
 from sientia_do.notifications.models import NotificationLevel
@@ -8,9 +9,21 @@ from sientia_do.observability.metrics_controller import MetricsController
 from sientia_do.observability.sientia_monitoring import SientiaMonitoring
 
 import ingestor.metrics as metrics
+from ingestor.config import GatewayConfig
+from ingestor.connection.certs import build_cert_provider
+from ingestor.connection.pool import OpcConnectionPool
+from ingestor.io.data_sink import DataSink
+from ingestor.io.slots import parse_slot
 from ingestor.managers.data_manager import DataManager
 from ingestor.managers.opc_manager import OpcManager
 from ingestor.managers.resource_manager import ResourceManager
+from ingestor.managers.server_reader import ServerReader
+
+# READ_ENGINE (SIENTIAPDE-1974): 'v2' = ServerReader/SubscriptionEngine over the shared
+# pool, certificate from MinIO (default). 'legacy' = plan B: the untouched rc3
+# OpcManager, slot v1 only, certificate files from the Gitea clone (cert_path). Kept
+# for validation until the end of 1974; remove once unused and the tests pass without it.
+READ_ENGINES = ('v2', 'legacy')
 
 
 class IngestorManager(SientiaMonitoring):
@@ -47,7 +60,10 @@ class IngestorManager(SientiaMonitoring):
 
     Attributes:
         data_manager (DataManager): Manages data persistence
-        opc_managers (dict): Dictionary of OPC managers keyed by server name
+        opc_managers (dict): ServerReader per OPC server, keyed by server name
+        pool (OpcConnectionPool): OPC connections shared by reads, writes and gRPC;
+            owned here because reads start in prepare_ingestor, before GatewayRuntime
+        sink (DataSink): read egress (fan-out to every slot v2 `names` entry)
         resource_manager (ResourceManager): Manages Redis-based resource coordination
         number_of_slots (int): Total number of slots configured in the system
         poll_interval (int): Main loop polling interval
@@ -68,6 +84,8 @@ class IngestorManager(SientiaMonitoring):
         logger: Logger,
         notification_handler: NotificationHandler,
         metrics_controller: MetricsController,
+        gateway_config: GatewayConfig | None = None,
+        read_engine: str | None = None,
     ):
         redis_host: str = redis_data['host']
         redis_port: int = int(redis_data['port'])
@@ -90,6 +108,26 @@ class IngestorManager(SientiaMonitoring):
             metrics_controller=metrics_controller,
         )
         self.opc_managers: dict = {}
+        # Read path (SIENTIAPDE-1974): pooled connections with the certificate from
+        # MinIO. Nothing here touches the network at construction (MinIO is lazy), so
+        # a MinIO/Postgres/NATS outage can't stop the reads before they start.
+        config = gateway_config or GatewayConfig.certs_from_env()
+        self.cert_provider = build_cert_provider(
+            mongo=self.data_manager.mongo_repository,
+            config=config,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+            metadata=metadata,
+        )
+        self.pool = OpcConnectionPool(
+            self.cert_provider, logger, notification_handler, metrics_controller
+        )
+        self.sink = DataSink(self.data_manager, None)
+        self._server_cert_warned: set[str] = set()
+        self.read_engine = self._resolve_read_engine(
+            read_engine or getenv('READ_ENGINE') or 'v2', logger
+        )
         self.resource_manager = ResourceManager(
             host=redis_host,
             port=redis_port,
@@ -109,36 +147,19 @@ class IngestorManager(SientiaMonitoring):
 
         self.metadata = metadata
 
-    async def initialize_opc_from_config(self, server_config: dict) -> OpcManager | None:
-        """
-        Initializes an OPC Manager instance using the provided server configuration.
+    @staticmethod
+    def _resolve_read_engine(value: str, logger: Logger) -> str:
+        engine = value.strip().lower()
+        if engine not in READ_ENGINES:
+            logger.warning(f'Invalid READ_ENGINE {value!r}, using v2 ({"|".join(READ_ENGINES)})')
+            engine = 'v2'
+        logger.info(f'OPC read engine: {engine}')
+        return engine
 
-        This method creates and configures an OPC Manager for a specific OPC UA server,
-        establishing the connection and preparing it for tag subscriptions.
-
-        Args:
-            server_config (dict): A dictionary containing the OPC server configuration.
-                Expected keys include:
-                    - 'name' (str): The name of the OPC server.
-                    - 'url' (str): The URL of the OPC server.
-                    - 'server_uri' (str): The URI of the OPC server.
-                    - 'cert_path' (str, optional): Path to the client certificate file.
-                    - 'private_key_path' (str, optional): Path to the private key file.
-                    - 'server_cert_path' (str, optional): Path to the server certificate file.
-
-        Returns:
-            OpcManager | None: An initialized OpcManager instance if successful,
-            otherwise None if an error occurs during initialization.
-
-        Raises:
-            Exception: If OPC manager initialization fails, the error is logged and
-                      a notification is sent, but the method returns None to allow
-                      the system to continue operating with other servers.
-        """
-
-        try:
-            self.logger.info(f'Initializing OpcManager at {server_config["url"]}')
-            manager = OpcManager(
+    def _build_reader(self, server_config: dict) -> ServerReader | OpcManager:
+        if self.read_engine == 'legacy':
+            # Plan B: the rc3 OpcManager, unchanged (own connection, cert files).
+            return OpcManager(
                 name=server_config['name'],
                 url=server_config['url'],
                 subscription_period_ms=server_config['subscription_period_ms'],
@@ -152,6 +173,41 @@ class IngestorManager(SientiaMonitoring):
                 server_cert_path=server_config.get('server_cert_path'),
                 metrics_controller=self.metrics_controller,
             )
+        return ServerReader(
+            name=server_config['name'],
+            server_config=server_config,
+            pool=self.pool,
+            sink=self.sink,
+            logger=self.logger,
+            notification_handler=self.notification_handler,
+            metrics_controller=self.metrics_controller,
+            metadata=self.metadata,
+        )
+
+    async def initialize_opc_from_config(
+        self, server_config: dict
+    ) -> ServerReader | OpcManager | None:
+        """
+        Initializes the reader of one OPC UA server from its slot configuration.
+
+        With READ_ENGINE=v2 (default) creates a ServerReader and opens its pooled
+        connection; with `use_security`, the certificate comes from MinIO. With
+        READ_ENGINE=legacy (plan B) creates the rc3 OpcManager, which connects on its
+        own with the certificate files of the slot v1 (`cert_path`).
+
+        Args:
+            server_config (dict): The server configuration of the slot, without the
+                'tags' key. v2: 'name', 'server_id', 'url', 'server_uri',
+                'use_security', 'subscription_period_ms'. legacy: the slot v1 fields.
+
+        Returns:
+            ServerReader | OpcManager | None: The connected reader, or None if the
+            connection fails (logged and notified, so the other servers keep working).
+        """
+
+        try:
+            self.logger.info(f'Initializing OPC reader at {server_config["url"]}')
+            manager = self._build_reader(server_config)
 
             manager.config = server_config
             await manager.connect()
@@ -176,23 +232,33 @@ class IngestorManager(SientiaMonitoring):
         Gracefully shuts down the IngestorManager and all its components.
 
         This method ensures proper cleanup of:
-        - All OPC manager instances and their connections
+        - All OPC readers, their subscriptions and the pooled connections
         - Data manager connections and resources
-        - Active subscriptions and server connections
 
         The shutdown process is performed asynchronously to allow proper cleanup
         of all managed resources before termination.
         """
         for _server_name, server in self.opc_managers.items():
             await server.shutdown()
+        await self.pool.close_all()
         self.data_manager.shutdown()
 
-    async def remove_server(self, server: str):
+    async def remove_server(self, server: str, evict: bool = True):
         """
         Removes an OPC server from the ingestor.
+
+        Args:
+            server (str): The server name.
+            evict (bool): Also drop the pooled connection (lost server, failed
+                subscription). False when the server merely left this pod's slots:
+                the shared connection stays for writes/gRPC until the idle sweep.
         """
         if server in self.opc_managers:
-            await self.opc_managers[server].shutdown()
+            if self.read_engine == 'legacy':
+                # OpcManager owns its connection: shutdown always disconnects it.
+                await self.opc_managers[server].shutdown()
+            else:
+                await self.opc_managers[server].shutdown(evict=evict)
             del self.opc_managers[server]
         for slot, _config in self.managed_tags.items():
             self.managed_tags[slot].pop(server, None)
@@ -258,9 +324,10 @@ class IngestorManager(SientiaMonitoring):
         for server in servers:
             if server not in registered_servers:
                 self.logger.warning(
-                    f'Server {server} not found in managed tags. Desconnecting from server.'
+                    f'Server {server} not found in managed tags. Unsubscribing from server.'
                 )
-                await self.remove_server(server)
+                # Slot moved away: keep the shared connection for writes/gRPC.
+                await self.remove_server(server, evict=False)
 
         await self.emit_metric(
             metric_object=metrics.OPC_MANAGERS_ACTIVE,
@@ -419,14 +486,18 @@ class IngestorManager(SientiaMonitoring):
 
         Notes:
             - If a slot is leased but its details cannot be retrieved
-              (i.e., `get_tag_slot` returns None), that slot is skipped.
+              (i.e., `get_tag_slot` returns None) or are malformed, that slot is skipped.
+            - Slot details are normalized to the slot v2 structure (parse_slot).
         """
 
         acquired = {}
         for i in range(1, self.number_of_slots + 1):
             if await self.resource_manager.lease_tag(str(i)):
                 self.logger.info(f'Leased slot {i}')
-                slots = await self.resource_manager.get_tag_slot(str(i))
+                raw = await self.resource_manager.get_tag_slot(str(i))
+                if raw is None:
+                    continue
+                slots = self.parse_slot_config(str(i), raw)
                 if slots is None:
                     continue
                 acquired[str(i)] = slots
@@ -518,16 +589,75 @@ class IngestorManager(SientiaMonitoring):
         removed_slots: list[str] = []
         for slot, _slot_config in self.managed_tags.items():
             await self.resource_manager.renew_tag_lease(slot)
-            update = await self.resource_manager.get_tag_slot(slot)
-            if update is None:
+            raw = await self.resource_manager.get_tag_slot(slot)
+            if raw is None:
                 removed_slots.append(slot)
                 continue
 
+            update = self.parse_slot_config(slot, raw)
+            if update is None:
+                # Malformed config: keep reading with the last good one.
+                continue
             self.managed_tags[slot] = update
 
         for slot in removed_slots:
             self.managed_tags.pop(slot, None)
 
+    def parse_slot_config(self, slot: str, raw: dict) -> dict | None:
+        """
+        Normalizes the raw slot read from Redis to the slot v2 structure.
+
+        Accepts slot v1 (`topics`) and v2 (`names`), so the gateway and the
+        Orchestrator can be deployed in any order (SIENTIAPDE-1974). With
+        READ_ENGINE=legacy the raw slot v1 is kept as is (what OpcManager reads), and
+        a slot v2 is rejected.
+
+        Returns:
+            dict | None: The parsed slot, or None when the config is malformed
+            (logged and notified).
+        """
+        try:
+            if self.read_engine == 'legacy':
+                self._validate_slot_v1(raw)
+                return raw
+            parsed = parse_slot(raw)
+        except (AttributeError, KeyError, TypeError, ValueError) as e:
+            self.send_notification(
+                metadata=self.metadata,
+                notification_id=f'OPC_SLOT_CONFIG_ERROR_{slot}',
+                message=f'Slot {slot} has an invalid configuration: {e!r}',
+                block='ingestor_manager',
+                level=NotificationLevel.ERROR,
+            )
+            self.logger.error(f'Slot {slot} has an invalid configuration: {e!r}')
+            return None
+        self._warn_server_cert_path(raw)
+        return parsed
+
+    @staticmethod
+    def _validate_slot_v1(raw: dict) -> None:
+        """READ_ENGINE=legacy: OpcManager only understands slot v1 (`tag_name` +
+        `topics` per tag); a v2 slot (`names`) would fail in its datachange callback."""
+        for name, server in raw.items():
+            for address, tag in server['tags'].items():
+                if 'topics' not in tag or 'tag_name' not in tag or 'frequency' not in tag:
+                    raise ValueError(
+                        f'READ_ENGINE=legacy needs a slot v1 (tag_name + topics); '
+                        f'server {name!r}, tag {address!r} is not'
+                    )
+
+    def _warn_server_cert_path(self, raw: dict) -> None:
+        """Logs once per server when a slot still pins a server certificate
+        (`server_cert_path`): connections now come from the pool with the platform
+        certificate from MinIO, and that file-based pin is no longer applied."""
+        for name, server in raw.items():
+            if server.get('server_cert_path') and name not in self._server_cert_warned:
+                self._server_cert_warned.add(name)
+                self.logger.warning(
+                    f'OPC server {name}: server_cert_path is set in the slot but is no '
+                    f'longer applied (server certificate validation is disabled)'
+                )
+
     async def drop_slot_leases(self, ids: list[str]) -> None:
         """
         Releases the leases associated with the specified slot IDs.

ingestor/managers/server_reader.py

@@ -0,0 +1,107 @@
+"""Per-server read path driven by the IngestorManager (SIENTIAPDE-1974).
+
+Replaces the legacy OpcManager behind the same interface the IngestorManager
+already uses (connect / create_subscription / subscribe / unsubscribe /
+check_cycles / check_opc_listenning / shutdown), so the slot lease loop is
+unchanged. The difference is underneath:
+
+- the connection comes from the shared OpcConnectionPool, so the certificate is
+  fetched from MinIO by CertProvider (Basic256Sha256) — never from a file shipped
+  in the Gitea repository;
+- readings go through SubscriptionEngine, which fans each value out to every
+  `names` entry of the slot v2 (one document per collection / tag name / model).
+"""
+
+from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+from sientia_do.observability.logger import Logger
+from sientia_do.observability.metrics_controller import MetricsController
+
+import ingestor.metrics as metrics
+from ingestor.connection.pool import OpcConnectionPool
+from ingestor.engines.subscription import SubscriptionEngine
+from ingestor.io.data_sink import DataSink
+from ingestor.io.slots import CONNECTION_FIELDS
+
+
+class ServerReader:
+    """Reads one OPC server: one SubscriptionEngine (its `non_receive_count` is
+    engine-wide, hence one per server) with one subscription per leased slot."""
+
+    def __init__(
+        self,
+        name: str,
+        server_config: dict,
+        pool: OpcConnectionPool,
+        sink: DataSink,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+        metadata: dict | None = None,
+    ):
+        self.name = name
+        self.server_id = str(server_config['server_id'])
+        self.url = server_config['url']
+        self.server_cfg = {k: server_config[k] for k in CONNECTION_FIELDS if k in server_config}
+        self.pool = pool
+        self.logger = logger
+        self.config: dict | None = None
+        self.engine = SubscriptionEngine(
+            pool, sink, logger, notification_handler, metrics_controller, metadata
+        )
+
+    def __str__(self):
+        return f'ServerReader(name={self.name}, server_id={self.server_id}, url={self.url})'
+
+    @property
+    def subscriptions(self) -> dict:
+        return self.engine.subscriptions
+
+    async def connect(self):
+        """Open (or reuse) the pooled connection; raises if the server is unreachable
+        or the certificate can't be fetched, like OpcManager.connect did."""
+        await self.pool.get(self.server_id, self.server_cfg)
+
+    async def create_subscription(self, slot: str):
+        await self.engine.create_subscription(self.server_id, self.server_cfg, slot)
+
+    async def subscribe(self, slot: str, nodes: dict, collect_period: int):
+        await self.engine.subscribe(slot, nodes, collect_period)
+
+    async def unsubscribe(self, slot: str):
+        await self.engine.unsubscribe(slot)
+
+    async def check_cycles(self):
+        await self.engine.check_cycles()
+
+    async def check_opc_listenning(self) -> bool:
+        return await self.engine.check_opc_listenning(self.server_id)
+
+    async def shutdown(self, evict: bool = True):
+        """Drop every subscription of this server.
+
+        With `evict` (lost server or changed server config) the pooled connection is
+        also dropped, so the next initialization reconnects. Without it (the slot just
+        moved to another pod) the connection stays for writes/gRPC: the connection is
+        shared, and the idle sweep closes it if nobody uses it anymore."""
+        # list() snapshots the keys: unsubscribe() pops from engine.subscriptions, so
+        # iterating the dict directly would raise "changed size during iteration".
+        for slot in list(self.engine.subscriptions):  # NOSONAR
+            await self.engine.unsubscribe(slot)
+        if evict:
+            await self.pool.evict(self.server_id)
+            await self.engine.emit_metric(
+                metric_object=metrics.OPC_CONNECTION_STATUS,
+                method='set',
+                value=0,
+                tags={
+                    'pod_id': self.engine.pod_id,
+                    'server_name': self.server_id,
+                    'server_url': self.url,
+                },
+            )
+        await self.engine.emit_metric(
+            metric_object=metrics.OPC_TAGS_SUBSCRIBED,
+            method='set',
+            value=0,
+            tags={'pod_id': self.engine.pod_id, 'server_name': self.server_id},
+        )

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "ingestor"
-version = "2.0.0-rc2"
+version = "2.1.0-rc1"
 description = "Sientia DataOps Ingestor - OPC Tag Ingestor"
 readme = "README.md"
 requires-python = ">=3.11"

requirements-dev.txt

@@ -17,6 +17,7 @@ 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
+python-dotenv>=1.0    # local dev only: app/seed read .env if present (in-cluster env comes from Helm)
 
 # gRPC code generation (regenerate ingestor/proto/*_pb2*.py) — not needed
 # at runtime: the generated stubs are committed. Kept out of requirements.txt

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())

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/connection/test_certs.py

@@ -26,3 +26,123 @@ async def test_fetch_returns_local_paths(tmp_path):
     )
     assert (tmp_path / 'client_cert_c.pem').read_bytes() == b'cert-bytes'
     assert (tmp_path / 'private_key_k.pem').read_bytes() == b'key-bytes'
+
+
+@pytest.mark.asyncio
+async def test_lazy_minio_builds_client_once_on_first_download():
+    from unittest.mock import MagicMock
+
+    from ingestor.connection.certs import LazyMinio
+
+    client = AsyncMock()
+    client.download_file.return_value = b'bytes'
+    factory = MagicMock(return_value=client)
+    minio = LazyMinio(factory)
+
+    factory.assert_not_called()  # no network at construction
+    assert await minio.download_file('a.pem', 'bucket', {}) == b'bytes'
+    await minio.download_file('b.pem', 'bucket', {})
+
+    factory.assert_called_once()
+    assert client.download_file.await_count == 2
+
+
+def test_build_cert_provider_is_lazy_and_uses_config(monkeypatch):
+    from unittest.mock import MagicMock
+
+    from ingestor.config import GatewayConfig
+    from ingestor.connection import certs
+
+    minio_cls = MagicMock()
+    monkeypatch.setattr(certs, 'MinioRepository', minio_cls)
+    mongo = MagicMock()
+
+    provider = certs.build_cert_provider(
+        mongo=mongo,
+        config=GatewayConfig(cert_bucket='streamlit-connectors', cert_cache_dir='certs-cache'),
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(),
+    )
+
+    assert provider.mongo is mongo
+    assert provider.bucket == 'streamlit-connectors'
+    assert provider.cache_dir == 'certs-cache'
+    assert isinstance(provider.minio, certs.LazyMinio)
+    minio_cls.assert_not_called()  # MinIO is only reached when a cert is needed
+
+
+def _provider(tmp_path, doc=None, data=b'pem'):
+    from unittest.mock import MagicMock
+
+    mongo = MagicMock()
+    mongo.find_one = AsyncMock(
+        return_value=doc
+        if doc is not None
+        else {
+            'client_cert': {'minio_object_name': 'platform_certificates/c.pem'},
+            'private_key': {'minio_object_name': 'platform_certificates/k.pem'},
+        }
+    )
+    minio = MagicMock()
+    minio.download_file = AsyncMock(return_value=data)
+    return CertProvider(mongo=mongo, minio=minio, cache_dir=str(tmp_path), bucket='b'), minio
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'doc',
+    [
+        None,
+        {},
+        {'client_cert': {'minio_object_name': 'c.pem'}},
+        {'client_cert': None, 'private_key': {}},
+    ],
+)
+async def test_fetch_without_platform_certificate_raises_clear_error(tmp_path, doc):
+    from ingestor.connection.certs import CertificateNotConfiguredError
+
+    provider, minio = _provider(tmp_path)
+    provider.mongo.find_one.return_value = doc
+
+    with pytest.raises(CertificateNotConfiguredError, match='platform_certificates'):
+        await provider.fetch()
+    minio.download_file.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_fetch_is_cached_until_invalidated(tmp_path):
+    provider, minio = _provider(tmp_path)
+
+    first = await provider.fetch()
+    second = await provider.fetch()
+    assert first == second
+    assert minio.download_file.await_count == 2  # cert + key, once
+
+    provider.invalidate()
+    await provider.fetch()
+    assert minio.download_file.await_count == 4
+
+
+@pytest.mark.asyncio
+async def test_concurrent_fetches_download_once(tmp_path):
+    import asyncio
+
+    provider, minio = _provider(tmp_path)
+
+    results = await asyncio.gather(*(provider.fetch() for _ in range(5)))
+
+    assert len(set(results)) == 1
+    assert minio.download_file.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_download_replaces_file_atomically(tmp_path):
+    provider, minio = _provider(tmp_path, data=b'new')
+    (tmp_path / 'c.pem').write_bytes(b'old')
+
+    cert_path, _ = await provider.fetch()
+
+    assert (tmp_path / 'c.pem').read_bytes() == b'new'
+    assert cert_path == str(tmp_path / 'c.pem')
+    assert not list(tmp_path.glob('*.tmp'))

tests/unit/connection/test_pool.py

@@ -182,3 +182,163 @@ async def test_concurrent_get_same_server_connects_once():
     assert c1 is c2 is clients[0]
     assert ctor.call_count == 1
     assert clients[0].connect.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_release_idle_keeps_pinned_connection():
+    """A subscribed connection only gets callbacks (never get()), so it looks idle:
+    the pin keeps release_idle from dropping it (SIENTIAPDE-1974 G4)."""
+    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.pin('srv')
+    await pool.release_idle(max_idle_s=0)
+
+    assert 'srv' in pool._connections
+    fake_client.disconnect.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_release_idle_evicts_after_last_unpin():
+    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.pin('srv')
+    pool.pin('srv')  # two slots subscribed on the same server
+    pool.unpin('srv')
+    await pool.release_idle(max_idle_s=0)
+    assert 'srv' in pool._connections  # one subscription still active
+
+    pool.unpin('srv')
+    await pool.release_idle(max_idle_s=0)
+    assert 'srv' not in pool._connections
+    fake_client.disconnect.assert_awaited_once()
+
+
+def test_unpin_never_goes_negative():
+    pool = make_pool()
+
+    pool.unpin('srv')
+    pool.pin('srv')
+
+    assert pool._pins == {'srv': 1}
+
+
+@pytest.mark.asyncio
+async def test_evict_disconnects_now_and_next_get_reconnects():
+    pool = make_pool()
+    first, second = AsyncMock(), AsyncMock()
+    with patch('ingestor.connection.pool.Client', side_effect=[first, second]):
+        await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
+        await pool.evict('srv')
+        client = await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
+
+    first.disconnect.assert_awaited_once()
+    assert client is second
+
+
+@pytest.mark.asyncio
+async def test_evict_unknown_server_is_noop():
+    pool = make_pool()
+
+    await pool.evict('nope')
+
+    pool.emit_metric.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_get_normalizes_server_id_to_str():
+    pool = make_pool()
+    fake_client = AsyncMock()
+    with patch('ingestor.connection.pool.Client', return_value=fake_client):
+        c1 = await pool.get(1, {'url': 'opc.tcp://h:4840', 'use_security': False})
+        c2 = await pool.get('1', {'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_with_different_cfg_reuses_client_and_warns():
+    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 = await pool.get('srv', {'url': 'opc.tcp://10.0.0.5:4840', 'use_security': False})
+
+    assert client is fake_client
+    assert fake_client.connect.await_count == 1
+    pool.logger.warning.assert_called_once()
+    assert 'url' in pool.logger.warning.call_args.args[0]
+
+
+@pytest.mark.asyncio
+async def test_secured_connect_failure_invalidates_cert_cache():
+    from unittest.mock import MagicMock
+
+    cert_provider = MagicMock()
+    cert_provider.fetch = AsyncMock(return_value=('cert.pem', 'key.pem'))
+    pool = make_pool(cert_provider=cert_provider)
+    fake_client = AsyncMock()
+    fake_client.connect.side_effect = RuntimeError('BadCertificateUntrusted')
+    with (
+        patch('ingestor.connection.pool.Client', return_value=fake_client),
+        patch('ingestor.connection.pool.apply_security', AsyncMock()),
+    ):
+        with pytest.raises(RuntimeError):
+            await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': True})
+
+    cert_provider.invalidate.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_insecure_connect_failure_keeps_cert_cache():
+    from unittest.mock import MagicMock
+
+    cert_provider = MagicMock()
+    pool = make_pool(cert_provider=cert_provider)
+    fake_client = AsyncMock()
+    fake_client.connect.side_effect = RuntimeError('unreachable')
+    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})
+
+    cert_provider.invalidate.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_connect_identifies_the_pod_on_the_server():
+    """Session/application name = pod id, as the legacy OpcManager did, so each pod's
+    session is recognizable in the KEPServer console."""
+    pool = make_pool()
+    fake_client = MagicMock(connect=AsyncMock())
+    with patch('ingestor.connection.pool.Client', return_value=fake_client):
+        await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
+
+    assert fake_client.name == pool.pod_id
+    assert fake_client.description == pool.pod_id
+    assert fake_client.product_uri == pool.pod_id.replace('-', ':')
+    assert fake_client.application_uri == pool.pod_id.replace('-', ':')
+
+
+@pytest.mark.asyncio
+async def test_connect_server_uri_still_wins_for_application_uri():
+    pool = make_pool()
+    fake_client = MagicMock(connect=AsyncMock())
+    with patch('ingestor.connection.pool.Client', return_value=fake_client):
+        await pool.get(
+            'srv',
+            {
+                'url': 'opc.tcp://h:4840',
+                'use_security': False,
+                'server_uri': 'urn:sientia:opc-ingestor',
+            },
+        )
+
+    # must match the certificate SAN URI (BadCertificateUriInvalid otherwise)
+    assert fake_client.application_uri == 'urn:sientia:opc-ingestor'
+    assert fake_client.name == pool.pod_id

tests/unit/engines/test_subscription.py

@@ -201,3 +201,100 @@ async def test_check_opc_listenning_triggers_reconnect_at_15():
 
     assert result is True
     assert eng.send_notification_async.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_create_subscription_pins_pool_connection():
+    eng = make_engine()
+    eng.pool = MagicMock(get=AsyncMock(return_value=AsyncMock()))
+
+    await eng.create_subscription('1', {'subscription_period_ms': 500}, 'slot1')
+
+    eng.pool.pin.assert_called_once_with('1')
+
+
+@pytest.mark.asyncio
+async def test_create_subscription_failure_does_not_pin():
+    eng = make_engine()
+    client = AsyncMock()
+    client.create_subscription.side_effect = RuntimeError('boom')
+    eng.pool = MagicMock(get=AsyncMock(return_value=client))
+
+    with pytest.raises(RuntimeError):
+        await eng.create_subscription('1', {'subscription_period_ms': 500}, 'slot1')
+
+    eng.pool.pin.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_unsubscribe_unpins_pool_connection():
+    eng = make_engine()
+    eng.pool = MagicMock(get=AsyncMock(return_value=AsyncMock()))
+    await eng.create_subscription('1', {'subscription_period_ms': 500}, 'slot1')
+
+    await eng.unsubscribe('slot1')
+
+    eng.pool.unpin.assert_called_once_with('1')
+    assert 'slot1' not in eng.subscriptions
+
+
+@pytest.mark.asyncio
+async def test_unsubscribe_tolerates_dead_connection():
+    """A lost server's subscription can't be deleted on the server side; it must
+    still be dropped locally (and unpinned) so the slot can be re-subscribed."""
+    eng = make_engine()
+    sub = AsyncMock()
+    sub.delete.side_effect = ConnectionError('closed')
+    client = AsyncMock()
+    client.create_subscription.return_value = sub
+    eng.pool = MagicMock(get=AsyncMock(return_value=client))
+    await eng.create_subscription('1', {'subscription_period_ms': 500}, 'slot1')
+
+    await eng.unsubscribe('slot1')  # must not raise
+
+    assert 'slot1' not in eng.subscriptions
+    eng.pool.unpin.assert_called_once_with('1')
+    eng.logger.warning.assert_called_once()
+
+
+async def _subscribed_engine(slots):
+    """Engine with one subscription per slot: {slot: [addresses]}."""
+    eng = make_engine()
+    eng.pool = MagicMock(
+        get=AsyncMock(
+            return_value=MagicMock(create_subscription=AsyncMock(return_value=AsyncMock()))
+        )
+    )
+    for slot, addresses in slots.items():
+        await eng.create_subscription('1', {'subscription_period_ms': 500}, slot)
+        nodes = {
+            a: {'frequency': '1000', 'names': [{'collection': 'c', 'tag_name': a}]}
+            for a in addresses
+        }
+        await eng.subscribe(slot, nodes, 10)
+    return eng
+
+
+@pytest.mark.asyncio
+async def test_unsubscribe_forgets_the_slot_nodes():
+    eng = await _subscribed_engine({'s1': ['ns=2;i=1', 'ns=2;i=2'], 's2': ['ns=2;i=3']})
+
+    await eng.unsubscribe('s1')
+
+    # no more cycle checks / tag counting for addresses nobody reads anymore
+    assert set(eng.nodes) == {'ns=2;i=3'}
+    eng.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_TAGS_SUBSCRIBED,
+        method='set',
+        value=1,
+        tags={'pod_id': eng.pod_id, 'server_name': '1'},
+    )
+
+
+@pytest.mark.asyncio
+async def test_unsubscribe_keeps_node_shared_with_another_slot():
+    eng = await _subscribed_engine({'s1': ['ns=2;i=1', 'ns=2;i=2'], 's2': ['ns=2;i=2']})
+
+    await eng.unsubscribe('s1')
+
+    assert set(eng.nodes) == {'ns=2;i=2'}

tests/unit/io/test_data_sink_publish.py

@@ -67,3 +67,26 @@ async def test_publish_reading_emits_tag_value_metric_per_name():
         value=42,
         tags={'pod_id': dm.pod_id, 'tag_name': 'Counter', 'model_id': '1'},
     )
+
+
+@pytest.mark.asyncio
+async def test_publish_reading_without_model_id_keeps_legacy_doc():
+    """Names converted from a slot v1 carry no model_id: the document must be the
+    same as the legacy OpcManager one (SIENTIAPDE-1974 NF2)."""
+    sink, dm = make_sink()
+    names = [{'collection': 'raw_x', 'tag_name': 'Counter'}]
+
+    await sink.publish_reading(
+        tag='ns=2;i=2', value=42, timestamp='2026-07-13 00:00:00+0000', names=names
+    )
+
+    collection, doc = dm.publish.await_args_list[0].args
+    assert collection == 'raw_x'
+    assert doc == {
+        'tag': 'ns=2;i=2',
+        'name': 'Counter',
+        'timestamp': '2026-07-13 00:00:00+0000',
+        'value': 42,
+    }
+    _, kwargs = sink.emit_metric.await_args
+    assert kwargs['tags']['model_id'] == ''

tests/unit/io/test_slots.py

@@ -62,3 +62,102 @@ def test_parse_slot_rejects_tag_without_names():
 
     with pytest.raises((KeyError, ValueError)):
         parse_slot(raw)
+
+
+RAW_V1 = {
+    'kepserver': {
+        'name': 'kepserver',
+        'server_id': '1',
+        'url': 'opc.tcp://kepserver:49320',
+        'server_uri': 'urn:sientia:opc-ingestor',
+        'cert_path': '/app/certs/opc-ingestor-cert.pem',
+        'private_key_path': '/app/certs/opc-ingestor-key.pem',
+        'server_cert_path': None,
+        'subscription_period_ms': 15000.0,
+        'tags': {
+            'ns=2;s=A.TargetSin': {
+                'aggr_func': 'lts',
+                'frequency': 30000,
+                'server_id': '1',
+                'tag_address': 'ns=2;s=A.TargetSin',
+                'tag_name': 'TargetSin',
+                'topics': ['raw_scouter-sin-approx', 'raw_other-model'],
+            }
+        },
+    }
+}
+
+
+def test_parse_slot_converts_v1_topics_to_names():
+    """Slot v1 (current Orchestrator) is read as v2: one name per topic, no model_id,
+    so the documents stay identical to the legacy ones (SIENTIAPDE-1974 G1)."""
+    server = parse_slot(RAW_V1)['kepserver']
+    tag = server['tags']['ns=2;s=A.TargetSin']
+
+    assert tag['names'] == [
+        {'collection': 'raw_scouter-sin-approx', 'tag_name': 'TargetSin'},
+        {'collection': 'raw_other-model', 'tag_name': 'TargetSin'},
+    ]
+    assert tag['frequency'] == 30000
+    assert tag['active_read'] == {'enabled': False}
+
+
+def test_parse_slot_v1_cert_path_means_security_but_path_is_dropped():
+    server = parse_slot(RAW_V1)['kepserver']
+
+    assert server['use_security'] is True
+    assert server['name'] == 'kepserver'
+    assert server['server_id'] == '1'
+    assert server['subscription_period_ms'] == 15000.0
+    # the certificate always comes from MinIO now: no file path survives the parse
+    assert 'cert_path' not in server
+    assert 'private_key_path' not in server
+
+
+def test_parse_slot_v1_without_cert_path_is_insecure():
+    raw = {
+        's': {
+            'server_id': '2',
+            'url': 'u',
+            'cert_path': None,
+            'tags': {'ns=2;i=1': {'tag_name': 't', 'frequency': '1000', 'topics': ['raw_c']}},
+        }
+    }
+
+    server = parse_slot(raw)['s']
+
+    assert server['use_security'] is False
+    assert server['name'] == 's'  # falls back to the slot key
+
+
+def test_parse_slot_v2_explicit_use_security_wins():
+    raw = {
+        's': {
+            'server_id': '1',
+            'url': 'u',
+            'use_security': False,
+            'cert_path': '/legacy/path',
+            'tags': {
+                'ns=2;i=1': {
+                    'frequency': '1000',
+                    'names': [{'collection': 'c', 'tag_name': 't', 'model_id': '1'}],
+                }
+            },
+        }
+    }
+
+    assert parse_slot(raw)['s']['use_security'] is False
+
+
+def test_parse_slot_rejects_v1_tag_without_tag_name():
+    raw = {
+        's': {
+            'server_id': '1',
+            'url': 'u',
+            'tags': {'ns=2;i=1': {'frequency': '1', 'topics': ['c']}},
+        }
+    }
+    import pytest
+
+    with pytest.raises(ValueError):
+        parse_slot(raw)

tests/unit/managers/test_ingestor_manager.py

@@ -4,6 +4,9 @@ from unittest.mock import AsyncMock, MagicMock, call, patch
 from pytest import fixture, mark
 from sientia_do.notifications.models import NotificationLevel
 
+from ingestor.config import GatewayConfig
+from ingestor.connection.pool import OpcConnectionPool
+from ingestor.io.slots import parse_slot
 from ingestor.managers.ingestor_manager import IngestorManager
 
 metadata = {
@@ -16,6 +19,21 @@ metadata = {
 }
 
 
+def raw_slot_v1(tag_name='Counter', address='ns=2;i=2'):
+    """A slot as the current (v1) Orchestrator writes it to Redis."""
+    return {
+        'server1': {
+            'name': 'server1',
+            'server_id': '1',
+            'url': 'opc.tcp://localhost:4840',
+            'server_uri': 'urn:sientia:opc-ingestor',
+            'cert_path': '/app/certs/opc-ingestor-cert.pem',
+            'subscription_period_ms': 500.0,
+            'tags': {address: {'tag_name': tag_name, 'frequency': '1000', 'topics': ['raw_a']}},
+        }
+    }
+
+
 @fixture(autouse=True)
 def _collect_garbage_after_test():
     # ponytail: IngestorManager.__del__ calls asyncio.run(self.shutdown()); if a
@@ -52,12 +70,12 @@ def ingestor_manager(data_manager_mock, resource_manager_mock):
     return ingestor
 
 
-@patch('ingestor.managers.ingestor_manager.OpcManager')
+@patch('ingestor.managers.ingestor_manager.ServerReader')
 @patch('ingestor.managers.ingestor_manager.DataManager')
 @patch('ingestor.managers.ingestor_manager.ResourceManager')
 @patch('ingestor.managers.ingestor_manager.NotificationHandler')
 def test___init__(
-    notification_handler_mock, resource_manager_mock, data_manager_mock, opc_manager_mock
+    notification_handler_mock, resource_manager_mock, data_manager_mock, server_reader_mock
 ):
     ingestor = IngestorManager(
         redis_data={'host': 'localhost', 'port': 6379},
@@ -72,7 +90,7 @@ def test___init__(
         metrics_controller=MagicMock(),
     )
 
-    opc_manager_mock.assert_not_called()
+    server_reader_mock.assert_not_called()
     data_manager_mock.assert_called_once_with(
         mongo_connection_string='mongodb://localhost:27017',
         mongo_database='sientia',
@@ -100,59 +118,86 @@ def test___init__(
     assert ingestor.data_manager == data_manager_mock.return_value
     assert ingestor.resource_manager == resource_manager_mock.return_value
 
+    # Read path (SIENTIAPDE-1974): pool + read sink built here, nothing connected yet.
+    assert isinstance(ingestor.pool, OpcConnectionPool)
+    assert ingestor.pool.cert_provider is ingestor.cert_provider
+    assert ingestor.cert_provider.mongo is data_manager_mock.return_value.mongo_repository
+    assert ingestor.sink.data_manager is data_manager_mock.return_value
+    assert ingestor.sink.postgres is None
 
-@mark.asyncio
-@patch('ingestor.managers.ingestor_manager.OpcManager')
-async def test_initialize_opc_from_config(opc_manager, ingestor_manager):
-    server_config = {
-        'name': 'server1',
-        'url': 'opc.tcp://localhost:4840',
-        'subscription_period_ms': 1000,
-        'server_uri': 'http://opcua-server.simulator',
-        'cert_path': '/path/to/cert',
-        'private_key_path': '/path/to/private_key',
-        'server_cert_path': '/path/to/server_cert',
-        'pod_id': 'test_pod',
-    }
 
-    opc_manager.return_value = MagicMock(connect=AsyncMock())
-    result = await ingestor_manager.initialize_opc_from_config(server_config)
-
-    opc_manager.assert_called_once_with(
-        name=server_config['name'],
-        url=server_config['url'],
-        data_manager=ingestor_manager.data_manager,
-        logger=ingestor_manager.logger,
-        subscription_period_ms=server_config['subscription_period_ms'],
-        server_uri=server_config['server_uri'],
-        notification_handler=ingestor_manager.notification_handler,
-        cert_path=server_config['cert_path'],
-        private_key_path=server_config['private_key_path'],
-        server_cert_path=server_config['server_cert_path'],
+@patch('ingestor.managers.ingestor_manager.DataManager')
+@patch('ingestor.managers.ingestor_manager.ResourceManager')
+def test___init___cert_provider_uses_gateway_config(resource_manager_mock, data_manager_mock):
+    ingestor = IngestorManager(
+        redis_data={'host': 'localhost', 'port': 6379},
+        lease_ttl=60,
+        heartbeat_ttl=60,
+        poll_interval=5,
+        mongo_connection_string='mongodb://localhost:27017',
+        mongo_database='sientia',
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
         metadata=metadata['metadata'],
-        metrics_controller=ingestor_manager.metrics_controller,
+        metrics_controller=MagicMock(),
+        gateway_config=GatewayConfig(cert_bucket='my-bucket', cert_cache_dir='certs-cache'),
     )
 
-    assert result == opc_manager.return_value
-    result.connect.assert_called_once()
+    assert ingestor.cert_provider.bucket == 'my-bucket'
+    assert ingestor.cert_provider.cache_dir == 'certs-cache'
 
 
 @mark.asyncio
-@patch('ingestor.managers.ingestor_manager.OpcManager')
-@patch('ingestor.managers.ingestor_manager.traceback')
-async def test_initialize_opc_from_config_exception(traceback_mock, opc_manager, ingestor_manager):
+@patch('ingestor.managers.ingestor_manager.ServerReader')
+async def test_initialize_opc_from_config(server_reader, ingestor_manager):
     server_config = {
         'name': 'server1',
+        'server_id': '1',
         'url': 'opc.tcp://localhost:4840',
+        'server_uri': 'urn:sientia:opc-ingestor',
+        'use_security': True,
+        'subscription_period_ms': 1000,
+    }
+
+    server_reader.return_value = MagicMock(connect=AsyncMock())
+    result = await ingestor_manager.initialize_opc_from_config(server_config)
+
+    server_reader.assert_called_once_with(
+        name=server_config['name'],
+        server_config=server_config,
+        pool=ingestor_manager.pool,
+        sink=ingestor_manager.sink,
+        logger=ingestor_manager.logger,
+        notification_handler=ingestor_manager.notification_handler,
+        metrics_controller=ingestor_manager.metrics_controller,
+        metadata=metadata['metadata'],
+    )
+
+    assert result == server_reader.return_value
+    assert result.config == server_config
+    result.connect.assert_awaited_once()
+
+
+@mark.asyncio
+@patch('ingestor.managers.ingestor_manager.ServerReader')
+@patch('ingestor.managers.ingestor_manager.traceback')
+async def test_initialize_opc_from_config_exception(
+    traceback_mock, server_reader, ingestor_manager
+):
+    server_config = {
+        'name': 'server1',
+        'server_id': '1',
+        'url': 'opc.tcp://localhost:4840',
+        'server_uri': 'urn:sientia:opc-ingestor',
+        'use_security': True,
         'subscription_period_ms': 1000,
-        'server_uri': 'http://opcua-server.simulator',
-        'cert_path': '/path/to/cert',
-        'private_key_path': '/path/to/private_key',
-        'server_cert_path': '/path/to/server_cert',
     }
 
     ingestor_manager.logger.error = MagicMock()
-    opc_manager.side_effect = Exception('Initialization error')
+    # e.g. the server is unreachable or the certificate can't be fetched from MinIO
+    server_reader.return_value = MagicMock(
+        connect=AsyncMock(side_effect=Exception('Initialization error'))
+    )
 
     result = await ingestor_manager.initialize_opc_from_config(server_config)
 
@@ -174,11 +219,14 @@ async def test_shutdown(ingestor_manager):
     ingestor_manager.opc_managers = {'server1': AsyncMock(), 'server2': AsyncMock()}
 
     ingestor_manager.data_manager.shutdown = MagicMock()
+    ingestor_manager.pool = MagicMock(close_all=AsyncMock())
 
     await ingestor_manager.shutdown()
 
     ingestor_manager.opc_managers['server1'].shutdown.assert_called_once()
     ingestor_manager.opc_managers['server2'].shutdown.assert_called_once()
+    # The pool is owned by the read path: closed after the readers unsubscribe.
+    ingestor_manager.pool.close_all.assert_awaited_once()
 
     ingestor_manager.data_manager.shutdown.assert_called_once()
 
@@ -218,9 +266,8 @@ async def test_remove_server_not_found(ingestor_manager):
 
 
 @mark.asyncio
-@patch('ingestor.managers.ingestor_manager.OpcManager')
 @patch('ingestor.managers.ingestor_manager.metrics')
-async def test_update_opc_servers(metrics, opc_manager, ingestor_manager):
+async def test_update_opc_servers(metrics, ingestor_manager):
     manager1 = MagicMock(config={'config': 'config1'})
     manager2 = MagicMock(config={'config': 'config2'})
     manager3 = MagicMock(config={'config': 'config3'})
@@ -267,7 +314,9 @@ async def test_update_opc_servers(metrics, opc_manager, ingestor_manager):
     assert ingestor_manager.opc_managers['server2'].config == {'config': 'config2'}
     assert ingestor_manager.opc_managers['server3'].config == {'config': 'config3'}
 
-    ingestor_manager.remove_server.assert_any_call('server4')
+    # server4 left this pod's slots: unsubscribe only, the connection stays for writes
+    ingestor_manager.remove_server.assert_any_call('server4', evict=False)
+    # server5 failed to initialize: removed with the default (evict)
     ingestor_manager.remove_server.assert_any_call('server5')
     assert ingestor_manager.remove_server.call_count == 2
 
@@ -431,25 +480,47 @@ async def test_get_number_of_slots_empty(metrics, ingestor_manager):
 @mark.asyncio
 async def test_get_slot_leases_1_success(ingestor_manager):
     ingestor_manager.resource_manager.lease_tag = AsyncMock(return_value=True)
-    ingestor_manager.resource_manager.get_tag_slot = AsyncMock(return_value={'tags': ['tag1']})
+    ingestor_manager.resource_manager.get_tag_slot = AsyncMock(return_value=raw_slot_v1())
 
     ingestor_manager.number_of_slots = 1
     result = await ingestor_manager.get_slot_leases()
 
-    assert result == {'1': {'tags': ['tag1']}}
+    # Stored already normalized to slot v2 (v1 converted by parse_slot).
+    assert result == {'1': parse_slot(raw_slot_v1())}
+    assert ingestor_manager.managed_tags == result
 
 
 @mark.asyncio
 async def test_get_slot_leases_2_success(ingestor_manager):
     ingestor_manager.resource_manager.lease_tag = AsyncMock(side_effect=[True, True])
     ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
-        side_effect=[{'tags': ['tag1']}, {'tags': ['tag2']}]
+        side_effect=[raw_slot_v1('Counter'), raw_slot_v1('Ramp', 'ns=2;i=3')]
     )
 
     ingestor_manager.number_of_slots = 2
     result = await ingestor_manager.get_slot_leases(max_slots=2)
 
-    assert result == {'1': {'tags': ['tag1']}, '2': {'tags': ['tag2']}}
+    assert result == {
+        '1': parse_slot(raw_slot_v1('Counter')),
+        '2': parse_slot(raw_slot_v1('Ramp', 'ns=2;i=3')),
+    }
+
+
+@mark.asyncio
+async def test_get_slot_leases_skips_malformed_slot(ingestor_manager):
+    ingestor_manager.resource_manager.lease_tag = AsyncMock(return_value=True)
+    # tag without names/topics/frequency
+    ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
+        return_value={'server1': {'server_id': '1', 'url': 'u', 'tags': {'ns=2;i=2': {}}}}
+    )
+
+    ingestor_manager.number_of_slots = 1
+    result = await ingestor_manager.get_slot_leases()
+
+    assert result == {}
+    _, kwargs = ingestor_manager.send_notification.call_args
+    assert kwargs['notification_id'] == 'OPC_SLOT_CONFIG_ERROR_1'
+    assert kwargs['level'] == NotificationLevel.ERROR
 
 
 @mark.asyncio
@@ -497,20 +568,20 @@ async def test_unsubscribe_slot(ingestor_manager):
 @mark.asyncio
 async def test_update_slot_config(ingestor_manager):
     ingestor_manager.managed_tags = {
-        'slot1': {'config': 'old_config'},
-        'slot2': {'config': 'new_config'},
-        'slot3': {'config': 'old_config'},
+        'slot1': parse_slot(raw_slot_v1('Old')),
+        'slot2': parse_slot(raw_slot_v1('Same')),
+        'slot3': parse_slot(raw_slot_v1('Old')),
     }
 
     ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
-        side_effect=[{'config': 'updated_config'}, {'config': 'new_config'}, None]
+        side_effect=[raw_slot_v1('Updated'), raw_slot_v1('Same'), None]
     )
     ingestor_manager.resource_manager.renew_tag_lease = AsyncMock()
 
     await ingestor_manager.update_slot_config()
 
-    assert ingestor_manager.managed_tags['slot1'] == {'config': 'updated_config'}
-    assert ingestor_manager.managed_tags['slot2'] == {'config': 'new_config'}
+    assert ingestor_manager.managed_tags['slot1'] == parse_slot(raw_slot_v1('Updated'))
+    assert ingestor_manager.managed_tags['slot2'] == parse_slot(raw_slot_v1('Same'))
     assert 'slot3' not in ingestor_manager.managed_tags
 
     ingestor_manager.resource_manager.renew_tag_lease.assert_any_call('slot1')
@@ -652,3 +723,202 @@ async def test_subscribe_to_tags(ingestor_manager):
     assert ingestor_manager.manage_server.call_count == 3
 
     ingestor_manager.managed_tags['slot1'].pop.assert_called_once_with('server3', None)
+
+
+@mark.asyncio
+async def test_update_slot_config_keeps_last_good_config_when_malformed(ingestor_manager):
+    good = parse_slot(raw_slot_v1('Counter'))
+    ingestor_manager.managed_tags = {'slot1': good}
+    ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
+        return_value={'server1': {'server_id': '1', 'url': 'u', 'tags': {'ns=2;i=2': {}}}}
+    )
+    ingestor_manager.resource_manager.renew_tag_lease = AsyncMock()
+
+    await ingestor_manager.update_slot_config()
+
+    # Still reading with the last good config; the bad one is notified, not applied.
+    assert ingestor_manager.managed_tags == {'slot1': good}
+    _, kwargs = ingestor_manager.send_notification.call_args
+    assert kwargs['notification_id'] == 'OPC_SLOT_CONFIG_ERROR_slot1'
+
+
+@mark.asyncio
+async def test_remove_server_forwards_evict(ingestor_manager):
+    reader = AsyncMock()
+    ingestor_manager.opc_managers = {'server1': reader}
+    ingestor_manager.managed_tags = {'slot1': {'server1': {}}}
+
+    await ingestor_manager.remove_server('server1', evict=False)
+
+    reader.shutdown.assert_awaited_once_with(evict=False)
+    assert 'server1' not in ingestor_manager.opc_managers
+    assert ingestor_manager.managed_tags == {'slot1': {}}
+
+
+@patch('ingestor.managers.ingestor_manager.DataManager')
+@patch('ingestor.managers.ingestor_manager.ResourceManager')
+def test___init___ignores_malformed_write_env(
+    resource_manager_mock, data_manager_mock, monkeypatch
+):
+    """A malformed write/gRPC-only env var must not stop the read path (review #1)."""
+    monkeypatch.setenv('GRPC_PORT', '')
+    monkeypatch.setenv('WRITE_MIN_PAUSE_S', '0,5')
+    monkeypatch.setenv('CERT_BUCKET', 'streamlit-connectors')
+
+    ingestor = IngestorManager(
+        redis_data={'host': 'localhost', 'port': 6379},
+        lease_ttl=60,
+        heartbeat_ttl=60,
+        poll_interval=5,
+        mongo_connection_string='mongodb://localhost:27017',
+        mongo_database='sientia',
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metadata=metadata['metadata'],
+        metrics_controller=MagicMock(),
+    )
+
+    assert ingestor.cert_provider.bucket == 'streamlit-connectors'
+
+
+def test_parse_slot_config_warns_once_about_server_cert_path(ingestor_manager):
+    raw = raw_slot_v1()
+    raw['server1']['server_cert_path'] = '/app/certs/kepserver.der'
+
+    ingestor_manager.parse_slot_config('1', raw)
+    ingestor_manager.parse_slot_config('1', raw)  # every loop refreshes the slot
+
+    warnings = [
+        c for c in ingestor_manager.logger.warning.call_args_list if 'server_cert_path' in str(c)
+    ]
+    assert len(warnings) == 1
+
+
+# --- READ_ENGINE=legacy: plan B with the untouched rc3 OpcManager -------------------
+
+
+def make_manager(read_engine=None):
+    with (
+        patch('ingestor.managers.ingestor_manager.DataManager'),
+        patch('ingestor.managers.ingestor_manager.ResourceManager'),
+    ):
+        manager = IngestorManager(
+            redis_data={'host': 'localhost', 'port': 6379},
+            lease_ttl=60,
+            heartbeat_ttl=60,
+            poll_interval=5,
+            mongo_connection_string='mongodb://localhost:27017',
+            mongo_database='sientia',
+            logger=MagicMock(),
+            notification_handler=MagicMock(),
+            metadata=metadata['metadata'],
+            metrics_controller=MagicMock(),
+            read_engine=read_engine,
+        )
+    manager.send_notification = MagicMock()
+    manager.emit_metric = AsyncMock()
+    return manager
+
+
+def test_read_engine_defaults_to_v2(monkeypatch):
+    monkeypatch.delenv('READ_ENGINE', raising=False)
+
+    assert make_manager().read_engine == 'v2'
+
+
+def test_read_engine_from_env(monkeypatch):
+    monkeypatch.setenv('READ_ENGINE', 'Legacy ')
+
+    assert make_manager().read_engine == 'legacy'
+
+
+def test_invalid_read_engine_falls_back_to_v2(monkeypatch):
+    monkeypatch.setenv('READ_ENGINE', 'v3')
+
+    manager = make_manager()
+
+    assert manager.read_engine == 'v2'
+    assert any('Invalid READ_ENGINE' in str(c) for c in manager.logger.warning.call_args_list)
+
+
+@mark.asyncio
+@patch('ingestor.managers.ingestor_manager.ServerReader')
+@patch('ingestor.managers.ingestor_manager.OpcManager')
+async def test_legacy_initialize_builds_rc3_opc_manager(opc_manager, server_reader):
+    manager = make_manager('legacy')
+    server_config = {
+        'name': 'server1',
+        'server_id': '1',
+        'url': 'opc.tcp://localhost:4840',
+        'subscription_period_ms': 1000,
+        'server_uri': 'urn:sientia:opc-ingestor',
+        'cert_path': '/app/certs/opc-ingestor-cert.pem',
+        'private_key_path': '/app/certs/opc-ingestor-key.pem',
+        'server_cert_path': None,
+    }
+    opc_manager.return_value = MagicMock(connect=AsyncMock())
+
+    result = await manager.initialize_opc_from_config(server_config)
+
+    server_reader.assert_not_called()
+    opc_manager.assert_called_once_with(
+        name='server1',
+        url='opc.tcp://localhost:4840',
+        subscription_period_ms=1000,
+        data_manager=manager.data_manager,
+        logger=manager.logger,
+        server_uri='urn:sientia:opc-ingestor',
+        notification_handler=manager.notification_handler,
+        metadata=metadata['metadata'],
+        cert_path='/app/certs/opc-ingestor-cert.pem',
+        private_key_path='/app/certs/opc-ingestor-key.pem',
+        server_cert_path=None,
+        metrics_controller=manager.metrics_controller,
+    )
+    assert result is opc_manager.return_value
+    result.connect.assert_awaited_once()
+
+
+def test_legacy_keeps_raw_slot_v1():
+    manager = make_manager('legacy')
+    raw = raw_slot_v1()
+
+    # OpcManager reads tag_name/topics and the file cert paths straight from the slot
+    assert manager.parse_slot_config('1', raw) is raw
+
+
+def test_legacy_rejects_slot_v2():
+    manager = make_manager('legacy')
+    raw_v2 = {
+        'server1': {
+            'name': 'server1',
+            'server_id': '1',
+            'url': 'opc.tcp://localhost:4840',
+            'use_security': True,
+            'subscription_period_ms': 500,
+            'tags': {
+                'ns=2;i=2': {
+                    'frequency': '1000',
+                    'names': [{'collection': 'raw_a', 'tag_name': 'Counter', 'model_id': '1'}],
+                }
+            },
+        }
+    }
+
+    assert manager.parse_slot_config('1', raw_v2) is None
+    _, kwargs = manager.send_notification.call_args
+    assert kwargs['notification_id'] == 'OPC_SLOT_CONFIG_ERROR_1'
+    assert 'READ_ENGINE=legacy' in kwargs['message']
+
+
+@mark.asyncio
+async def test_legacy_remove_server_calls_plain_shutdown():
+    manager = make_manager('legacy')
+    opc_manager = MagicMock(shutdown=AsyncMock())
+    manager.opc_managers = {'server1': opc_manager}
+    manager.managed_tags = {'slot1': {'server1': {}}}
+
+    await manager.remove_server('server1', evict=False)
+
+    opc_manager.shutdown.assert_awaited_once_with()
+    assert manager.opc_managers == {}

tests/unit/managers/test_server_reader.py

@@ -0,0 +1,164 @@
+from unittest.mock import AsyncMock, MagicMock, call
+
+import pytest
+
+import ingestor.metrics as metrics
+from ingestor.managers.server_reader import ServerReader
+
+SERVER_CONFIG = {
+    'name': 'kepserver',
+    'server_id': 1,  # the slot may carry it as int; the pool keys by str
+    'url': 'opc.tcp://kepserver:49320',
+    'server_uri': 'urn:sientia:opc-ingestor',
+    'use_security': True,
+    'subscription_period_ms': 15000.0,
+}
+
+
+def make_reader():
+    pool = MagicMock(get=AsyncMock(), evict=AsyncMock())
+    reader = ServerReader(
+        name='kepserver',
+        server_config=SERVER_CONFIG,
+        pool=pool,
+        sink=AsyncMock(),
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=AsyncMock(),
+        metadata={},
+    )
+    reader.engine = MagicMock(
+        subscriptions={},
+        pod_id='pod',
+        create_subscription=AsyncMock(),
+        subscribe=AsyncMock(),
+        unsubscribe=AsyncMock(),
+        check_cycles=AsyncMock(),
+        check_opc_listenning=AsyncMock(return_value=False),
+        emit_metric=AsyncMock(),
+    )
+    return reader, pool
+
+
+def test_server_cfg_keeps_only_connection_fields():
+    reader, _ = make_reader()
+
+    assert reader.server_id == '1'
+    assert reader.server_cfg == {
+        'url': 'opc.tcp://kepserver:49320',
+        'server_uri': 'urn:sientia:opc-ingestor',
+        'use_security': True,
+        'subscription_period_ms': 15000.0,
+    }
+
+
+@pytest.mark.asyncio
+async def test_connect_opens_pooled_connection():
+    reader, pool = make_reader()
+
+    await reader.connect()
+
+    # use_security=True: the pool fetches the certificate from MinIO (CertProvider)
+    pool.get.assert_awaited_once_with('1', reader.server_cfg)
+
+
+@pytest.mark.asyncio
+async def test_connect_propagates_failure():
+    reader, pool = make_reader()
+    pool.get.side_effect = ConnectionError('unreachable')
+
+    with pytest.raises(ConnectionError):
+        await reader.connect()
+
+
+@pytest.mark.asyncio
+async def test_subscription_calls_delegate_to_engine_per_slot():
+    reader, _ = make_reader()
+    nodes = {'ns=2;i=2': {'frequency': '1000', 'names': [{'collection': 'raw_a', 'tag_name': 'a'}]}}
+
+    await reader.create_subscription('1')
+    await reader.subscribe('1', nodes, 10)
+    await reader.unsubscribe('1')
+
+    reader.engine.create_subscription.assert_awaited_once_with('1', reader.server_cfg, '1')
+    reader.engine.subscribe.assert_awaited_once_with('1', nodes, 10)
+    reader.engine.unsubscribe.assert_awaited_once_with('1')
+
+
+@pytest.mark.asyncio
+async def test_supervision_delegates_to_engine():
+    reader, _ = make_reader()
+    reader.engine.check_opc_listenning.return_value = True
+
+    await reader.check_cycles()
+    lost = await reader.check_opc_listenning()
+
+    reader.engine.check_cycles.assert_awaited_once()
+    reader.engine.check_opc_listenning.assert_awaited_once_with('1')
+    assert lost is True
+
+
+def test_subscriptions_are_the_engine_ones():
+    reader, _ = make_reader()
+    reader.engine.subscriptions['1'] = MagicMock()
+
+    assert reader.subscriptions is reader.engine.subscriptions
+
+
+@pytest.mark.asyncio
+async def test_shutdown_unsubscribes_all_slots_evicts_and_zeroes_metrics():
+    reader, pool = make_reader()
+    reader.engine.subscriptions.update({'1': MagicMock(), '2': MagicMock()})
+
+    await reader.shutdown()
+
+    reader.engine.unsubscribe.assert_has_awaits([call('1'), call('2')])
+    # Dropping the pooled connection makes the next initialization reconnect.
+    pool.evict.assert_awaited_once_with('1')
+    reader.engine.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_CONNECTION_STATUS,
+        method='set',
+        value=0,
+        tags={'pod_id': 'pod', 'server_name': '1', 'server_url': 'opc.tcp://kepserver:49320'},
+    )
+    reader.engine.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_TAGS_SUBSCRIBED,
+        method='set',
+        value=0,
+        tags={'pod_id': 'pod', 'server_name': '1'},
+    )
+
+
+@pytest.mark.asyncio
+async def test_shutdown_without_evict_keeps_the_shared_connection():
+    """Slot moved to another pod: drop the subscriptions, keep the pooled
+    connection for NATS writes/gRPC (review #7)."""
+    reader, pool = make_reader()
+    reader.engine.subscriptions.update({'1': MagicMock()})
+
+    await reader.shutdown(evict=False)
+
+    reader.engine.unsubscribe.assert_awaited_once_with('1')
+    pool.evict.assert_not_awaited()
+    for c in reader.engine.emit_metric.await_args_list:
+        assert c.kwargs['metric_object'] is not metrics.OPC_CONNECTION_STATUS
+
+
+@pytest.mark.asyncio
+async def test_shutdown_unsubscribes_every_slot_while_engine_pops_them():
+    """The real unsubscribe() pops from engine.subscriptions: shutdown must iterate a
+    snapshot (list()), or the loop breaks with "dictionary changed size during
+    iteration" and the remaining slots stay subscribed (SonarQube S7504 does not apply)."""
+    reader, pool = make_reader()
+    subs = {'1': MagicMock(), '2': MagicMock(), '3': MagicMock()}
+    reader.engine.subscriptions = subs
+
+    async def pop(slot):
+        subs.pop(slot)
+
+    reader.engine.unsubscribe = AsyncMock(side_effect=pop)
+
+    await reader.shutdown()
+
+    assert subs == {}
+    assert reader.engine.unsubscribe.await_count == 3

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,22 @@ 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,
+        pool=mock_ingestor_instance.ingestor_manager.pool,
+        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 +323,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 +347,35 @@ 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 failing (e.g. Postgres config) must not kill the
+    app: the read path lives in the ingestor_manager, so the main loop 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,87 @@
+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 == 'streamlit-connectors'
+    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
+
+
+def test_certs_from_env_ignores_malformed_numeric_vars(monkeypatch):
+    import pytest
+
+    monkeypatch.setenv('GRPC_PORT', '')
+    monkeypatch.setenv('IDLE_CONN_MAX_S', 'abc')
+    monkeypatch.setenv('CERT_BUCKET', 'my-bucket')
+    monkeypatch.setenv('CERT_CACHE_DIR', 'certs-cache')
+
+    cfg = GatewayConfig.certs_from_env()
+
+    assert cfg.cert_bucket == 'my-bucket'
+    assert cfg.cert_cache_dir == 'certs-cache'
+    with pytest.raises(ValueError):
+        GatewayConfig.from_env()  # the full config still fails, only the runtime uses it

tests/unit/test_gateway_runtime.py

@@ -0,0 +1,175 @@
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from ingestor import gateway_runtime as gr
+from ingestor.config import GatewayConfig
+
+_PATCHED_CLASSES = (
+    '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,
+    }
+    # The pool is owned by IngestorManager (read path) and only borrowed here.
+    patched['pool'] = MagicMock()
+    patched['pool'].close_all = AsyncMock()
+    patched['pool'].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(pool=None, **config_overrides):
+    if pool is None:
+        pool = MagicMock(close_all=AsyncMock(), release_idle=AsyncMock())
+    return gr.GatewayRuntime(
+        config=GatewayConfig(**config_overrides),
+        data_manager=MagicMock(),
+        pool=pool,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(),
+    )
+
+
+def test_builds_object_graph_with_config_flags(mocks):
+    runtime = make_runtime(
+        pool=mocks['pool'],
+        write_timestamp=False,
+        write_quality=False,
+        write_min_pause_s=0.25,
+        grpc_enabled=False,
+    )
+
+    # Writes and active reads share the pool built by the read path.
+    assert runtime.pool is mocks['pool']
+    mocks['DataSink'].assert_called_once()
+
+    mocks['WriteEngine'].assert_called_once()
+    args, kwargs = mocks['WriteEngine'].call_args
+    assert args[0] is mocks['pool']
+    assert kwargs['write_timestamp'] is False
+    assert kwargs['write_quality'] is False
+    assert kwargs['min_pause_s'] == 0.25
+
+    args, _ = mocks['ActiveReadEngine'].call_args
+    assert args[0] is mocks['pool']
+
+    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_but_not_the_pool(mocks):
+    runtime = make_runtime(pool=mocks['pool'], 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()
+    # IngestorManager.shutdown() closes it, after its readers unsubscribe.
+    mocks['pool'].close_all.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_stop_skips_nats_stop_if_never_started(mocks):
+    runtime = make_runtime(pool=mocks['pool'])
+    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['pool'].close_all.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_idle_sweep_releases_idle_connections(mocks):
+    runtime = make_runtime(pool=mocks['pool'], idle_conn_max_s=42.0)
+    runtime._nats_started = True  # isolate from the retry path
+
+    await runtime.idle_sweep()
+
+    mocks['pool'].release_idle.assert_awaited_once_with(42.0)