chore: bump version to 2.2.0-rc1

SIENTIAPDE-1974: fix gateway main loop crash (KeyError) on slot unsubscribe

.env.example

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

.github/workflows/deploy.yml

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

.github/workflows/release.yml

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

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

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

.gitignore

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

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
 ```

certs/opc-ingestor-cert.pem

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

certs/opc-ingestor-key.pem

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

ingestor/adapters/grpc_server.py

@@ -0,0 +1,70 @@
+"""gRPC input adapter for on-demand tag reads (contract C4).
+
+Thin adapter: unpack the request, call ActiveReadEngine.read_now, map the
+result onto ReadTagResponse. Internal-only for now -- no auth/TLS, per the
+feature owner (no consumer exists yet, likely a frontend later).
+"""
+
+import grpc
+from sientia_do.observability.logger import Logger
+
+from ingestor.engines.active_read import ActiveReadEngine
+from ingestor.proto import opc_gateway_pb2, opc_gateway_pb2_grpc
+
+
+class ReadTagServicer(opc_gateway_pb2_grpc.OpcGatewayServicer):
+    def __init__(self, active_read: ActiveReadEngine, logger: Logger, slot_lookup=None):
+        self.active_read = active_read
+        self.logger = logger
+        # ponytail: slot_lookup wiring (address -> names from the current slot)
+        # lands at the SIENTIAPDE-1974 integration -- `names` only exist in slot v2,
+        # which the Orchestrator produces after 1974. Until then ReadTag never
+        # persists (persisted=False); optional callable keeps this servicer usable now.
+        self.slot_lookup = slot_lookup
+
+    async def ReadTag(self, request, context):
+        names = None
+        if self.slot_lookup is not None:
+            names = self.slot_lookup(request.server_id, request.addr)
+        try:
+            result = await self.active_read.read_now(request.server_id, request.addr, names=names)
+        except Exception as e:
+            self.logger.error(f'ReadTag failed for {request.server_id}:{request.addr}: {e}')
+            await context.abort(grpc.StatusCode.INTERNAL, f'read failed: {e}')
+            return
+        return opc_gateway_pb2.ReadTagResponse(
+            value=str(result['value']),
+            opc_status_code=result['opc_status_code'] or '',
+            source_timestamp=result['source_timestamp'] or '',
+            server_timestamp=result['server_timestamp'] or '',
+            persisted=result['persisted'],
+        )
+
+
+class GrpcServer:
+    def __init__(
+        self,
+        active_read: ActiveReadEngine,
+        logger: Logger,
+        port: int = 50051,
+        slot_lookup=None,
+    ):
+        self.active_read = active_read
+        self.logger = logger
+        self.port = port
+        self.slot_lookup = slot_lookup
+        self.server: grpc.aio.Server | None = None
+
+    async def start(self) -> None:
+        self.server = grpc.aio.server()
+        opc_gateway_pb2_grpc.add_OpcGatewayServicer_to_server(
+            ReadTagServicer(self.active_read, self.logger, self.slot_lookup), self.server
+        )
+        self.server.add_insecure_port(f'[::]:{self.port}')
+        await self.server.start()
+        self.logger.info(f'GrpcServer started on port {self.port}')
+
+    async def stop(self) -> None:
+        if self.server is None:
+            return
+        await self.server.stop(grace=5)

ingestor/adapters/nats_consumer.py

@@ -0,0 +1,55 @@
+"""NATS input adapter for OPC write commands (contract C2).
+
+Thin adapter: decode the message JSON and hand the dict to WriteEngine,
+which is already raise-safe against malformed commands. No metrics of its
+own — the engine records write metrics; a malformed-JSON drop is logged
+here. No reconnect logic — nats-py handles reconnection internally.
+"""
+
+import json
+
+import nats
+from nats.aio.client import Client
+from nats.aio.subscription import Subscription
+from sientia_do.observability.logger import Logger
+
+from ingestor.engines.write import WriteEngine
+
+
+class NatsWriteConsumer:
+    def __init__(
+        self,
+        nats_url: str,
+        write_engine: WriteEngine,
+        logger: Logger,
+        subject: str = 'opc.write',
+        queue: str = 'gateway-writers',
+    ):
+        self.nats_url = nats_url
+        self.write_engine = write_engine
+        self.logger = logger
+        self.subject = subject
+        self.queue = queue
+        self.nc: Client | None = None
+        self.sub: Subscription | None = None
+
+    async def start(self) -> None:
+        self.nc = await nats.connect(servers=self.nats_url)
+        self.sub = await self.nc.subscribe(self.subject, queue=self.queue, cb=self._handle)
+        self.logger.info(f'NatsWriteConsumer started: subject={self.subject} queue={self.queue}')
+
+    async def _handle(self, msg) -> None:
+        try:
+            payload = json.loads(msg.data)
+        except ValueError as e:  # includes JSONDecodeError and UnicodeDecodeError
+            preview = msg.data[:200]
+            self.logger.error(f'NatsWriteConsumer dropped malformed JSON ({e}): {preview!r}')
+            return
+        await self.write_engine.write(payload)
+
+    async def stop(self) -> None:
+        if self.nc is None:
+            return
+        if self.sub is not None:
+            await self.sub.unsubscribe()
+        await self.nc.drain()

ingestor/app.py

@@ -7,7 +7,16 @@ from time import time
 
 from prometheus_client import start_http_server
 
+try:  # optional: local dev reads .env; in-cluster env comes from Helm (no .env file)
+    from dotenv import load_dotenv
+
+    load_dotenv()
+except ImportError:
+    pass
+
 import ingestor.metrics as metrics
+from ingestor.config import GatewayConfig
+from ingestor.gateway_runtime import GatewayRuntime
 from ingestor.ingestor import Ingestor
 
 exit_signal = Event()
@@ -48,10 +57,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

@@ -0,0 +1,114 @@
+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.
+
+    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
+        self.minio = minio
+        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(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 = 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/clock.py

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

ingestor/connection/pool.py

@@ -0,0 +1,198 @@
+import asyncio
+
+from asyncua import Client
+from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+from sientia_do.observability.logger import Logger
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+
+import ingestor.metrics as metrics
+from ingestor.connection.certs import CertProvider
+from ingestor.connection.clock import MonotonicClock
+from ingestor.connection.security import apply_security
+
+
+class OpcConnectionPool(SientiaMonitoring):
+    """Per-server OPC UA client cache (F0 capstone): reuses live connections,
+    evicts idle ones, and reconnects after a failed attempt (never caches it)."""
+
+    def __init__(
+        self,
+        cert_provider: CertProvider,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+    ):
+        self.cert_provider = cert_provider
+        self.clock = MonotonicClock()
+        self._connections: dict[str, tuple[Client, float]] = {}
+        # 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] = {}
+
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+    async def get(self, server_id: str, server_cfg: dict) -> Client:
+        """Return the cached client for `server_id`, connecting one if needed.
+
+        Double-checked under a per-server lock so concurrent callers for the
+        same new server share one connection (no leaked, untracked sessions).
+        `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:
+            return None
+        client, _ = entry
+        self._connections[server_id] = (client, self.clock.elapsed())
+        return client
+
+    async def _open(self, server_id: str, server_cfg: dict) -> Client:
+        try:
+            client = await self._connect(server_cfg)
+        except Exception:
+            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},
+            )
+            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()
+            await apply_security(client, server_cfg.get('server_uri'), cert, key, None)
+        await client.connect()
+        return client
+
+    async def _record_connected(self, server_id: str, url: str):
+        await self.emit_metric(
+            metric_object=metrics.OPC_CONNECTIONS_TOTAL,
+            tags={'pod_id': self.pod_id, 'server_name': server_id},
+        )
+        await self.emit_metric(
+            metric_object=metrics.OPC_CONNECTION_STATUS,
+            method='set',
+            value=1,
+            tags={'pod_id': self.pod_id, 'server_name': server_id, 'server_url': url},
+        )
+
+    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`,
+        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 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,
+        # so iterating the dict directly would raise "changed size during iteration".
+        for server_id in list(self._connections):  # NOSONAR
+            await self._evict(server_id)
+
+    async def _evict(self, server_id: str):
+        client, _ = self._connections.pop(server_id)
+        self._configs.pop(server_id, None)
+        await self._disconnect_with_retries(client, server_id)
+        await self.emit_metric(
+            metric_object=metrics.OPC_DISCONNECTIONS_TOTAL,
+            tags={'pod_id': self.pod_id, 'server_name': server_id},
+        )
+
+    async def _disconnect_with_retries(self, client: Client, server_id: str) -> None:
+        """Tries 5 times to disconnect, with a delay of 100ms x try (mirrors
+        the legacy OpcManager.disconnection_fallback)."""
+        for i in range(5):
+            try:
+                await client.disconnect()
+                return
+            except Exception as e:
+                self.logger.error(
+                    f'Failed to disconnect from {server_id}, attempt {i + 1} of 5: {e}'
+                )
+                await asyncio.sleep(0.1 * i)

ingestor/connection/security.py

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

ingestor/converters.py

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

ingestor/engines/active_read.py

@@ -0,0 +1,132 @@
+"""Active OPC reads (F3): on-demand (`read_now`, used by the gRPC server -
+Task 16) and scheduled (`scheduled_tick`, driven by the app main loop each
+collect cycle). Emits metrics only -- unlike WriteEngine there is no
+notification path here."""
+
+from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+from sientia_do.observability.logger import Logger
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, OPC_TIMEZONE
+
+import ingestor.metrics as metrics
+from ingestor.connection.clock import MonotonicClock
+from ingestor.connection.pool import OpcConnectionPool
+from ingestor.io.data_sink import DataSink
+
+
+class ActiveReadEngine(SientiaMonitoring):
+    """On-demand + scheduled OPC reads, sharing the pool/sink with WriteEngine
+    and SubscriptionEngine."""
+
+    def __init__(
+        self,
+        pool: OpcConnectionPool,
+        sink: DataSink,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+    ):
+        self.pool = pool
+        self.sink = sink
+        self.clock = MonotonicClock()
+        self._ticks: dict[tuple[str, str], int] = {}
+
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+    async def read_now(
+        self,
+        server_id: str,
+        addr: str,
+        names: list[dict] | None = None,
+        server_cfg: dict | None = None,
+    ) -> dict:
+        """Read `addr` on `server_id` right now; persist via sink only if `names` given."""
+        tags = {'pod_id': self.pod_id, 'server_name': server_id}
+        start = self.clock.elapsed()
+        try:
+            client = await self.pool.get(server_id, server_cfg or {})
+            dv = await client.get_node(addr).read_data_value()
+        except Exception as e:
+            await self.emit_metric(
+                metric_object=metrics.OPC_ACTIVE_READ_TOTAL, tags={**tags, 'result': 'error'}
+            )
+            self.logger.error(f'Active read failed for {server_id}:{addr}: {e}')
+            raise
+
+        await self.emit_metric(
+            metric_object=metrics.OPC_ACTIVE_READ_LATENCY,
+            method='observe',
+            value=self.clock.elapsed() - start,
+            tags=tags,
+        )
+        await self.emit_metric(
+            metric_object=metrics.OPC_ACTIVE_READ_TOTAL, tags={**tags, 'result': 'success'}
+        )
+        return await self._build_result(dv, addr, names, tags)
+
+    async def _build_result(self, dv, addr: str, names: list[dict] | None, tags: dict) -> dict:
+        value, status_name, source_ts, server_ts = self._extract(dv)
+        await self._record_read_lag(source_ts, tags)
+        source_str = self._format_ts(source_ts)
+        persisted = bool(names)
+        if names:
+            await self.sink.publish_reading(
+                tag=addr, value=value, timestamp=source_str, names=names
+            )
+        return {
+            'value': value,
+            'opc_status_code': status_name,
+            'source_timestamp': source_str,
+            'server_timestamp': self._format_ts(server_ts),
+            'persisted': persisted,
+        }
+
+    @staticmethod
+    def _extract(dv):
+        status_name = dv.StatusCode.name if dv.StatusCode is not None else None
+        return dv.Value.Value, status_name, dv.SourceTimestamp, dv.ServerTimestamp
+
+    @staticmethod
+    def _format_ts(ts) -> str | None:
+        if ts is None:
+            return None
+        return ts.replace(tzinfo=OPC_TIMEZONE).strftime(DATETIME_FORMAT_WITH_TZ)
+
+    async def _record_read_lag(self, source_ts, tags: dict) -> None:
+        if source_ts is None:
+            return
+        lag = MonotonicClock.now_wall() - source_ts.replace(tzinfo=OPC_TIMEZONE).timestamp()
+        await self.emit_metric(
+            metric_object=metrics.OPC_READ_LAG, method='set', value=lag, tags=tags
+        )
+
+    async def scheduled_tick(self, slot: dict) -> None:
+        """Sweep every server/tag in a parsed slot, reading tags due for an active read."""
+        for server in slot.values():
+            server_id = server['server_id']
+            server_cfg = {k: v for k, v in server.items() if k != 'tags'}
+            for addr, tag in server['tags'].items():
+                await self._tick_tag(server_id, addr, tag, server_cfg)
+
+    async def _tick_tag(self, server_id: str, addr: str, tag: dict, server_cfg: dict) -> None:
+        active_read = tag.get('active_read') or {}
+        if not active_read.get('enabled'):
+            return
+        if not self._should_read(server_id, addr, active_read.get('cycle_multiple', 1)):
+            return
+        try:
+            await self.read_now(server_id, addr, names=tag['names'], server_cfg=server_cfg)
+        except Exception as e:
+            self.logger.error(f'Scheduled active read failed for {server_id}:{addr}: {e}')
+
+    def _should_read(self, server_id: str, addr: str, cycle_multiple: int) -> bool:
+        key = (server_id, addr)
+        count = self._ticks.get(key, 0) + 1
+        self._ticks[key] = count
+        return count % max(cycle_multiple, 1) == 0

ingestor/engines/subscription.py

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

ingestor/engines/write.py

@@ -0,0 +1,195 @@
+"""OPC write execution (F2 capstone) from a NATS C2 command dict.
+
+Fixes 1879 correctness gaps: safe value coercion (never raises on bad
+input), StatusCode/SourceTimestamp fidelity on the DataValue actually sent
+to the server, a rate-limit floor between writes, and OPC error typing.
+"""
+
+import asyncio
+from dataclasses import dataclass
+
+from asyncua import ua
+from asyncua.ua.uaerrors import UaStatusCodeError
+from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+from sientia_do.observability.logger import Logger
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+
+import ingestor.metrics as metrics
+from ingestor.connection.clock import MonotonicClock
+from ingestor.connection.pool import OpcConnectionPool
+from ingestor.converters import ConversionError, to_typed_value
+from ingestor.io.data_sink import DataSink
+
+_VARIANT_TYPES = {
+    'float': ua.VariantType.Float,
+    'double': ua.VariantType.Double,
+    'real': ua.VariantType.Double,
+    'int': ua.VariantType.Int32,
+    'integer': ua.VariantType.Int32,
+    'bool': ua.VariantType.Boolean,
+    'boolean': ua.VariantType.Boolean,
+    'str': ua.VariantType.String,
+    'string': ua.VariantType.String,
+}
+
+# ponytail: same reconnect-worthy Bad*-status subset as laborious's
+# opc_repository.py (different repo, not importable here); keep in sync by
+# hand or promote to a shared lib if it drifts.
+_SESSION_BAD_NAMES = frozenset(
+    {
+        'BadSessionIdInvalid',
+        'BadSessionClosed',
+        'BadSessionNotActivated',
+        'BadSecureChannelIdInvalid',
+        'BadSecureChannelClosed',
+        'BadSecureChannelTokenUnknown',
+        'BadTcpSecureChannelUnknown',
+        'BadServerNotConnected',
+        'BadConnectionClosed',
+        'BadConnectionRejected',
+        'BadNotConnected',
+        'BadTimeout',
+        'BadRequestTimeout',
+        'BadTooManySessions',
+    }
+)
+
+
+@dataclass
+class WriteResult:
+    ok: bool
+    status: str  # 'ok' | 'error' | 'session_bad'
+    opc_status_code: str
+    message: str
+
+
+class WriteEngine(SientiaMonitoring):
+    """Executes one OPC write from a NATS C2 command."""
+
+    def __init__(
+        self,
+        pool: OpcConnectionPool,
+        sink: DataSink,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+        write_timestamp: bool = True,
+        write_quality: bool = True,
+        min_pause_s: float = 0.0,
+    ):
+        self.pool = pool
+        self.sink = sink
+        self.write_timestamp = write_timestamp
+        self.write_quality = write_quality
+        self.min_pause_s = min_pause_s
+        self.clock = MonotonicClock()
+        self._last_write_at: float | None = None
+
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+    async def write(self, cmd: dict) -> WriteResult:
+        await self._rate_limit()
+
+        missing = [k for k in ('value', 'data_type', 'server_id', 'addr') if k not in cmd]
+        if missing:
+            return await self._emit_error(cmd, 'bad_command', f'missing fields: {missing}', 'error')
+
+        try:
+            typed = to_typed_value(cmd['value'], cmd['data_type'])
+        except ConversionError as e:
+            return await self._emit_error(cmd, 'conversion', str(e), 'error')
+
+        dv = self._build_datavalue(cmd, typed)
+
+        start = self.clock.elapsed()
+        try:
+            client = await self.pool.get(cmd['server_id'], self._server_cfg(cmd))
+            node = client.get_node(cmd['addr'])
+            await node.write_value(dv)
+        except Exception as e:
+            status = 'session_bad' if self._is_session_bad(e) else 'error'
+            return await self._emit_error(cmd, type(e).__name__, str(e), status)
+
+        return await self._emit_success(cmd, start)
+
+    async def _rate_limit(self) -> None:
+        now = self.clock.elapsed()
+        if self.min_pause_s > 0 and self._last_write_at is not None:
+            remaining = self.min_pause_s - (now - self._last_write_at)
+            if remaining > 0:
+                await asyncio.sleep(remaining)
+                now = self.clock.elapsed()
+        self._last_write_at = now
+
+    def _server_cfg(self, cmd: dict) -> dict:
+        # ponytail: minimal cfg built straight from the C2 message. Resolving
+        # the full connector config (url/security) from the connectors DB by
+        # server_id is F4 wiring, not this task — pool is mocked in tests.
+        return {
+            'url': cmd.get('url'),
+            'use_security': cmd.get('use_security', False),
+            'server_uri': cmd.get('server_uri'),
+        }
+
+    def _variant_type(self, data_type: str) -> ua.VariantType:
+        # direct index: unknown dtypes already became ConversionError in to_typed_value
+        return _VARIANT_TYPES[data_type.strip().lower()]
+
+    def _build_datavalue(self, cmd: dict, typed_value) -> ua.DataValue:
+        variant = ua.Variant(typed_value, self._variant_type(cmd['data_type']))
+        write_quality = cmd.get('write_quality', self.write_quality)
+        write_timestamp = cmd.get('write_timestamp', self.write_timestamp)
+        # NOTE: DataValue.StatusCode default_factory populates a Good status
+        # even when omitted, so "off" must pass StatusCode=None explicitly
+        # or the flag silently no-ops.
+        status_code = ua.StatusCode(ua.UInt32(ua.StatusCodes.Good)) if write_quality else None
+        source_ts = ua.DateTime.fromisoformat(cmd['timestamp']) if write_timestamp else None
+        return ua.DataValue(variant, StatusCode=status_code, SourceTimestamp=source_ts)
+
+    def _is_session_bad(self, exc: Exception) -> bool:
+        return isinstance(exc, UaStatusCodeError) and type(exc).__name__ in _SESSION_BAD_NAMES
+
+    async def _emit_error(self, cmd: dict, error_type: str, message: str, status: str):
+        tags = {'pod_id': self.pod_id, 'server_name': cmd.get('server_id', 'unknown')}
+        await self.emit_metric(
+            metric_object=metrics.OPC_ERROR_BY_TYPE, tags={**tags, 'error_type': error_type}
+        )
+        await self.emit_metric(
+            metric_object=metrics.OPC_WRITE_TOTAL,
+            tags={**tags, 'model_id': cmd.get('model_id', 'unknown'), 'result': 'error'},
+        )
+        result = WriteResult(ok=False, status=status, opc_status_code=error_type, message=message)
+        await self.sink.update_write_status(cmd, result)
+        return result
+
+    async def _emit_success(self, cmd: dict, start: float):
+        tags = {'pod_id': self.pod_id, 'server_name': cmd.get('server_id', 'unknown')}
+        await self.emit_metric(
+            metric_object=metrics.OPC_WRITE_TOTAL,
+            tags={**tags, 'model_id': cmd.get('model_id', 'unknown'), 'result': 'success'},
+        )
+        await self.emit_metric(
+            metric_object=metrics.OPC_WRITE_LATENCY,
+            method='observe',
+            value=self.clock.elapsed() - start,
+            tags=tags,
+        )
+        await self.emit_metric(
+            metric_object=metrics.OPC_WRITE_STATUS_BY_MODEL,
+            method='set',
+            value=1,
+            tags={
+                'pod_id': self.pod_id,
+                'model_id': cmd.get('model_id', 'unknown'),
+                'status': 'ok',
+            },
+        )
+        result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='')
+        await self.sink.update_write_status(cmd, result)
+        return result

ingestor/gateway_runtime.py

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

@@ -268,12 +268,14 @@ class Ingestor(SientiaMonitoring):
 
             overleases = list(self.ingestor_manager.managed_tags.keys())[1:]
 
-            await self.ingestor_manager.drop_slot_leases(overleases)
-
+            # Unsubscribe before giving the leases back: another pod may take a released
+            # slot right away, and must not find this pod still reading it
             for lease in overleases:
                 await self.ingestor_manager.unsubscribe_slot(lease)
                 self.ingestor_manager.managed_tags.pop(lease)
 
+            await self.ingestor_manager.drop_slot_leases(overleases)
+
             await self.emit_metric(
                 metric_object=metrics.SLOTS_MANAGED,
                 method='set',

ingestor/io/data_sink.py

@@ -0,0 +1,75 @@
+import re
+
+from sqlalchemy import text
+
+import ingestor.metrics as metrics
+
+# fullmatch (no ^/$ anchors) so a trailing newline can't slip through — `$` would
+# match before a trailing '\n', which is not what we want for a SQL identifier.
+_IDENTIFIER_RE = re.compile(r'\w+', re.ASCII)
+
+
+class DataSink:
+    """Gateway egress wrapper: fans a reading out to every configured name/collection."""
+
+    def __init__(self, data_manager, postgres) -> None:
+        self.data_manager = data_manager
+        self.postgres = postgres
+
+    async def emit_metric(self, **kwargs) -> None:
+        # ponytail: delegate to data_manager, it's already a SientiaMonitoring
+        await self.data_manager.emit_metric(**kwargs)
+
+    async def publish_reading(self, tag: str, value, timestamp, names: list[dict]) -> None:
+        """Publish one MongoDB document per name entry and record its tag value metric.
+
+        `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']}
+            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,
+                method='set',
+                value=value,
+                tags={
+                    'pod_id': self.data_manager.pod_id,
+                    'tag_name': name['tag_name'],
+                    'model_id': name.get('model_id', ''),
+                },
+            )
+
+    async def update_write_status(self, cmd: dict, result) -> None:  # NOSONAR
+        """UPDATE the per-model prediction row's write_status (C3), by timestamp+model_id.
+
+        schema/table_name are per-model SQL identifiers, not values — they can't be
+        bound params, so they're checked against an allowlist before interpolation.
+        status/timestamp/model_id ARE bound params, never f-stringed.
+        """
+        schema = self._validate_identifier(cmd['schema'])
+        table = self._validate_identifier(cmd['table_name'])
+        # schema/table are allowlist-validated identifiers above; status/timestamp/model_id
+        # are bound params below, never interpolated — not a real injection vector.
+        stmt = text(
+            f'UPDATE {schema}.{table} SET write_status = :status '  # noqa: S608 # nosec B608
+            'WHERE timestamp = :timestamp AND model_id = :model_id'
+        )
+        params = {
+            'status': result.status,
+            'timestamp': cmd['timestamp'],
+            'model_id': cmd['model_id'],
+        }
+        # ponytail: sync engine call inside async def — same pattern sientia_do's own
+        # Postgres activities use (e.g. export_data_to_postgres); no async pg driver here.
+        with self.postgres.engine.begin() as connection:
+            connection.execute(stmt, params)
+
+    @staticmethod
+    def _validate_identifier(name: str) -> str:
+        if not isinstance(name, str) or not _IDENTIFIER_RE.fullmatch(name):
+            raise ValueError(f'invalid SQL identifier: {name!r}')
+        return name

ingestor/io/slots.py

@@ -0,0 +1,47 @@
+"""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:
+    return {
+        'frequency': tag['frequency'],
+        'names': _parse_names(address, tag),
+        'active_read': tag.get('active_read', {'enabled': False}),
+    }
+
+
+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 (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
@@ -467,26 +538,84 @@ class IngestorManager(SientiaMonitoring):
 
     async def unsubscribe_slot(self, slot: str):
         """
-        Unsubscribes a specific slot from all associated OPC servers.
+        Unsubscribes a specific slot from every OPC reader that holds a subscription for it.
 
-        This method removes all subscriptions for a given slot across all
-        OPC servers that were managing it. It ensures clean cleanup of
-        resources when slots are released or reconfigured.
+        The readers' own subscriptions are the source of truth, not managed_tags: the
+        main loop removes a slot from managed_tags before update_ingestor_manager asks
+        to unsubscribe it, both when manage_leases drops an extra lease (which already
+        unsubscribed it) and when update_slot_config finds the slot gone from Redis
+        (which did not). Looking the slot up in managed_tags raised KeyError in both
+        cases, which stopped the main loop and restarted the container; the second case
+        also left the subscription open on the OPC server. Unsubscribing a slot that no
+        reader holds is a no-op.
+
+        A reader that fails to unsubscribe is handled by _unsubscribe_reader: the other
+        readers are still unsubscribed and the main loop goes on.
 
         Args:
             slot (str): The name of the slot to unsubscribe.
 
-        Raises:
-            KeyError: If the specified slot does not exist in the managed tags.
-
         Side Effects:
-            - Removes subscriptions from all OPC servers for the specified slot
+            - Removes the slot's subscription from every OPC reader that has one
             - Cleans up subscription resources on the OPC servers
         """
 
-        for server in self.managed_tags[slot].keys():
-            if server in self.opc_managers:
-                await self.opc_managers[server].unsubscribe(slot)
+        # list() snapshots the readers: the loop awaits, so a reader added or removed
+        # meanwhile must not break the iteration. SonarQube flags it as an unnecessary
+        # list() (S7504); the issue is accepted in SonarQube for this reason.
+        for server, reader in list(self.opc_managers.items()):
+            if slot in reader.subscriptions:
+                await self._unsubscribe_reader(server, reader, slot)
+
+    async def _unsubscribe_reader(
+        self, server: str, reader: ServerReader | OpcManager, slot: str
+    ) -> None:
+        """
+        Unsubscribe one reader from a slot without letting a failure stop the main loop.
+
+        A failure (e.g. the legacy OpcManager, whose delete() raises when the server
+        session is gone) increments OPC_SUBSCRIPTION_ERRORS, sends
+        OPC_UNSUBSCRIBE_ERROR_<slot>:<server> and is logged. The slot is dropped from the
+        reader's books, as the v2 engine already does, so it can be subscribed again.
+
+        Known limitation: if delete() failed while the session is still alive (e.g. a
+        timeout), the server-side subscription may keep delivering data until that
+        session ends (reconnection or pod restart).
+
+        Args:
+            server (str): The reader's key in opc_managers
+            reader (ServerReader | OpcManager): The OPC reader
+            slot (str): The slot to unsubscribe
+        """
+        try:
+            await reader.unsubscribe(slot)
+        except Exception as e:
+            await self.emit_metric(
+                metric_object=metrics.OPC_SUBSCRIPTION_ERRORS,
+                tags={
+                    'pod_id': self.pod_id,
+                    'server': server,
+                    'slot': slot,
+                },
+            )
+            trace = traceback.format_exc()
+            await self.send_notification_async(
+                metadata=self.metadata,
+                notification_id=f'OPC_UNSUBSCRIBE_ERROR_{slot}:{server}',
+                message=f'Failed to unsubscribe slot {slot} from {server}: {e}',
+                block='opc_manager',
+                level=NotificationLevel.ERROR,
+                attachment_content=trace,
+            )
+            self.logger.error(trace)
+            reader.subscriptions.pop(slot, None)
+
+    async def _unsubscribe_removed_slots(self, slots: list[str]) -> None:
+        """Unsubscribe and forget slots whose configuration is gone from Redis."""
+        for slot in slots:
+            self.logger.warning(f'Slot {slot} is no longer configured; unsubscribing it')
+            await self.unsubscribe_slot(slot)
+            self.managed_tags.pop(slot, None)
 
     async def update_slot_config(self):
         """
@@ -496,7 +625,7 @@ class IngestorManager(SientiaMonitoring):
         This method performs the following steps:
         1. Renews the lease for each managed slot using the resource manager.
         2. Fetches the latest configuration for each slot.
-        3. Logs and removes slots whose configurations are no longer available.
+        3. Logs, unsubscribes and removes slots whose configurations are no longer available.
         4. Updates the configuration of slots if changes are detected.
         5. Unsubscribes and re-subscribes to slots with updated configurations.
         6. Removes slots from the managed tags if they are no longer valid.
@@ -518,15 +647,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)
+        # Unsubscribe right away: leaving it to update_ingestor_manager's snapshot diff
+        # would leak the subscription if anything in between failed
+        await self._unsubscribe_removed_slots(removed_slots)
+
+    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:
         """
@@ -634,7 +823,7 @@ class IngestorManager(SientiaMonitoring):
             self.logger.error(trace)
 
             self.logger.warning(f'Removing subscription from server {server} for slot {slot}')
-            await self.opc_managers[server].unsubscribe(slot)
+            await self._unsubscribe_reader(server, self.opc_managers[server], slot)
             return 2
         return 0
 

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},
+        )

ingestor/metrics.py

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

ingestor/proto/opc_gateway.proto

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

ingestor/proto/opc_gateway_pb2.py

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

ingestor/proto/opc_gateway_pb2_grpc.py

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

pyproject.toml

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

requirements-dev.txt

@@ -16,4 +16,10 @@ pytest-asyncio>=0.21.0   # Async test support (already in main requirements)
 
 # Development Tools
 ipython>=8.12.0          # Enhanced Python shell
-ipdb>=0.13.13            # IPython debugger
+ipdb>=0.13.13            # IPython 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
+# because workers install only from the cluster package index.
+grpcio-tools>=1.60

requirements.txt

@@ -1,5 +1,13 @@
 asyncua
 redis
+# TODO(SIENTIAPDE-1972): bump once sientia-dataops-library releases SIENTIAPDE-1590
+# (generic Mongo find_one/distinct/aggregate methods) — not in 1.12.2 yet.
+# Local dev uses an editable install of the local checkout (feature/SIENTIAPDE-1590).
 sientia_do==1.12.2
 prometheus_client
-pymongo
+pymongo
+nats-py>=2.6
+grpcio>=1.60
+# Runtime of the committed gRPC stubs: ingestor/proto/opc_gateway_pb2.py was
+# generated with protobuf 7.35.0 and validates the runtime version on import.
+protobuf>=7.35

scripts/seed_e2e.py

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

sonar-project.properties

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

tests/functional/conftest.py

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

tests/functional/test_single_node.py

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

tests/unit/adapters/test_grpc_server.py

@@ -0,0 +1,97 @@
+from unittest.mock import AsyncMock, MagicMock
+
+import grpc
+import pytest
+
+from ingestor.adapters.grpc_server import GrpcServer, ReadTagServicer
+from ingestor.proto.opc_gateway_pb2 import ReadTagRequest
+
+
+def make_request(server_id='srv1', addr='ns=2;i=5'):
+    return ReadTagRequest(server_id=server_id, addr=addr, data_type='float')
+
+
+@pytest.mark.asyncio
+async def test_read_tag_maps_full_result_onto_response():
+    active_read = AsyncMock()
+    active_read.read_now.return_value = {
+        'value': 1.5,
+        'opc_status_code': 'Good',
+        'source_timestamp': '2026-07-15T10:00:00-03:00',
+        'server_timestamp': '2026-07-15T10:00:01-03:00',
+        'persisted': False,
+    }
+    servicer = ReadTagServicer(active_read, MagicMock())
+
+    response = await servicer.ReadTag(make_request(), AsyncMock())
+
+    assert response.value == '1.5'
+    assert response.opc_status_code == 'Good'
+    assert response.source_timestamp == '2026-07-15T10:00:00-03:00'
+    assert response.server_timestamp == '2026-07-15T10:00:01-03:00'
+    assert response.persisted is False
+
+
+@pytest.mark.asyncio
+async def test_read_tag_none_timestamps_become_empty_string():
+    active_read = AsyncMock()
+    active_read.read_now.return_value = {
+        'value': 1.5,
+        'opc_status_code': None,
+        'source_timestamp': None,
+        'server_timestamp': None,
+        'persisted': False,
+    }
+    servicer = ReadTagServicer(active_read, MagicMock())
+
+    response = await servicer.ReadTag(make_request(), AsyncMock())
+
+    assert response.opc_status_code == ''
+    assert response.source_timestamp == ''
+    assert response.server_timestamp == ''
+
+
+@pytest.mark.asyncio
+async def test_read_tag_aborts_with_internal_on_engine_failure():
+    active_read = AsyncMock()
+    active_read.read_now.side_effect = RuntimeError('boom')
+    logger = MagicMock()
+    servicer = ReadTagServicer(active_read, logger)
+    context = AsyncMock()
+    context.abort.side_effect = grpc.RpcError('aborted')
+
+    request = make_request()
+    with pytest.raises(grpc.RpcError):
+        await servicer.ReadTag(request, context)
+
+    context.abort.assert_awaited_once()
+    args = context.abort.call_args.args
+    assert args[0] == grpc.StatusCode.INTERNAL
+    logger.error.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_read_tag_uses_slot_lookup_names_when_provided():
+    active_read = AsyncMock()
+    active_read.read_now.return_value = {
+        'value': 1,
+        'opc_status_code': 'Good',
+        'source_timestamp': None,
+        'server_timestamp': None,
+        'persisted': True,
+    }
+    names = [{'collection': 'c1', 'name': 't1'}]
+    slot_lookup = MagicMock(return_value=names)
+    servicer = ReadTagServicer(active_read, MagicMock(), slot_lookup=slot_lookup)
+
+    await servicer.ReadTag(make_request(server_id='srv1', addr='ns=2;i=5'), AsyncMock())
+
+    slot_lookup.assert_called_once_with('srv1', 'ns=2;i=5')
+    active_read.read_now.assert_awaited_once_with('srv1', 'ns=2;i=5', names=names)
+
+
+@pytest.mark.asyncio
+async def test_grpc_server_stop_before_start_does_not_raise():
+    server = GrpcServer(active_read=AsyncMock(), logger=MagicMock())
+
+    await server.stop()  # no raise

tests/unit/adapters/test_nats_consumer.py

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

tests/unit/conftest.py

@@ -0,0 +1,20 @@
+from unittest.mock import AsyncMock, MagicMock
+
+from pytest import fixture
+
+
+@fixture
+def reader_double():
+    """Factory of OPC reader doubles holding subscriptions for the given slots.
+
+    Like the real readers (ServerReader, OpcManager), `subscriptions` is keyed by slot
+    and `unsubscribe` drops the slot from it.
+    """
+
+    def make(*slots):
+        reader = MagicMock()
+        reader.subscriptions = {slot: MagicMock() for slot in slots}
+        reader.unsubscribe = AsyncMock(side_effect=lambda slot: reader.subscriptions.pop(slot))
+        return reader
+
+    return make

tests/unit/connection/test_certs.py

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

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

tests/unit/connection/test_pool.py

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

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

tests/unit/engines/test_active_read.py

@@ -0,0 +1,179 @@
+from datetime import datetime
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from asyncua import ua
+
+import ingestor.metrics as metrics
+from ingestor.engines.active_read import ActiveReadEngine
+
+
+def make_engine():
+    eng = ActiveReadEngine(
+        pool=AsyncMock(),
+        sink=AsyncMock(),
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=AsyncMock(),
+    )
+    eng.emit_metric = AsyncMock()
+    return eng
+
+
+def make_dv(value=42.0, status='Good', source_ts=None, server_ts=None):
+    source_ts = source_ts or datetime(2026, 7, 13, 12, 0, 0)
+    server_ts = server_ts or source_ts
+    return ua.DataValue(
+        ua.Variant(value),
+        StatusCode=ua.StatusCode(getattr(ua.StatusCodes, status)),
+        SourceTimestamp=source_ts,
+        ServerTimestamp=server_ts,
+    )
+
+
+def make_client(dv=None):
+    node = MagicMock()
+    node.read_data_value = AsyncMock(return_value=dv or make_dv())
+    client = MagicMock()
+    client.get_node = MagicMock(return_value=node)
+    return client, node
+
+
+NAMES = [{'tag_name': 'a', 'model_id': 'm1', 'collection': 'coll'}]
+
+
+@pytest.mark.asyncio
+async def test_read_now_persists_and_returns_fields_when_names_given():
+    eng = make_engine()
+    client, node = make_client()
+    eng.pool.get.return_value = client
+
+    result = await eng.read_now('srv1', 'ns=2;i=5', names=NAMES)
+
+    client.get_node.assert_called_once_with('ns=2;i=5')
+    node.read_data_value.assert_awaited_once()
+    eng.sink.publish_reading.assert_awaited_once()
+    _, kwargs = eng.sink.publish_reading.call_args
+    assert kwargs['tag'] == 'ns=2;i=5'
+    assert kwargs['value'] == 42.0
+    assert kwargs['names'] == NAMES
+    assert result['value'] == 42.0
+    assert result['opc_status_code'] == 'Good'
+    assert result['persisted'] is True
+    assert result['source_timestamp'].startswith('2026-07-13 12:00:00')
+    assert result['server_timestamp'].startswith('2026-07-13 12:00:00')
+    eng.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_ACTIVE_READ_TOTAL,
+        tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'result': 'success'},
+    )
+    latency_call = next(
+        c
+        for c in eng.emit_metric.await_args_list
+        if c.kwargs['metric_object'] is metrics.OPC_ACTIVE_READ_LATENCY
+    )
+    assert latency_call.kwargs['method'] == 'observe'
+    assert latency_call.kwargs['value'] >= 0
+    lag_call = next(
+        c
+        for c in eng.emit_metric.await_args_list
+        if c.kwargs['metric_object'] is metrics.OPC_READ_LAG
+    )
+    assert lag_call.kwargs['method'] == 'set'
+
+
+@pytest.mark.asyncio
+async def test_read_now_without_names_does_not_persist():
+    eng = make_engine()
+    client, _ = make_client()
+    eng.pool.get.return_value = client
+
+    result = await eng.read_now('srv1', 'ns=2;i=5')
+
+    eng.sink.publish_reading.assert_not_awaited()
+    assert result['persisted'] is False
+
+
+@pytest.mark.asyncio
+async def test_read_now_error_emits_metric_and_reraises():
+    eng = make_engine()
+    node = MagicMock()
+    node.read_data_value = AsyncMock(side_effect=RuntimeError('boom'))
+    client = MagicMock()
+    client.get_node = MagicMock(return_value=node)
+    eng.pool.get.return_value = client
+
+    with pytest.raises(RuntimeError, match='boom'):
+        await eng.read_now('srv1', 'ns=2;i=5')
+
+    eng.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_ACTIVE_READ_TOTAL,
+        tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'result': 'error'},
+    )
+    eng.sink.publish_reading.assert_not_awaited()
+
+
+def make_slot(cycle_multiple=None, enabled=True):
+    active_read = {'enabled': enabled}
+    if cycle_multiple is not None:
+        active_read['cycle_multiple'] = cycle_multiple
+    return {
+        'srv_name': {
+            'server_id': 'srv1',
+            'url': 'opc.tcp://x',
+            'tags': {
+                'ns=2;i=5': {'frequency': 1000, 'names': NAMES, 'active_read': active_read},
+            },
+        }
+    }
+
+
+@pytest.mark.asyncio
+async def test_scheduled_tick_respects_cycle_multiple():
+    eng = make_engine()
+    eng.read_now = AsyncMock()
+    slot = make_slot(cycle_multiple=3)
+
+    await eng.scheduled_tick(slot)
+    await eng.scheduled_tick(slot)
+    eng.read_now.assert_not_awaited()
+    await eng.scheduled_tick(slot)
+    eng.read_now.assert_awaited_once_with(
+        'srv1', 'ns=2;i=5', names=NAMES, server_cfg={'server_id': 'srv1', 'url': 'opc.tcp://x'}
+    )
+
+
+@pytest.mark.asyncio
+async def test_scheduled_tick_skips_disabled_tags():
+    eng = make_engine()
+    eng.read_now = AsyncMock()
+    slot = make_slot(enabled=False)
+
+    await eng.scheduled_tick(slot)
+
+    eng.read_now.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_scheduled_tick_continues_past_tag_error():
+    eng = make_engine()
+    bad_slot = make_slot(cycle_multiple=1)
+    good_names = [{'tag_name': 'b', 'model_id': 'm2', 'collection': 'coll'}]
+    bad_slot['srv_name']['tags']['ns=2;i=6'] = {
+        'frequency': 1000,
+        'names': good_names,
+        'active_read': {'enabled': True, 'cycle_multiple': 1},
+    }
+
+    calls = []
+
+    async def fake_read_now(server_id, addr, names=None, server_cfg=None):
+        calls.append(addr)
+        if addr == 'ns=2;i=5':
+            raise RuntimeError('boom')
+
+    eng.read_now = AsyncMock(side_effect=fake_read_now)
+
+    await eng.scheduled_tick(bad_slot)
+
+    assert set(calls) == {'ns=2;i=5', 'ns=2;i=6'}
+    eng.logger.error.assert_called_once()

tests/unit/engines/test_subscription.py

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

@@ -0,0 +1,259 @@
+from datetime import datetime
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from asyncua import ua
+
+import ingestor.metrics as metrics
+from ingestor.engines.write import WriteEngine
+
+
+def make_engine(**kwargs):
+    eng = WriteEngine(
+        pool=AsyncMock(),
+        sink=AsyncMock(),
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=AsyncMock(),
+        **kwargs,
+    )
+    eng.emit_metric = AsyncMock()
+    return eng
+
+
+def make_client(node=None):
+    node = node or make_node()
+    client = MagicMock()
+    client.get_node = MagicMock(return_value=node)
+    return client
+
+
+def make_node():
+    node = MagicMock()
+    node.write_value = AsyncMock()
+    return node
+
+
+def base_cmd(**overrides):
+    cmd = {
+        'server_id': 'srv1',
+        'addr': 'ns=2;i=5',
+        'value': 42.0,
+        'data_type': 'float',
+        'tag_type': 'prediction',
+        'model_id': 'm1',
+        'model_name': 'model-1',
+        'timestamp': '2026-07-13T12:00:00Z',
+        'schema': 'sientia_data',
+        'table_name': 'predictions_m1',
+    }
+    cmd.update(overrides)
+    return cmd
+
+
+@pytest.mark.asyncio
+async def test_write_success_writes_datavalue_and_returns_ok():
+    eng = make_engine()
+    node = make_node()
+    client = make_client(node)
+    eng.pool.get.return_value = client
+
+    result = await eng.write(base_cmd())
+
+    client.get_node.assert_called_once_with('ns=2;i=5')
+    node.write_value.assert_awaited_once()
+    dv = node.write_value.await_args.args[0]
+    assert isinstance(dv, ua.DataValue)
+    assert dv.Value.Value == 42.0
+    assert dv.Value.VariantType == ua.VariantType.Float
+    assert result.ok is True
+    assert result.status == 'ok'
+    assert result.opc_status_code == 'Good'
+    eng.sink.update_write_status.assert_awaited_once_with(base_cmd(), result)
+
+
+@pytest.mark.asyncio
+async def test_write_success_emits_metrics():
+    eng = make_engine()
+    eng.pool.get.return_value = make_client()
+
+    await eng.write(base_cmd())
+
+    eng.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_WRITE_TOTAL,
+        tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'model_id': 'm1', 'result': 'success'},
+    )
+    status_call = next(
+        c
+        for c in eng.emit_metric.await_args_list
+        if c.kwargs['metric_object'] is metrics.OPC_WRITE_STATUS_BY_MODEL
+    )
+    assert status_call.kwargs['value'] == 1
+    assert status_call.kwargs['tags'] == {'pod_id': eng.pod_id, 'model_id': 'm1', 'status': 'ok'}
+    latency_call = next(
+        c
+        for c in eng.emit_metric.await_args_list
+        if c.kwargs['metric_object'] is metrics.OPC_WRITE_LATENCY
+    )
+    assert latency_call.kwargs['method'] == 'observe'
+    assert latency_call.kwargs['value'] >= 0
+
+
+@pytest.mark.asyncio
+async def test_write_defaults_carry_statuscode_and_timestamp():
+    eng = make_engine(write_timestamp=True, write_quality=True)
+    node = make_node()
+    eng.pool.get.return_value = make_client(node)
+
+    await eng.write(base_cmd())
+
+    dv = node.write_value.await_args.args[0]
+    assert dv.StatusCode is not None
+    assert dv.StatusCode.is_good()
+    assert dv.SourceTimestamp == datetime.fromisoformat('2026-07-13T12:00:00Z')
+
+
+@pytest.mark.asyncio
+async def test_write_engine_flags_off_omit_statuscode_and_timestamp():
+    eng = make_engine(write_timestamp=False, write_quality=False)
+    node = make_node()
+    eng.pool.get.return_value = make_client(node)
+
+    await eng.write(base_cmd())
+
+    dv = node.write_value.await_args.args[0]
+    assert dv.StatusCode is None
+    assert dv.SourceTimestamp is None
+
+
+@pytest.mark.asyncio
+async def test_per_message_flag_override_disables_timestamp():
+    # Engine default is True, but the command explicitly turns it off.
+    eng = make_engine(write_timestamp=True, write_quality=True)
+    node = make_node()
+    eng.pool.get.return_value = make_client(node)
+
+    await eng.write(base_cmd(write_timestamp=False))
+
+    dv = node.write_value.await_args.args[0]
+    assert dv.SourceTimestamp is None
+    assert dv.StatusCode is not None  # write_quality untouched, stays on
+
+
+@pytest.mark.asyncio
+async def test_write_bad_type_returns_error_without_raising():
+    eng = make_engine()
+
+    result = await eng.write(base_cmd(value='abc', data_type='int'))
+
+    assert result.ok is False
+    assert result.status == 'error'
+    eng.sink.update_write_status.assert_awaited_once()
+    eng.pool.get.assert_not_awaited()
+    eng.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_ERROR_BY_TYPE,
+        tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'error_type': 'conversion'},
+    )
+    eng.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_WRITE_TOTAL,
+        tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'model_id': 'm1', 'result': 'error'},
+    )
+
+
+@pytest.mark.asyncio
+async def test_write_none_data_type_returns_error_without_raising():
+    eng = make_engine()
+
+    result = await eng.write(base_cmd(data_type=None))
+
+    assert result.ok is False
+    assert result.status == 'error'
+    eng.sink.update_write_status.assert_awaited_once()
+    eng.pool.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_write_missing_value_key_returns_error_without_raising():
+    eng = make_engine()
+    cmd = base_cmd()
+    del cmd['value']
+
+    result = await eng.write(cmd)
+
+    assert result.ok is False
+    assert result.status == 'error'
+    eng.sink.update_write_status.assert_awaited_once()
+    eng.pool.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_write_missing_data_type_key_returns_error_without_raising():
+    eng = make_engine()
+    cmd = base_cmd()
+    del cmd['data_type']
+
+    result = await eng.write(cmd)
+
+    assert result.ok is False
+    assert result.status == 'error'
+    eng.sink.update_write_status.assert_awaited_once()
+    eng.pool.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_write_value_raises_maps_to_error_result_not_raised():
+    eng = make_engine()
+    node = make_node()
+    node.write_value.side_effect = RuntimeError('boom')
+    eng.pool.get.return_value = make_client(node)
+
+    result = await eng.write(base_cmd())
+
+    assert result.ok is False
+    assert result.status == 'error'
+    assert result.opc_status_code == 'RuntimeError'
+    eng.sink.update_write_status.assert_awaited_once()
+    eng.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_ERROR_BY_TYPE,
+        tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'error_type': 'RuntimeError'},
+    )
+
+
+@pytest.mark.asyncio
+async def test_write_session_bad_status_code_error_maps_to_session_bad():
+    from asyncua.ua.uaerrors import BadSessionClosed
+
+    eng = make_engine()
+    node = make_node()
+    node.write_value.side_effect = BadSessionClosed('closed')
+    eng.pool.get.return_value = make_client(node)
+
+    result = await eng.write(base_cmd())
+
+    assert result.ok is False
+    assert result.status == 'session_bad'
+
+
+@pytest.mark.asyncio
+async def test_rate_limit_sleeps_between_back_to_back_writes():
+    eng = make_engine(min_pause_s=0.05)
+    eng.pool.get.return_value = make_client()
+
+    with patch('ingestor.engines.write.asyncio.sleep', new_callable=AsyncMock) as sleep_mock:
+        await eng.write(base_cmd())
+        sleep_mock.assert_not_awaited()
+        await eng.write(base_cmd())
+        sleep_mock.assert_awaited_once()
+        remaining = sleep_mock.await_args.args[0]
+        assert 0 < remaining <= 0.05
+
+
+@pytest.mark.asyncio
+async def test_no_rate_limit_when_min_pause_is_zero():
+    eng = make_engine(min_pause_s=0.0)
+    eng.pool.get.return_value = make_client()
+
+    with patch('ingestor.engines.write.asyncio.sleep', new_callable=AsyncMock) as sleep_mock:
+        await eng.write(base_cmd())
+        await eng.write(base_cmd())
+        sleep_mock.assert_not_awaited()

tests/unit/io/test_data_sink_publish.py

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

@@ -0,0 +1,89 @@
+from unittest.mock import MagicMock
+
+import pytest
+
+from ingestor.engines.write import WriteResult
+from ingestor.io.data_sink import DataSink
+
+
+def make_sink():
+    postgres = MagicMock()
+    connection = MagicMock()
+    postgres.engine.begin.return_value.__enter__.return_value = connection
+    sink = DataSink(data_manager=MagicMock(), postgres=postgres)
+    return sink, connection
+
+
+@pytest.mark.asyncio
+async def test_update_write_status_targets_model_row():
+    sink, connection = make_sink()
+    cmd = {
+        'schema': 'sientia_data',
+        'table_name': 'pred_m1',
+        'model_id': '1',
+        'timestamp': '2026-07-13T00:00:00Z',
+    }
+    result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='')
+
+    await sink.update_write_status(cmd, result)
+
+    assert connection.execute.call_count == 1
+    stmt, params = connection.execute.call_args[0]
+    sql = str(stmt)
+    assert 'update' in sql.lower() and 'write_status' in sql.lower()
+    assert 'sientia_data.pred_m1' in sql
+    # values must be bound params, never interpolated into the SQL text itself
+    assert params == {'status': 'ok', 'timestamp': '2026-07-13T00:00:00Z', 'model_id': '1'}
+    assert 'ok' not in sql
+    assert '2026-07-13T00:00:00Z' not in sql
+
+
+@pytest.mark.asyncio
+async def test_update_write_status_rejects_bad_schema():
+    sink, connection = make_sink()
+    cmd = {
+        'schema': 'a; DROP TABLE x',
+        'table_name': 'pred_m1',
+        'model_id': '1',
+        'timestamp': '2026-07-13T00:00:00Z',
+    }
+    result = WriteResult(ok=False, status='error', opc_status_code='Bad', message='x')
+
+    with pytest.raises(ValueError):
+        await sink.update_write_status(cmd, result)
+
+    connection.execute.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_update_write_status_rejects_trailing_newline_identifier():
+    sink, connection = make_sink()
+    cmd = {
+        'schema': 'sientia_data\n',
+        'table_name': 'pred_m1',
+        'model_id': '1',
+        'timestamp': '2026-07-13T00:00:00Z',
+    }
+    result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='')
+
+    with pytest.raises(ValueError):
+        await sink.update_write_status(cmd, result)
+
+    connection.execute.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_update_write_status_rejects_bad_table_name():
+    sink, connection = make_sink()
+    cmd = {
+        'schema': 'sientia_data',
+        'table_name': 'pred_m1; DROP TABLE x',
+        'model_id': '1',
+        'timestamp': '2026-07-13T00:00:00Z',
+    }
+    result = WriteResult(ok=False, status='error', opc_status_code='Bad', message='x')
+
+    with pytest.raises(ValueError):
+        await sink.update_write_status(cmd, result)
+
+    connection.execute.assert_not_called()

tests/unit/io/test_slots.py

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

@@ -1,9 +1,16 @@
-from unittest.mock import AsyncMock, MagicMock, call, patch
+import gc
+from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
 
 from pytest import fixture, mark
 from sientia_do.notifications.models import NotificationLevel
 
+import ingestor.metrics as metrics
+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
+from ingestor.managers.opc_manager import OpcManager
+from ingestor.managers.server_reader import ServerReader
 
 metadata = {
     'metadata': {
@@ -15,6 +22,33 @@ 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
+    # cyclic-referenced instance from a previous test is swept mid-test by the
+    # generational GC, it can double-call whatever asyncio mock happens to be
+    # patched at that moment (e.g. test___del__). Force a deterministic sweep
+    # after each test so finalizers run in their own test's scope. Upgrade path:
+    # only needed if IngestorManager stops relying on __del__ for cleanup.
+    yield
+    gc.collect()
+
+
 @fixture
 @patch('ingestor.managers.ingestor_manager.DataManager')
 @patch('ingestor.managers.ingestor_manager.ResourceManager')
@@ -39,12 +73,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},
@@ -59,7 +93,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',
@@ -87,59 +121,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)
 
@@ -161,11 +222,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()
 
@@ -205,9 +269,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'})
@@ -254,7 +317,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
 
@@ -418,25 +483,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
@@ -464,40 +551,172 @@ async def test_get_slot_leases_1_failure(ingestor_manager):
 
 
 @mark.asyncio
-async def test_unsubscribe_slot(ingestor_manager):
+async def test_unsubscribe_slot(ingestor_manager, reader_double):
     ingestor_manager.managed_tags = {
         'slot1': {'server1': {'tags': 'config1'}, 'server2': {'tags': 'config2'}},
         'slot2': {'server3': {'tags': 'config3'}, 'server1': {'tags': 'config1'}},
     }
     ingestor_manager.opc_managers = {
-        'server1': AsyncMock(),
-        'server2': AsyncMock(),
-        'server3': AsyncMock(),
+        'server1': reader_double('slot1', 'slot2'),
+        'server2': reader_double('slot1'),
+        'server3': reader_double('slot2'),
     }
     await ingestor_manager.unsubscribe_slot('slot1')
 
     ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot1')
     ingestor_manager.opc_managers['server2'].unsubscribe.assert_called_once_with('slot1')
     ingestor_manager.opc_managers['server3'].unsubscribe.assert_not_called()
+    assert list(ingestor_manager.opc_managers['server1'].subscriptions) == ['slot2']
 
 
 @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'},
+async def test_unsubscribe_slot_not_in_managed_tags(ingestor_manager, reader_double):
+    """The slot is already gone from managed_tags: the readers still holding it are unsubscribed"""
+    ingestor_manager.managed_tags = {'slot1': {'server1': {'tags': 'config1'}}}
+    ingestor_manager.opc_managers = {
+        'server1': reader_double('slot1', 'slot3'),
+        'server2': reader_double('slot3'),
     }
 
+    await ingestor_manager.unsubscribe_slot('slot3')
+
+    ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot3')
+    ingestor_manager.opc_managers['server2'].unsubscribe.assert_called_once_with('slot3')
+
+
+@mark.asyncio
+async def test_unsubscribe_slot_twice_is_a_no_op(ingestor_manager, reader_double):
+    """A second unsubscribe of the same slot (lease already dropped) does nothing and never raises"""
+    ingestor_manager.managed_tags = {}
+    ingestor_manager.opc_managers = {'server1': reader_double('slot2')}
+
+    await ingestor_manager.unsubscribe_slot('slot2')
+    await ingestor_manager.unsubscribe_slot('slot2')
+
+    ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot2')
+
+
+@mark.asyncio
+async def test_unsubscribe_slot_reader_failure_does_not_raise(ingestor_manager, reader_double):
+    """A reader that fails to unsubscribe is notified and forgotten; the others still go"""
+    broken = reader_double('slot1')
+    broken.unsubscribe = AsyncMock(side_effect=ConnectionError('session closed'))
+    healthy = reader_double('slot1')
+    ingestor_manager.opc_managers = {'server1': broken, 'server2': healthy}
+
+    await ingestor_manager.unsubscribe_slot('slot1')
+
+    healthy.unsubscribe.assert_called_once_with('slot1')
+    assert 'slot1' not in broken.subscriptions
+    ingestor_manager.emit_metric.assert_called_once_with(
+        metric_object=metrics.OPC_SUBSCRIPTION_ERRORS,
+        tags={'pod_id': ingestor_manager.pod_id, 'server': 'server1', 'slot': 'slot1'},
+    )
+    ingestor_manager.send_notification_async.assert_called_once_with(
+        metadata=ingestor_manager.metadata,
+        notification_id='OPC_UNSUBSCRIBE_ERROR_slot1:server1',
+        message='Failed to unsubscribe slot slot1 from server1: session closed',
+        block='opc_manager',
+        level=NotificationLevel.ERROR,
+        attachment_content=ANY,
+    )
+
+
+@mark.asyncio
+async def test_unsubscribe_slot_with_real_readers(ingestor_manager):
+    """Contract: both real readers keep `subscriptions` keyed by slot, and
+    unsubscribe_slot closes the slot's subscription on each of them"""
+    v2_subscription = AsyncMock()
+    client = MagicMock(create_subscription=AsyncMock(return_value=v2_subscription))
+    pool = MagicMock(get=AsyncMock(return_value=client), evict=AsyncMock())
+    server_reader = ServerReader(
+        name='kepserver',
+        server_config={
+            'server_id': '1',
+            'url': 'opc.tcp://kepserver:49320',
+            'subscription_period_ms': 1000.0,
+        },
+        pool=pool,
+        sink=AsyncMock(),
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=AsyncMock(),
+        metadata={},
+    )
+    server_reader.engine.emit_metric = AsyncMock()
+    await server_reader.create_subscription('slot1')
+
+    with patch('ingestor.managers.opc_manager.metrics'):
+        legacy_reader = OpcManager(
+            name='legacy',
+            url='opc.tcp://legacy:4840',
+            data_manager=AsyncMock(),
+            subscription_period_ms=1000,
+            logger=MagicMock(),
+            server_uri='opc.tcp://legacy:4840',
+            notification_handler=MagicMock(),
+            metadata={},
+            metrics_controller=AsyncMock(),
+        )
+    legacy_reader.emit_metric = AsyncMock()
+    legacy_subscription = AsyncMock()
+    legacy_reader.client = MagicMock(
+        create_subscription=AsyncMock(return_value=legacy_subscription)
+    )
+    await legacy_reader.create_subscription('slot1')
+
+    assert 'slot1' in server_reader.subscriptions
+    assert 'slot1' in legacy_reader.subscriptions
+    ingestor_manager.opc_managers = {'kepserver': server_reader, 'legacy': legacy_reader}
+
+    await ingestor_manager.unsubscribe_slot('slot1')
+
+    v2_subscription.delete.assert_awaited_once()
+    legacy_subscription.delete.assert_awaited_once()
+    assert 'slot1' not in server_reader.subscriptions
+    assert 'slot1' not in legacy_reader.subscriptions
+    pool.pin.assert_called_once_with('1')
+    pool.unpin.assert_called_once_with('1')
+
+
+@mark.asyncio
+async def test_update_slot_config_unsubscribes_removed_slot(ingestor_manager, reader_double):
+    """A slot deleted from Redis is unsubscribed right away, not left to the main loop"""
+    ingestor_manager.managed_tags = {
+        'slot1': parse_slot(raw_slot_v1('Same')),
+        'slot2': parse_slot(raw_slot_v1('Same')),
+    }
+    reader = reader_double('slot1', 'slot2')
+    ingestor_manager.opc_managers = {'test_server': reader}
     ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
-        side_effect=[{'config': 'updated_config'}, {'config': 'new_config'}, None]
+        side_effect=lambda slot: raw_slot_v1('Same') if slot == 'slot1' else 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'}
+    reader.unsubscribe.assert_called_once_with('slot2')
+    assert list(ingestor_manager.managed_tags) == ['slot1']
+    assert list(reader.subscriptions) == ['slot1']
+
+
+@mark.asyncio
+async def test_update_slot_config(ingestor_manager):
+    ingestor_manager.managed_tags = {
+        '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=[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'] == 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')
@@ -562,6 +781,30 @@ async def test_manage_server_create_subscription_failure(ingestor_manager):
     ingestor_manager.opc_managers['server1'].subscribe.assert_not_called()
 
 
+@mark.asyncio
+async def test_manage_server_subscribe_and_unsubscribe_failure(ingestor_manager):
+    """subscribe() fails and so does the cleanup unsubscribe (legacy reader, session
+    gone): manage_server reports the failure and nothing escapes to the main loop"""
+    reader = AsyncMock()
+    reader.subscriptions = {}
+    reader.subscribe.side_effect = Exception('Subscription error')
+    reader.unsubscribe.side_effect = ConnectionError('session closed')
+    ingestor_manager.opc_managers = {'server1': reader}
+    server_config = {'tags': {'tag1': 'config1'}}
+
+    result = await ingestor_manager.manage_server('slot1', 'server1', server_config, server_config)
+
+    assert result == 2
+    reader.unsubscribe.assert_called_once_with('slot1')
+    notified = [
+        c.kwargs['notification_id'] for c in ingestor_manager.send_notification_async.call_args_list
+    ]
+    assert notified == [
+        'OPC_SUBSCRIPTION_ERROR_slot1:server1',
+        'OPC_UNSUBSCRIBE_ERROR_slot1:server1',
+    ]
+
+
 @mark.asyncio
 async def test_manage_server(ingestor_manager):
     ingestor_manager.opc_managers = {'server1': AsyncMock(), 'server2': AsyncMock()}
@@ -639,3 +882,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/proto/test_proto.py

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

tests/unit/test_app.py

@@ -43,6 +43,15 @@ def mock_app_env(monkeypatch):
     monkeypatch.setattr(app, 'Ingestor', mocks['Ingestor'])
     monkeypatch.setattr(asyncio, 'sleep', mocks['asyncio_sleep'])
 
+    mocks['GatewayConfig'] = MagicMock()
+    mocks['GatewayRuntime'] = MagicMock()
+    monkeypatch.setattr(app, 'GatewayConfig', mocks['GatewayConfig'])
+    monkeypatch.setattr(app, 'GatewayRuntime', mocks['GatewayRuntime'])
+    mock_runtime_instance = mocks['GatewayRuntime'].return_value
+    mock_runtime_instance.start = AsyncMock()
+    mock_runtime_instance.idle_sweep = AsyncMock()
+    mock_runtime_instance.stop = AsyncMock()
+
     monkeypatch.setattr(
         app.metrics.APP_UP,
         'labels',
@@ -77,6 +86,7 @@ def mock_app_env(monkeypatch):
     mock_ingestor_instance = mocks['Ingestor'].return_value
     mock_ingestor_instance.poll_interval = 0.01
     mock_ingestor_instance.logger = MagicMock()
+    mock_ingestor_instance.ingestor_manager = MagicMock()
 
     # Make async methods async mocks
     mock_ingestor_instance.prepare_ingestor = AsyncMock()
@@ -124,6 +134,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_converters.py

@@ -0,0 +1,47 @@
+import pytest
+
+from ingestor.converters import ConversionError, to_typed_value
+
+
+@pytest.mark.parametrize(
+    'raw,expected',
+    [
+        ('false', False),
+        ('False', False),
+        ('0', False),
+        ('no', False),
+        ('off', False),
+        ('', False),
+        ('true', True),
+        ('1', True),
+        (True, True),
+    ],
+)
+def test_bool_from_json_string(raw, expected):
+    assert to_typed_value(raw, 'boolean') is expected
+
+
+def test_float_and_int():
+    assert to_typed_value('42.5', 'float') == 42.5
+    assert to_typed_value('7', 'int') == 7
+
+
+@pytest.mark.parametrize(
+    'raw,dtype',
+    [
+        (None, 'int'),
+        ('abc', 'float'),
+        ('123.123', 'int'),
+        (None, 'float'),
+    ],
+)
+def test_invalid_raises_conversion_error(raw, dtype):
+    with pytest.raises(ConversionError):
+        to_typed_value(raw, dtype)
+
+
+@pytest.mark.parametrize('dtype', [None, 123])
+def test_non_string_data_type_raises_conversion_error(dtype):
+    # contract: never a bare AttributeError/TypeError, always ConversionError
+    with pytest.raises(ConversionError):
+        to_typed_value('5', dtype)

tests/unit/test_gateway_runtime.py

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

tests/unit/test_ingestor.py

@@ -1,8 +1,10 @@
+from copy import deepcopy
 from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
 
 from pytest import fixture, mark
 
 from ingestor.ingestor import Ingestor
+from ingestor.managers.ingestor_manager import IngestorManager
 
 
 @patch('ingestor.ingestor.getenv')
@@ -229,6 +231,10 @@ async def test_manage_leases_no_available_slots_extra_sltos(ingestor_manager_sta
     ingestor_manager_started.ingestor_manager.drop_slot_leases.assert_called_once_with(
         ['tag2', 'tag3']
     )
+    # The extra slots are unsubscribed before their leases are given back
+    calls = [c[0] for c in ingestor_manager_started.ingestor_manager.mock_calls]
+    assert calls.index('drop_slot_leases') > calls.index('unsubscribe_slot')
+    assert list(ingestor_manager_started.ingestor_manager.managed_tags) == ['tag1']
 
 
 @mark.asyncio
@@ -332,3 +338,52 @@ async def test_update_ingestor_manager_no_ingestor_manager(ingestor_manager_star
     ingestor_manager_started.ingestor_manager = None
 
     await ingestor_manager_started.update_ingestor_manager({})
+
+
+def _manager_with_real_unsubscribe(managed_tags, readers):
+    """A mocked IngestorManager whose unsubscribe path (unsubscribe_slot and
+    _unsubscribe_reader) is the real one."""
+    manager = AsyncMock(update_opc_servers=AsyncMock(), subscribe_to_tags=AsyncMock())
+    manager.managed_tags = managed_tags
+    manager.opc_managers = readers
+    manager.unsubscribe_slot = lambda slot: IngestorManager.unsubscribe_slot(manager, slot)
+    manager._unsubscribe_reader = lambda server, reader, slot: IngestorManager._unsubscribe_reader(
+        manager, server, reader, slot
+    )
+    return manager
+
+
+@mark.asyncio
+async def test_dropped_extra_lease_does_not_stop_the_loop(ingestor_manager_started, reader_double):
+    """Regression: manage_leases drops slot 2, then update_ingestor_manager sees slot 2 in
+    the snapshot only and unsubscribes it again. That used to raise KeyError and stop
+    the main loop (container restart)."""
+    reader = reader_double('1', '2')
+    ingestor_manager_started.ingestor_manager = _manager_with_real_unsubscribe(
+        {'1': {'kepserver': {}}, '2': {'kepserver': {}}}, {'kepserver': reader}
+    )
+    snapshot = deepcopy(ingestor_manager_started.ingestor_manager.managed_tags)
+
+    await ingestor_manager_started.manage_leases(0, 0, 1)
+    await ingestor_manager_started.update_ingestor_manager(snapshot)
+
+    reader.unsubscribe.assert_called_once_with('2')
+    assert list(reader.subscriptions) == ['1']
+    assert list(ingestor_manager_started.ingestor_manager.managed_tags) == ['1']
+
+
+@mark.asyncio
+async def test_slot_removed_from_redis_is_unsubscribed(ingestor_manager_started, reader_double):
+    """Regression: update_slot_config drops a slot deleted by the orchestrator from
+    managed_tags without unsubscribing it; update_ingestor_manager must close its
+    subscription instead of raising KeyError."""
+    reader = reader_double('1', '3')
+    ingestor_manager_started.ingestor_manager = _manager_with_real_unsubscribe(
+        {'1': {'kepserver': {}}}, {'kepserver': reader}
+    )
+    snapshot = {'1': {'kepserver': {}}, '3': {'kepserver': {}}}
+
+    await ingestor_manager_started.update_ingestor_manager(snapshot)
+
+    reader.unsubscribe.assert_called_once_with('3')
+    assert list(reader.subscriptions) == ['1']

tests/unit/test_metrics.py

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