TEST ONLY: 1975 gateway (8ab9af1) + PR #46 (279484b), as #43 will be after #46

DataManager logged the full connection string, user and password included,
at startup. Log it through redact_uri (mongodb://***@host:port/...).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CLAUDE.md

@@ -60,6 +60,8 @@ Any change to slot/lease counts should be traced through `manage_leases`/`manage
 
 **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.
 
+**Writes** (SIENTIAPDE-1975): `NatsWriteConsumer` → `WriteEngine` → pool, then `DataSink.update_write_status`. Commands (C2) carry only `server_id`; `ServerConfigResolver` (`connection/servers.py`) resolves the connection (command override → live pooled connection → `connectors` → `opc-servers`). `write_status` keeps the worst outcome (`ok` < `error` < `session_bad`); the gateway is the only one that updates it (the Laborious inserts the row as `pending` and never records OPC outcomes). The NATS first connect is bounded (it runs on the main loop); reconnection after it is unbounded.
+
 **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.
 
 **Error handling convention**: broad `except Exception` + `traceback.print_exc()` + notification-handler alert is intentional throughout (see `ruff.lint.ignore` for `BLE001` in `pyproject.toml`) — this is an always-on service where a single tag/server failure must not kill the pod; failures are surfaced via notifications/metrics instead of raised.

README.md

@@ -56,7 +56,8 @@ 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`).
+  (`ingestor/adapters/nats_consumer.py`, `ingestor/engines/write.py`). See
+  [Write commands](#write-commands-c2-and-write_status-c3) below.
 - **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`)
@@ -70,6 +71,37 @@ 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.
 
+### Write commands (C2) and `write_status` (C3)
+
+The Laborious publishes one command per tag (SIENTIAPDE-1975):
+
+```json
+{"server_id": "1", "addr": "ns=2;s=Pred", "value": 1.25, "data_type": "float",
+ "model_id": "2", "model_name": "sin-approx", "schema": "sientia_data",
+ "table_name": "predictions", "timestamp": "2026-10-13T12:00:00-03:00",
+ "write_timestamp": false, "write_quality": false}
+```
+
+- **Server connection**: the command carries only `server_id`. The connection is
+  resolved by `ServerConfigResolver` (`ingestor/connection/servers.py`): `url`/
+  `server_uri`/`use_security` in the command when present (override), else the
+  live pooled connection opened by the reads, else the `connectors` document
+  (`type: opc_ua`, `status: Active`), else the `opc-servers` document
+  (`use_security = bool(cert_path)`) — the orchestrator's rule for the read slots.
+  Lookups are cached for 60 s. An unknown server is a failed write (`server_not_found`).
+- **Outcome**: `UPDATE ... SET write_status` on the prediction row (by `timestamp` +
+  `model_id`), keeping the worst outcome: `ok` < `error` < `session_bad`, and
+  `pending`/NULL are replaced by any answer, so the order of the answers does not
+  matter. The gateway is the only one that updates the column: the Laborious inserts the
+  row as `pending` (merged with its PI Web API outcome) and never records OPC outcomes.
+  The UPDATE never raises; its outcome (`updated`, `kept`, `row_not_found`,
+  `invalid`, `error`) is logged and counted in `opc_write_status_update_total`.
+  The reason of a failed write is logged by `WriteEngine`.
+- **NATS connection**: the first connection gives up after 5 s (it runs on the main
+  loop: nats-py's defaults would block it for minutes while NATS is down); once
+  connected, it reconnects forever. A connection closed for good is started again
+  by `GatewayRuntime.idle_sweep`.
+
 ## 📋 Prerequisites
 
 - Python 3.11+
@@ -290,6 +322,9 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring:
 ### OPC Gateway Metrics
 - `opc_write_total`: Active write attempts (NATS-driven), by result
 - `opc_write_latency_seconds`: Active write latency
+- `opc_write_status_by_model`: Last write outcome per model (1 = ok, 0 = failed; label `status`)
+- `opc_write_status_update_total`: `write_status` UPDATEs, by result
+  (`updated`, `kept`, `row_not_found`, `invalid`, `error`)
 - `opc_active_read_total`: Scheduled/on-demand read attempts, by result
 - `opc_error_by_type_total`: OPC errors by type (connection, conversion, session, ...)
 

ingestor/adapters/nats_consumer.py

@@ -3,12 +3,14 @@
 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.
+here. Once connected, nats-py reconnects by itself, forever.
 """
 
+import asyncio
 import json
+import math
+import traceback
 
-import nats
 from nats.aio.client import Client
 from nats.aio.subscription import Subscription
 from sientia_do.observability.logger import Logger
@@ -24,18 +26,55 @@ class NatsWriteConsumer:
         logger: Logger,
         subject: str = 'opc.write',
         queue: str = 'gateway-writers',
+        connect_timeout_s: float = 5.0,
     ):
         self.nats_url = nats_url
         self.write_engine = write_engine
         self.logger = logger
         self.subject = subject
         self.queue = queue
+        self.connect_timeout_s = connect_timeout_s
         self.nc: Client | None = None
         self.sub: Subscription | None = None
 
+    @property
+    def closed(self) -> bool:
+        """True when there is no connection, or it was closed for good."""
+        return self.nc is None or self.nc.is_closed
+
     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)
+        """Connect and subscribe, giving up after ``connect_timeout_s``.
+
+        It runs from the main loop (startup and ``GatewayRuntime.idle_sweep``), so it
+        must never block it: nats-py's defaults retry the first connection 60 times,
+        about 2 minutes, longer than the heartbeat TTL, and with
+        ``max_reconnect_attempts=-1`` they would retry it forever. Once connected,
+        reconnection is unbounded, so an outage never closes the connection for good.
+
+        Raises:
+            Exception: When the connection or the subscription failed; nothing is left
+                open in that case.
+        """
+        nc = Client()
+        try:
+            await asyncio.wait_for(
+                nc.connect(
+                    servers=self.nats_url,
+                    connect_timeout=math.ceil(self.connect_timeout_s),
+                    max_reconnect_attempts=-1,
+                    name='opc-gateway',
+                    # One log line per event instead of nats-py's traceback per attempt.
+                    error_cb=self._on_error,
+                    disconnected_cb=self._on_disconnected,
+                    reconnected_cb=self._on_reconnected,
+                ),
+                timeout=self.connect_timeout_s,
+            )
+            self.sub = await nc.subscribe(self.subject, queue=self.queue, cb=self._handle)
+        except (Exception, asyncio.CancelledError):
+            await self._discard(nc)
+            raise
+        self.nc = nc
         self.logger.info(f'NatsWriteConsumer started: subject={self.subject} queue={self.queue}')
 
     async def _handle(self, msg) -> None:
@@ -45,11 +84,41 @@ class NatsWriteConsumer:
             preview = msg.data[:200]
             self.logger.error(f'NatsWriteConsumer dropped malformed JSON ({e}): {preview!r}')
             return
-        await self.write_engine.write(payload)
+        try:
+            await self.write_engine.write(payload)
+        except Exception as e:
+            # The engine is raise-safe; this only keeps an unexpected bug from going to
+            # nats-py's error callback without context.
+            self.logger.error(
+                f'NatsWriteConsumer: unexpected error handling a write command ({e}):\n'
+                f'{traceback.format_exc()}'
+            )
+
+    # These stay async with nothing to await: nats-py rejects callbacks that are not
+    # coroutine functions (InvalidCallbackTypeError on connect).
+    async def _on_error(self, e: Exception) -> None:
+        self.logger.warning(f'NATS error: {type(e).__name__} {e}')
+
+    async def _on_disconnected(self) -> None:
+        self.logger.warning('Disconnected from NATS; reconnecting')
+
+    async def _on_reconnected(self) -> None:
+        self.logger.info('Reconnected to NATS')
+
+    async def _discard(self, nc: Client) -> None:
+        try:
+            await nc.close()
+        except Exception as e:
+            self.logger.warning(f'Error closing a failed NATS connection: {e}')
 
     async def stop(self) -> None:
         if self.nc is None:
             return
+        if self.nc.is_closed:
+            # Closed for good already: unsubscribe/drain would raise ConnectionClosedError
+            # and abort the shutdown sequence.
+            self.nc = None
+            return
         if self.sub is not None:
             await self.sub.unsubscribe()
         await self.nc.drain()

ingestor/connection/pool.py

@@ -76,6 +76,16 @@ class OpcConnectionPool(SientiaMonitoring):
                 f'from the settings it was opened with'
             )
 
+    def config_for(self, server_id: str) -> dict | None:
+        """Connection settings of the pooled client for `server_id`, or None when the
+        server has no live pooled connection. Lets a writer reuse exactly the settings
+        the read path opened the connection with."""
+        server_id = str(server_id)
+        if server_id not in self._connections:
+            return None
+        cfg = self._configs.get(server_id)
+        return dict(cfg) if cfg is not None else None
+
     def _cached(self, server_id: str) -> Client | None:
         entry = self._connections.get(server_id)
         if entry is None:

ingestor/connection/servers.py

@@ -0,0 +1,120 @@
+"""Connection settings of the OPC server a write command targets (SIENTIAPDE-1975 GW1).
+
+The Laborious only knows the `server_id` of a write tag (pipeline `write_tags`); the
+connection (url, server URI, security) belongs to the gateway, which owns every OPC
+session. It is resolved with the same rule the orchestrator uses to build the read slots
+(SIENTIAPDE-1974): the tag database `connectors` first, the legacy `opc-servers` otherwise.
+"""
+
+from typing import Any
+
+from sientia_do.observability.logger import Logger
+
+from ingestor.connection.clock import MonotonicClock
+from ingestor.connection.pool import OpcConnectionPool
+
+CONNECTORS = 'connectors'
+OPC_SERVERS = 'opc-servers'
+
+
+def _id_forms(server_id: str) -> list[Any]:
+    """`server_id` as stored: the tag database keeps ints, opc-servers strings or ints."""
+    forms: list[Any] = [server_id]
+    if server_id.isdigit():
+        forms.append(int(server_id))
+    return forms
+
+
+def _connector_cfg(connector: dict[str, Any]) -> dict[str, Any] | None:
+    if not connector.get('url'):
+        return None
+    return {
+        'url': connector['url'],
+        'server_uri': connector.get('server_uri'),
+        'use_security': bool(connector.get('use_security', False)),
+    }
+
+
+def _opc_server_cfg(server: dict[str, Any]) -> dict[str, Any] | None:
+    if not server.get('url'):
+        return None
+    return {
+        'url': server['url'],
+        'server_uri': server.get('uri'),
+        # The certificate comes from MinIO now: a cert_path only says the server uses
+        # security, as in the orchestrator's slot (SIENTIAPDE-1974 RF4).
+        'use_security': bool(server.get('cert_path')),
+    }
+
+
+class ServerConfigResolver:
+    """Resolve and cache the connection settings of an OPC server by `server_id`.
+
+    Order: settings carried by the command itself (an explicit override; the C2
+    command of the Laborious carries none), the live pooled connection (what the read
+    path opened, so writes never fight the reads over settings), then MongoDB. Lookups
+    are cached for `ttl_s`; a server that is not found, for `missing_ttl_s`, so a
+    pipeline pointing at a removed server does not query MongoDB on every command."""
+
+    def __init__(
+        self,
+        pool: OpcConnectionPool,
+        mongo,
+        logger: Logger,
+        ttl_s: float = 60.0,
+        missing_ttl_s: float = 10.0,
+        metadata: dict | None = None,
+    ):
+        self.pool = pool
+        self.mongo = mongo
+        self.logger = logger
+        self.ttl_s = ttl_s
+        self.missing_ttl_s = missing_ttl_s
+        self.metadata = metadata or {}
+        self.clock = MonotonicClock()
+        self._cache: dict[str, tuple[dict[str, Any] | None, float]] = {}
+
+    async def resolve(self, server_id: str, cmd: dict) -> dict[str, Any] | None:
+        """Connection settings for `server_id`, or None when no source knows it.
+
+        Raises:
+            Exception: When MongoDB cannot be read (the caller reports the write as
+                failed).
+        """
+        if cmd.get('url'):
+            return {
+                'url': cmd['url'],
+                'server_uri': cmd.get('server_uri'),
+                'use_security': cmd.get('use_security', False),
+            }
+        server_id = str(server_id)
+        pooled = self.pool.config_for(server_id)
+        if pooled is not None:
+            return pooled
+        now = self.clock.elapsed()
+        cached = self._cache.get(server_id)
+        if cached is not None:
+            cfg, cached_at = cached
+            if now - cached_at < (self.ttl_s if cfg is not None else self.missing_ttl_s):
+                return dict(cfg) if cfg is not None else None
+        cfg = await self._lookup(server_id)
+        self._cache[server_id] = (cfg, now)
+        return dict(cfg) if cfg is not None else None
+
+    async def _lookup(self, server_id: str) -> dict[str, Any] | None:
+        ids = {'$in': _id_forms(server_id)}
+        connector = await self.mongo.find_one(
+            CONNECTORS,
+            {'connector_id': ids, 'type': 'opc_ua', 'status': 'Active'},
+            self.metadata,
+        )
+        cfg = _connector_cfg(connector) if connector else None
+        if cfg is not None:
+            return cfg
+        server = await self.mongo.find_one(OPC_SERVERS, {'id': ids}, self.metadata)
+        cfg = _opc_server_cfg(server) if server else None
+        if cfg is None:
+            self.logger.warning(
+                f'OPC server {server_id} not found in {CONNECTORS} nor {OPC_SERVERS}'
+            )
+        return cfg

ingestor/engines/write.py

@@ -18,6 +18,7 @@ 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.connection.servers import ServerConfigResolver
 from ingestor.converters import ConversionError, to_typed_value
 from ingestor.io.data_sink import DataSink
 
@@ -77,9 +78,13 @@ class WriteEngine(SientiaMonitoring):
         write_timestamp: bool = True,
         write_quality: bool = True,
         min_pause_s: float = 0.0,
+        servers: ServerConfigResolver | None = None,
     ):
         self.pool = pool
         self.sink = sink
+        # Resolves the server connection by server_id (SIENTIAPDE-1975 GW1). Without it,
+        # only the connection carried by the command itself is used.
+        self.servers = servers
         self.write_timestamp = write_timestamp
         self.write_quality = write_quality
         self.min_pause_s = min_pause_s
@@ -105,11 +110,24 @@ class WriteEngine(SientiaMonitoring):
         except ConversionError as e:
             return await self._emit_error(cmd, 'conversion', str(e), 'error')
 
-        dv = self._build_datavalue(cmd, typed)
+        try:
+            dv = self._build_datavalue(cmd, typed)
+        except (KeyError, TypeError, ValueError) as e:
+            # write_timestamp on with a missing or non-ISO timestamp.
+            return await self._emit_error(cmd, 'bad_timestamp', repr(e), 'error')
+
+        try:
+            server_cfg = await self._server_cfg(cmd)
+        except Exception as e:
+            return await self._emit_error(cmd, 'server_lookup', str(e), 'error')
+        if server_cfg is None:
+            return await self._emit_error(
+                cmd, 'server_not_found', f'OPC server {cmd["server_id"]} is not configured', 'error'
+            )
 
         start = self.clock.elapsed()
         try:
-            client = await self.pool.get(cmd['server_id'], self._server_cfg(cmd))
+            client = await self.pool.get(cmd['server_id'], server_cfg)
             node = client.get_node(cmd['addr'])
             await node.write_value(dv)
         except Exception as e:
@@ -127,10 +145,14 @@ class WriteEngine(SientiaMonitoring):
                 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.
+    async def _server_cfg(self, cmd: dict) -> dict | None:
+        """Connection settings of the target server: resolved by server_id (connection
+        carried by the command, live pooled connection, connectors, opc-servers), or only
+        the ones carried by the command when no resolver is wired."""
+        if self.servers is not None:
+            return await self.servers.resolve(str(cmd['server_id']), cmd)
+        if not cmd.get('url'):
+            return None
         return {
             'url': cmd.get('url'),
             'use_security': cmd.get('use_security', False),
@@ -156,6 +178,12 @@ class WriteEngine(SientiaMonitoring):
         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):
+        # The reason only lives here: the row gets write_status alone (SIENTIAPDE-1975).
+        self.logger.error(
+            f'OPC write failed ({status}, {error_type}): server={cmd.get("server_id")} '
+            f'addr={cmd.get("addr")} model_id={cmd.get("model_id")} '
+            f'timestamp={cmd.get("timestamp")}: {message}'
+        )
         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}
@@ -164,6 +192,17 @@ class WriteEngine(SientiaMonitoring):
             metric_object=metrics.OPC_WRITE_TOTAL,
             tags={**tags, 'model_id': cmd.get('model_id', 'unknown'), 'result': 'error'},
         )
+        # Same series as a success (status='ok'), so it always shows the last outcome.
+        await self.emit_metric(
+            metric_object=metrics.OPC_WRITE_STATUS_BY_MODEL,
+            method='set',
+            value=0,
+            tags={
+                'pod_id': self.pod_id,
+                'model_id': cmd.get('model_id', 'unknown'),
+                'status': 'ok',
+            },
+        )
         result = WriteResult(ok=False, status=status, opc_status_code=error_type, message=message)
         await self.sink.update_write_status(cmd, result)
         return result

ingestor/gateway_runtime.py

@@ -22,6 +22,7 @@ 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.connection.servers import ServerConfigResolver
 from ingestor.engines.active_read import ActiveReadEngine
 from ingestor.engines.write import WriteEngine
 from ingestor.io.data_sink import DataSink
@@ -61,6 +62,12 @@ class GatewayRuntime(SientiaMonitoring):
         )
         self.sink = DataSink(data_manager, postgres)
 
+        # Write commands carry only the server_id (SIENTIAPDE-1975): the connection is
+        # resolved here, from the live pooled connection or the connectors/opc-servers
+        # documents, with the same rule the orchestrator uses for the read slots.
+        self.servers = ServerConfigResolver(
+            self.pool, data_manager.mongo_repository, logger, metadata=self.metadata
+        )
         self.write_engine = WriteEngine(
             self.pool,
             self.sink,
@@ -70,6 +77,7 @@ class GatewayRuntime(SientiaMonitoring):
             write_timestamp=config.write_timestamp,
             write_quality=config.write_quality,
             min_pause_s=config.write_min_pause_s,
+            servers=self.servers,
         )
         self.nats_consumer = NatsWriteConsumer(
             config.nats_url,
@@ -110,6 +118,11 @@ class GatewayRuntime(SientiaMonitoring):
 
     async def idle_sweep(self) -> None:
         await self.pool.release_idle(self.config.idle_conn_max_s)
+        if self._nats_started and self.nats_consumer.closed:
+            # Reconnection is unbounded once connected; a closed connection means it
+            # was closed for good (e.g. by the server), so subscribe again.
+            self.logger.warning('GatewayRuntime: NATS connection closed; starting again')
+            self._nats_started = False
         if not self._nats_started:
             await self._start_nats()
 

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

@@ -1,6 +1,7 @@
+import asyncio
 import re
 
-from sqlalchemy import text
+from sqlalchemy import Select, Update, case, column, select, table, update
 
 import ingestor.metrics as metrics
 
@@ -8,6 +9,34 @@ import ingestor.metrics as metrics
 # match before a trailing '\n', which is not what we want for a SQL identifier.
 _IDENTIFIER_RE = re.compile(r'\w+', re.ASCII)
 
+# Worst-wins precedence of predictions.write_status (SIENTIAPDE-1975), from best to worst;
+# NULL, 'pending' and unknown values rank 0. The gateway is the only writer that updates the
+# column: the Laborious only sets the value of the rows it inserts ('pending', PI outcome).
+STATUS_RANK = {'ok': 1, 'error': 2, 'session_bad': 3}
+
+
+def write_status_statements(
+    schema: str, table_name: str, status: str, timestamp, model_id
+) -> tuple[Update, Select]:
+    """The UPDATE that records `status` unless the row has a worse one, and the SELECT that
+    tells whether the row exists. Built with SQLAlchemy Core: identifiers and values are
+    never formatted into the SQL text."""
+    predictions = table(
+        table_name,
+        column('write_status'),
+        column('timestamp'),
+        column('model_id'),
+        schema=schema,
+    )
+    row = (predictions.c.timestamp == timestamp) & (predictions.c.model_id == model_id)
+    current_rank = case(STATUS_RANK, value=predictions.c.write_status, else_=0)
+    return (
+        update(predictions)
+        .where(row, current_rank <= STATUS_RANK.get(status, 0))
+        .values(write_status=status),
+        select(1).select_from(predictions).where(row),
+    )
+
 
 class DataSink:
     """Gateway egress wrapper: fans a reading out to every configured name/collection."""
@@ -43,33 +72,72 @@ class DataSink:
                 },
             )
 
-    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.
+    async def update_write_status(self, cmd: dict, result) -> str:
+        """Record a write outcome on the prediction row (C3), keeping the worst one.
 
-        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.
+        The row is found by timestamp + model_id. A prediction is written to several
+        tags, each answered by its own command, so a better outcome never replaces a
+        worse one: ``ok`` < ``error`` < ``session_bad``, and ``pending``/NULL are
+        replaced by any answer, so the order of the answers does not matter. Only the
+        gateway updates the column (SIENTIAPDE-1975): the Laborious publishes the commands
+        and sets ``pending`` (merged with its PI Web API outcome) on the row it inserts.
+
+        Never raises: a failure here must not reach the NATS callback nor hide the OPC
+        write itself, which already happened. The outcome is logged and counted instead.
+
+        Return:
+            str: ``updated``, ``kept`` (the row already has a worse outcome),
+            ``row_not_found`` (e.g. the Laborious INSERT failed), ``invalid`` (bad or
+            missing schema/table/timestamp/model_id) or ``error``.
         """
-        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'
+        try:
+            schema = self._validate_identifier(cmd.get('schema'))
+            table = self._validate_identifier(cmd.get('table_name'))
+            if cmd.get('timestamp') is None or cmd.get('model_id') is None:
+                raise ValueError('missing timestamp or model_id')
+        except ValueError as e:
+            return await self._status_outcome('invalid', cmd, f'write_status not recorded: {e}')
+
+        statements = write_status_statements(
+            schema, table, result.status, cmd['timestamp'], cmd['model_id']
         )
-        params = {
-            'status': result.status,
-            'timestamp': cmd['timestamp'],
-            'model_id': cmd['model_id'],
-        }
-        # ponytail: sync engine call inside async def — same pattern sientia_do's own
-        # Postgres activities use (e.g. export_data_to_postgres); no async pg driver here.
+        try:
+            # The engine is synchronous (no async pg driver here): run it off the event
+            # loop, which also serves the subscriptions of the read path.
+            outcome = await asyncio.to_thread(self._update_sync, *statements)
+        except Exception as e:
+            return await self._status_outcome('error', cmd, f'write_status UPDATE failed: {e}')
+        detail = {
+            'updated': f'write_status={result.status} recorded',
+            'kept': f'write_status={result.status} not recorded: the row has a worse outcome',
+            'row_not_found': f'write_status={result.status} not recorded: no prediction row',
+        }[outcome]
+        return await self._status_outcome(outcome, cmd, detail)
+
+    def _update_sync(self, update_stmt: Update, exists_stmt: Select) -> str:
         with self.postgres.engine.begin() as connection:
-            connection.execute(stmt, params)
+            if connection.execute(update_stmt).rowcount:
+                return 'updated'
+            row = connection.execute(exists_stmt).first()
+        return 'kept' if row is not None else 'row_not_found'
+
+    async def _status_outcome(self, outcome: str, cmd: dict, detail: str) -> str:
+        message = (
+            f'{detail} (server={cmd.get("server_id")} addr={cmd.get("addr")} '
+            f'model_id={cmd.get("model_id")} timestamp={cmd.get("timestamp")})'
+        )
+        if outcome in ('updated', 'kept'):
+            self.data_manager.logger.debug(message)
+        else:
+            self.data_manager.logger.warning(message)
+        await self.emit_metric(
+            metric_object=metrics.OPC_WRITE_STATUS_UPDATE_TOTAL,
+            tags={'pod_id': self.data_manager.pod_id, 'result': outcome},
+        )
+        return outcome
 
     @staticmethod
-    def _validate_identifier(name: str) -> str:
+    def _validate_identifier(name: object) -> str:
         if not isinstance(name, str) or not _IDENTIFIER_RE.fullmatch(name):
             raise ValueError(f'invalid SQL identifier: {name!r}')
         return name

ingestor/managers/data_manager.py

@@ -12,6 +12,22 @@ from sientia_do.temporal.constants import now
 import ingestor.metrics as metrics
 
 
+def redact_uri(uri: str) -> str:
+    """
+    Returns ``uri`` without the user and password, safe to log
+    (``mongodb://user:pass@host:27017/db`` -> ``mongodb://***@host:27017/db``).
+    """
+    prefix, sep, rest = uri.partition('://')
+    if not sep:
+        prefix, rest = '', uri
+    ends = [i for i in (rest.find('/'), rest.find('?')) if i != -1]
+    end = min(ends) if ends else len(rest)
+    authority, tail = rest[:end], rest[end:]
+    if '@' not in authority:
+        return uri
+    return f'{prefix}{sep}***@{authority.rpartition("@")[2]}{tail}'
+
+
 class DataManager(SientiaMonitoring):
     """
     Manages data persistence operations for the OPC Ingestor.
@@ -66,7 +82,8 @@ class DataManager(SientiaMonitoring):
         )
 
         logger.info(
-            f'Trying to initializing DataManager with MongoDB servers: {mongo_connection_string}'
+            'Trying to initializing DataManager with MongoDB servers: '
+            f'{redact_uri(mongo_connection_string)}'
         )
 
         self.connection_string = mongo_connection_string
@@ -82,7 +99,9 @@ class DataManager(SientiaMonitoring):
 
         self.metadata = metadata
 
-        logger.info(f'DataManager initialized with MongoDB servers: {self.connection_string}')
+        logger.info(
+            f'DataManager initialized with MongoDB servers: {redact_uri(self.connection_string)}'
+        )
 
     def shutdown(self):
         """

ingestor/managers/ingestor_manager.py

@@ -538,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):
         """
@@ -567,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.
@@ -600,8 +658,9 @@ class IngestorManager(SientiaMonitoring):
                 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:
         """
@@ -764,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/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'],
 )
 
@@ -156,6 +156,12 @@ OPC_WRITE_STATUS_BY_MODEL = Gauge(
     'Last write status per model (1=ok,0=error)',
     ['pod_id', 'model_id', 'status'],
 )
+OPC_WRITE_STATUS_UPDATE_TOTAL = Counter(
+    'opc_write_status_update_total',
+    'write_status UPDATEs on the prediction rows',
+    # result: updated|kept|row_not_found|invalid|error
+    ['pod_id', 'result'],
+)
 # --- Gateway: Active read metrics ---
 OPC_ACTIVE_READ_TOTAL = Counter(
     'opc_active_read_total',

pyproject.toml

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

requirements.txt

@@ -1,9 +1,8 @@
 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
+# 1.12.5 builds the Postgres engine with an explicit postgresql+psycopg2:// URL: SQLAlchemy 2.1
+# defaults postgresql:// to psycopg (v3), which is not installed ("No module named 'psycopg'").
+sientia_do==1.12.5
 prometheus_client
 pymongo
 nats-py>=2.6

tests/unit/adapters/test_nats_consumer.py

@@ -1,3 +1,5 @@
+import asyncio
+import inspect
 import json
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -37,20 +39,99 @@ async def test_handle_invalid_json_drops_message_and_logs_error():
     consumer.logger.error.assert_called_once()
 
 
+CLIENT = 'ingestor.adapters.nats_consumer.Client'
+
+
+def _fake_nc(connect=None):
+    nc = MagicMock()
+    nc.connect = connect or AsyncMock()
+    nc.subscribe = AsyncMock()
+    nc.close = AsyncMock()
+    nc.drain = AsyncMock()
+    nc.is_closed = False
+    return nc
+
+
+async def _hang(*_args, **_kwargs):
+    await asyncio.sleep(10)
+
+
 @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()
+    nc = _fake_nc()
 
-    with patch('ingestor.adapters.nats_consumer.nats') as mock_nats:
-        mock_nats.connect = AsyncMock(return_value=mock_nc)
+    with patch(CLIENT, return_value=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
-    )
+    kwargs = nc.connect.await_args.kwargs
+    assert kwargs['servers'] == 'nats://localhost:4222'
+    # Reconnection is unbounded once connected: an outage never closes it for good.
+    assert kwargs['max_reconnect_attempts'] == -1
+    nc.subscribe.assert_awaited_once_with('opc.write', queue='gateway-writers', cb=consumer._handle)
+    assert consumer.closed is False
+
+
+@pytest.mark.asyncio
+async def test_start_gives_up_after_the_connect_timeout():
+    """start() runs on the main loop: an unreachable NATS must not block it."""
+    consumer = make_consumer(connect_timeout_s=0.1)
+    nc = _fake_nc(connect=AsyncMock(side_effect=_hang))
+    # Outer guard: fails the test instead of hanging it if start() ignores its timeout.
+    start = asyncio.wait_for(consumer.start(), timeout=5)
+
+    with patch(CLIENT, return_value=nc):
+        with pytest.raises(TimeoutError):
+            await start
+
+    nc.close.assert_awaited_once()
+    assert consumer.closed is True
+
+
+@pytest.mark.asyncio
+async def test_start_closes_the_connection_when_subscribe_fails():
+    consumer = make_consumer()
+    nc = _fake_nc()
+    nc.subscribe.side_effect = RuntimeError('permissions violation')
+
+    with patch(CLIENT, return_value=nc), pytest.raises(RuntimeError):
+        await consumer.start()
+
+    nc.close.assert_awaited_once()
+    assert consumer.closed is True
+
+
+@pytest.mark.asyncio
+async def test_handle_unexpected_engine_error_is_logged_not_raised():
+    consumer = make_consumer()
+    consumer.write_engine.write.side_effect = RuntimeError('bug')
+    msg = MagicMock(data=json.dumps({'server_id': '1'}).encode())
+
+    await consumer._handle(msg)
+
+    assert 'bug' in consumer.logger.error.call_args.args[0]
+
+
+@pytest.mark.asyncio
+async def test_connection_events_are_logged_in_one_line():
+    consumer = make_consumer()
+    nc = _fake_nc()
+    with patch(CLIENT, return_value=nc):
+        await consumer.start()
+    callbacks = nc.connect.await_args.kwargs
+    # nats-py raises InvalidCallbackTypeError on connect for a non-coroutine callback.
+    for name in ('error_cb', 'disconnected_cb', 'reconnected_cb'):
+        assert inspect.iscoroutinefunction(callbacks[name]), name
+
+    await callbacks['error_cb'](OSError('connection refused'))
+    await callbacks['disconnected_cb']()
+    await callbacks['reconnected_cb']()
+
+    warnings = [c.args[0] for c in consumer.logger.warning.call_args_list]
+    assert warnings == [
+        'NATS error: OSError connection refused',
+        'Disconnected from NATS; reconnecting',
+    ]
 
 
 @pytest.mark.asyncio
@@ -63,14 +144,28 @@ async def test_stop_before_start_does_not_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)
+    nc = _fake_nc()
+    sub = AsyncMock()
+    nc.subscribe = AsyncMock(return_value=sub)
 
-    with patch('ingestor.adapters.nats_consumer.nats') as mock_nats:
-        mock_nats.connect = AsyncMock(return_value=mock_nc)
+    with patch(CLIENT, return_value=nc):
         await consumer.start()
         await consumer.stop()
 
-    mock_sub.unsubscribe.assert_awaited_once()
-    mock_nc.drain.assert_awaited_once()
+    sub.unsubscribe.assert_awaited_once()
+    nc.drain.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_stop_on_a_connection_closed_for_good_does_not_raise():
+    consumer = make_consumer()
+    nc = _fake_nc()
+    with patch(CLIENT, return_value=nc):
+        await consumer.start()
+    nc.is_closed = True
+    consumer.sub.unsubscribe = AsyncMock(side_effect=RuntimeError('connection closed'))
+
+    await consumer.stop()  # no raise
+
+    nc.drain.assert_not_awaited()
+    assert consumer.closed is True

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_pool.py

@@ -342,3 +342,23 @@ async def test_connect_server_uri_still_wins_for_application_uri():
     # must match the certificate SAN URI (BadCertificateUriInvalid otherwise)
     assert fake_client.application_uri == 'urn:sientia:opc-ingestor'
     assert fake_client.name == pool.pod_id
+
+
+@pytest.mark.asyncio
+async def test_config_for_returns_a_copy_of_the_live_connection_settings():
+    pool = make_pool()
+    cfg = {'url': 'opc.tcp://h:4840', 'server_uri': 'urn:h', 'use_security': False}
+    with patch('ingestor.connection.pool.Client', return_value=AsyncMock()):
+        await pool.get(1, cfg)
+
+    found = pool.config_for('1')
+    assert found == cfg
+    found['url'] = 'mutated'
+    assert pool.config_for(1)['url'] == 'opc.tcp://h:4840'
+
+    await pool.evict('1')
+    assert pool.config_for('1') is None
+
+
+def test_config_for_unknown_server_is_none():
+    assert make_pool().config_for('missing') is None

tests/unit/connection/test_servers.py

@@ -0,0 +1,151 @@
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from ingestor.connection.servers import CONNECTORS, OPC_SERVERS, ServerConfigResolver
+
+CONNECTOR = {
+    'connector_id': 1,
+    'type': 'opc_ua',
+    'status': 'Active',
+    'name': 'kepserver',
+    'url': 'opc.tcp://kepserver:49320',
+    'server_uri': 'urn:kepserver',
+    'use_security': True,
+}
+OPC_SERVER = {
+    'id': '1',
+    'server_name': 'kepserver',
+    'url': 'opc.tcp://legacy:49320',
+    'uri': 'urn:legacy',
+    'cert_path': '/app/certs/cert.pem',
+}
+
+
+def make_resolver(connector=None, opc_server=None, pooled=None, ttl_s=60.0):
+    pool = MagicMock()
+    pool.config_for = MagicMock(return_value=pooled)
+    mongo = MagicMock()
+
+    async def find_one(collection, _filters, _metadata):
+        return {CONNECTORS: connector, OPC_SERVERS: opc_server}[collection]
+
+    mongo.find_one = AsyncMock(side_effect=find_one)
+    return ServerConfigResolver(pool, mongo, MagicMock(), ttl_s=ttl_s)
+
+
+@pytest.mark.asyncio
+async def test_command_connection_overrides_every_other_source():
+    resolver = make_resolver(connector=CONNECTOR)
+
+    cfg = await resolver.resolve('1', {'url': 'opc.tcp://h:1', 'server_uri': 'urn:h'})
+
+    assert cfg == {'url': 'opc.tcp://h:1', 'server_uri': 'urn:h', 'use_security': False}
+    resolver.mongo.find_one.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_live_pooled_connection_wins_over_mongo():
+    """Writes reuse exactly the settings the read path opened the connection with."""
+    pooled = {'url': 'opc.tcp://read:1', 'server_uri': 'urn:read', 'use_security': True}
+    resolver = make_resolver(connector=CONNECTOR, pooled=pooled)
+
+    assert await resolver.resolve('1', {}) == pooled
+    resolver.pool.config_for.assert_called_once_with('1')
+    resolver.mongo.find_one.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_tag_database_connector_first():
+    resolver = make_resolver(connector=CONNECTOR, opc_server=OPC_SERVER)
+
+    cfg = await resolver.resolve('1', {})
+
+    assert cfg == {
+        'url': 'opc.tcp://kepserver:49320',
+        'server_uri': 'urn:kepserver',
+        'use_security': True,
+    }
+    collection, filters, _ = resolver.mongo.find_one.await_args_list[0].args
+    assert collection == CONNECTORS
+    # connector_id is an int in the tag database; the command carries a string.
+    assert filters == {'connector_id': {'$in': ['1', 1]}, 'type': 'opc_ua', 'status': 'Active'}
+
+
+@pytest.mark.asyncio
+async def test_opc_servers_fallback_without_tag_database():
+    resolver = make_resolver(connector=None, opc_server=OPC_SERVER)
+
+    cfg = await resolver.resolve('1', {})
+
+    # Same rule as the orchestrator's read slot: a cert_path means security.
+    assert cfg == {
+        'url': 'opc.tcp://legacy:49320',
+        'server_uri': 'urn:legacy',
+        'use_security': True,
+    }
+    collection, filters, _ = resolver.mongo.find_one.await_args_list[1].args
+    assert collection == OPC_SERVERS
+    assert filters == {'id': {'$in': ['1', 1]}}
+
+
+@pytest.mark.asyncio
+async def test_opc_server_without_cert_path_is_insecure():
+    resolver = make_resolver(opc_server={**OPC_SERVER, 'cert_path': ''})
+
+    assert (await resolver.resolve('1', {}))['use_security'] is False
+
+
+@pytest.mark.asyncio
+async def test_non_numeric_server_id_is_looked_up_as_is():
+    resolver = make_resolver(opc_server={**OPC_SERVER, 'id': 'kep-a'})
+
+    await resolver.resolve('kep-a', {})
+
+    _, filters, _ = resolver.mongo.find_one.await_args_list[0].args
+    assert filters['connector_id'] == {'$in': ['kep-a']}
+
+
+@pytest.mark.asyncio
+async def test_unknown_server_is_cached_briefly():
+    resolver = make_resolver(ttl_s=60.0)
+    resolver.missing_ttl_s = 10.0
+    resolver.clock = MagicMock()
+    resolver.clock.elapsed.side_effect = [0.0, 5.0, 11.0]
+
+    assert await resolver.resolve('9', {}) is None
+    assert await resolver.resolve('9', {}) is None  # within missing_ttl_s: no lookup
+    assert await resolver.resolve('9', {}) is None  # expired: looked up again
+    assert resolver.mongo.find_one.await_count == 4  # 2 lookups (connectors + opc-servers)
+    assert resolver.logger.warning.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_documents_without_url_are_ignored():
+    resolver = make_resolver(connector={**CONNECTOR, 'url': ''}, opc_server=OPC_SERVER)
+
+    assert (await resolver.resolve('1', {}))['url'] == 'opc.tcp://legacy:49320'
+
+
+@pytest.mark.asyncio
+async def test_lookups_are_cached_for_the_ttl():
+    resolver = make_resolver(connector=CONNECTOR, ttl_s=60.0)
+    resolver.clock = MagicMock()
+    resolver.clock.elapsed.side_effect = [0.0, 30.0, 61.0]
+
+    first = await resolver.resolve('1', {})
+    first['url'] = 'mutated by the caller'
+    second = await resolver.resolve('1', {})  # within the TTL: cached, and a copy
+    await resolver.resolve('1', {})  # expired: looked up again
+
+    assert second['url'] == 'opc.tcp://kepserver:49320'
+    assert resolver.mongo.find_one.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_mongo_errors_propagate_to_the_caller():
+    resolver = make_resolver()
+    resolver.mongo.find_one = AsyncMock(side_effect=RuntimeError('mongo down'))
+
+    with pytest.raises(RuntimeError, match='mongo down'):
+        await resolver.resolve('1', {})

tests/unit/engines/test_write.py

@@ -7,8 +7,17 @@ from asyncua import ua
 import ingestor.metrics as metrics
 from ingestor.engines.write import WriteEngine
 
+SERVER_CFG = {'url': 'opc.tcp://kepserver:49320', 'server_uri': 'urn:kep', 'use_security': True}
+
+
+def make_resolver(cfg=SERVER_CFG):
+    resolver = MagicMock()
+    resolver.resolve = AsyncMock(return_value=cfg)
+    return resolver
+
 
 def make_engine(**kwargs):
+    kwargs.setdefault('servers', make_resolver())
     eng = WriteEngine(
         pool=AsyncMock(),
         sink=AsyncMock(),
@@ -257,3 +266,104 @@ async def test_no_rate_limit_when_min_pause_is_zero():
         await eng.write(base_cmd())
         await eng.write(base_cmd())
         sleep_mock.assert_not_awaited()
+
+
+# --- SIENTIAPDE-1975: server resolution (GW1) and error reporting (GW4) ---
+
+
+@pytest.mark.asyncio
+async def test_write_uses_the_resolved_server_connection():
+    """The C2 command of the Laborious carries only the server_id."""
+    eng = make_engine()
+    eng.pool.get.return_value = make_client()
+
+    await eng.write(base_cmd())
+
+    eng.servers.resolve.assert_awaited_once_with('srv1', base_cmd())
+    eng.pool.get.assert_awaited_once_with('srv1', SERVER_CFG)
+
+
+@pytest.mark.asyncio
+async def test_write_unknown_server_records_error_without_connecting():
+    eng = make_engine(servers=make_resolver(cfg=None))
+
+    result = await eng.write(base_cmd())
+
+    assert result.status == 'error'
+    assert result.opc_status_code == 'server_not_found'
+    eng.pool.get.assert_not_awaited()
+    eng.sink.update_write_status.assert_awaited_once_with(base_cmd(), result)
+
+
+@pytest.mark.asyncio
+async def test_write_server_lookup_failure_records_error():
+    resolver = make_resolver()
+    resolver.resolve.side_effect = RuntimeError('mongo down')
+    eng = make_engine(servers=resolver)
+
+    result = await eng.write(base_cmd())
+
+    assert result.status == 'error'
+    assert result.opc_status_code == 'server_lookup'
+    assert 'mongo down' in result.message
+    eng.pool.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_write_without_resolver_uses_the_command_connection():
+    eng = make_engine(servers=None)
+    eng.pool.get.return_value = make_client()
+
+    await eng.write(base_cmd(url='opc.tcp://h:4840', server_uri='urn:h', use_security=True))
+
+    eng.pool.get.assert_awaited_once_with(
+        'srv1', {'url': 'opc.tcp://h:4840', 'use_security': True, 'server_uri': 'urn:h'}
+    )
+
+
+@pytest.mark.asyncio
+async def test_write_without_resolver_or_url_is_server_not_found():
+    eng = make_engine(servers=None)
+
+    result = await eng.write(base_cmd())
+
+    assert result.opc_status_code == 'server_not_found'
+    eng.pool.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_write_error_logs_the_reason_and_sets_the_model_status():
+    eng = make_engine()
+    node = make_node()
+    node.write_value.side_effect = RuntimeError('BadWriteNotSupported')
+    eng.pool.get.return_value = make_client(node)
+
+    await eng.write(base_cmd())
+
+    logged = eng.logger.error.call_args.args[0]
+    assert 'RuntimeError' in logged and 'BadWriteNotSupported' in logged
+    assert 'addr=ns=2;i=5' in logged and 'model_id=m1' in logged
+    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
+    )
+    # Same series as a success, so it shows the last outcome.
+    assert status_call.kwargs['value'] == 0
+    assert status_call.kwargs['tags'] == {'pod_id': eng.pod_id, 'model_id': 'm1', 'status': 'ok'}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('timestamp', [None, 'not-a-timestamp'])
+async def test_write_bad_timestamp_records_error_without_raising(timestamp):
+    eng = make_engine(write_timestamp=True)
+    cmd = base_cmd(timestamp=timestamp)
+    if timestamp is None:
+        del cmd['timestamp']
+
+    result = await eng.write(cmd)
+
+    assert result.status == 'error'
+    assert result.opc_status_code == 'bad_timestamp'
+    eng.pool.get.assert_not_awaited()
+    eng.sink.update_write_status.assert_awaited_once()

tests/unit/io/test_data_sink_status.py

@@ -1,89 +1,149 @@
-from unittest.mock import MagicMock
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.dialects import postgresql
 
+import ingestor.metrics as metrics
 from ingestor.engines.write import WriteResult
-from ingestor.io.data_sink import DataSink
+from ingestor.io.data_sink import DataSink, write_status_statements
+
+CMD = {
+    'server_id': '1',
+    'addr': 'ns=2;s=Pred',
+    'schema': 'sientia_data',
+    'table_name': 'pred_m1',
+    'model_id': '1',
+    'timestamp': '2026-07-13T00:00:00Z',
+}
 
 
-def make_sink():
+def _result(status):
+    return WriteResult(ok=status == 'ok', status=status, opc_status_code='x', message='')
+
+
+def make_sink(rowcount=1, existing_row=True):
     postgres = MagicMock()
     connection = MagicMock()
+    connection.execute.return_value.rowcount = rowcount
+    connection.execute.return_value.first.return_value = (1,) if existing_row else None
     postgres.engine.begin.return_value.__enter__.return_value = connection
-    sink = DataSink(data_manager=MagicMock(), postgres=postgres)
+    data_manager = MagicMock()
+    data_manager.emit_metric = AsyncMock()
+    sink = DataSink(data_manager=data_manager, postgres=postgres)
     return sink, connection
 
 
+def _outcome_metric(sink):
+    calls = [
+        c
+        for c in sink.data_manager.emit_metric.await_args_list
+        if c.kwargs['metric_object'] is metrics.OPC_WRITE_STATUS_UPDATE_TOTAL
+    ]
+    assert len(calls) == 1
+    return calls[0].kwargs['tags']['result']
+
+
 @pytest.mark.asyncio
-async def test_update_write_status_targets_model_row():
+async def test_update_write_status_targets_model_row_with_precedence():
     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)
+    outcome = await sink.update_write_status(CMD, _result('ok'))
 
+    assert outcome == 'updated'
     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
+    (stmt,) = connection.execute.call_args[0]
+    compiled = stmt.compile(dialect=postgresql.dialect())
+    sql = str(compiled)
+    assert sql.startswith('UPDATE sientia_data.pred_m1 SET write_status=')
+    assert 'CASE sientia_data.pred_m1.write_status WHEN' 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
+    assert '2026-07-13T00:00:00Z' not in sql and "'ok'" not in sql
+    assert compiled.params['write_status'] == 'ok'
+    assert '2026-07-13T00:00:00Z' in compiled.params.values()
+    assert _outcome_metric(sink) == 'updated'
 
 
 @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')
+async def test_update_write_status_kept_when_the_row_has_a_worse_outcome():
+    sink, _ = make_sink(rowcount=0, existing_row=True)
 
-    with pytest.raises(ValueError):
-        await sink.update_write_status(cmd, result)
-
-    connection.execute.assert_not_called()
+    assert await sink.update_write_status(CMD, _result('ok')) == 'kept'
+    assert _outcome_metric(sink) == 'kept'
 
 
 @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='')
+async def test_update_write_status_row_not_found_is_logged():
+    """E.g. the Laborious INSERT failed and it wrote to OPC anyway (SIENTIAPDE-1975 D10)."""
+    sink, _ = make_sink(rowcount=0, existing_row=False)
 
-    with pytest.raises(ValueError):
-        await sink.update_write_status(cmd, result)
-
-    connection.execute.assert_not_called()
+    assert await sink.update_write_status(CMD, _result('error')) == 'row_not_found'
+    assert 'no prediction row' in sink.data_manager.logger.warning.call_args.args[0]
+    assert _outcome_metric(sink) == 'row_not_found'
 
 
 @pytest.mark.asyncio
-async def test_update_write_status_rejects_bad_table_name():
+async def test_update_write_status_database_error_is_not_raised():
     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')
+    connection.execute.side_effect = RuntimeError('column "write_status" does not exist')
 
-    with pytest.raises(ValueError):
-        await sink.update_write_status(cmd, result)
+    assert await sink.update_write_status(CMD, _result('ok')) == 'error'
+    assert 'write_status' in sink.data_manager.logger.warning.call_args.args[0]
+    assert _outcome_metric(sink) == 'error'
 
+
+@pytest.mark.parametrize(
+    ('field', 'value'),
+    [
+        ('schema', 'a; DROP TABLE x'),
+        ('schema', 'sientia_data\n'),
+        ('table_name', 'pred_m1; DROP TABLE x'),
+        ('schema', None),
+        ('timestamp', None),
+        ('model_id', None),
+    ],
+)
+@pytest.mark.asyncio
+async def test_update_write_status_rejects_invalid_commands(field, value):
+    sink, connection = make_sink()
+
+    outcome = await sink.update_write_status({**CMD, field: value}, _result('error'))
+
+    assert outcome == 'invalid'
     connection.execute.assert_not_called()
+    assert _outcome_metric(sink) == 'invalid'
+
+
+def _apply(initial, answers):
+    """Run the real statements on SQLite for each answer; return the final status."""
+    engine = create_engine('sqlite://')
+    with engine.begin() as connection:
+        connection.execute(
+            text('CREATE TABLE pred (model_id INTEGER, timestamp TEXT, write_status TEXT)')
+        )
+        connection.execute(
+            text('INSERT INTO pred VALUES (1, :ts, :status)'),
+            {'ts': '2026-07-13T00:00:00Z', 'status': initial},
+        )
+        for answer in answers:
+            update_stmt, exists_stmt = write_status_statements(
+                None, 'pred', answer, '2026-07-13T00:00:00Z', 1
+            )
+            connection.execute(update_stmt)
+            assert connection.execute(exists_stmt).first() is not None
+        return connection.execute(text('SELECT write_status FROM pred')).scalar()
+
+
+@pytest.mark.parametrize(
+    ('initial', 'answers', 'expected'),
+    [
+        ('pending', ['ok'], 'ok'),
+        ('pending', ['error', 'ok'], 'error'),
+        ('pending', ['ok', 'error'], 'error'),
+        ('pending', ['session_bad', 'error', 'ok'], 'session_bad'),
+        ('error', ['ok'], 'error'),
+        (None, ['ok'], 'ok'),
+    ],
+)
+def test_update_statement_keeps_the_worst_answer(initial, answers, expected):
+    assert _apply(initial, answers) == expected

tests/unit/managers/test_data_manager.py

@@ -1,9 +1,10 @@
 from unittest.mock import ANY, AsyncMock, MagicMock, patch
 
+import pytest
 from pytest import fixture, mark
 from sientia_do.notifications.models import NotificationLevel
 
-from ingestor.managers.data_manager import DataManager
+from ingestor.managers.data_manager import DataManager, redact_uri
 
 metadata = {
     'metadata': {
@@ -54,6 +55,48 @@ def test___init___success(mongodb_repository):
     assert data_manager.connection_string == 'mongodb://localhost:27017'
 
 
+@patch('ingestor.managers.data_manager.MongoDBRepository')
+def test___init___does_not_log_the_credentials(mongodb_repository):
+    logger_mock = MagicMock()
+    uri = 'mongodb://root:s3cr3t@mongodb.mongodb.svc.cluster.local:27017'
+
+    data_manager = DataManager(
+        metadata=metadata['metadata'],
+        mongo_connection_string=uri,
+        mongo_database='sientia',
+        logger=logger_mock,
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(),
+    )
+
+    logged = ' '.join(str(c) for c in logger_mock.info.call_args_list)
+    assert 's3cr3t' not in logged
+    assert 'root' not in logged
+    logger_mock.info.assert_any_call(
+        'DataManager initialized with MongoDB servers: '
+        'mongodb://***@mongodb.mongodb.svc.cluster.local:27017'
+    )
+    assert data_manager.connection_string == uri
+
+
+@pytest.mark.parametrize(
+    ('uri', 'expected'),
+    [
+        ('mongodb://localhost:27017', 'mongodb://localhost:27017'),
+        (
+            'mongodb://u:p@h1:27017,h2:27017/db?authSource=admin',
+            'mongodb://***@h1:27017,h2:27017/db?authSource=admin',
+        ),
+        ('mongodb://u:p@h:27017?authSource=admin', 'mongodb://***@h:27017?authSource=admin'),
+        ('mongodb+srv://u@cluster.example.com', 'mongodb+srv://***@cluster.example.com'),
+        ('u:p@h:27017', '***@h:27017'),
+        ('mongodb://h:27017/db?x=a@b', 'mongodb://h:27017/db?x=a@b'),
+    ],
+)
+def test_redact_uri(uri, expected):
+    assert redact_uri(uri) == expected
+
+
 def test_shutdown(data_manager):
     close_mock = MagicMock()
     data_manager.mongo_repository.close = close_mock

tests/unit/managers/test_ingestor_manager.py

@@ -1,13 +1,16 @@
 import gc
-from unittest.mock import AsyncMock, MagicMock, call, patch
+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': {
@@ -548,21 +551,153 @@ 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_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=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()
+
+    reader.unsubscribe.assert_called_once_with('slot2')
+    assert list(ingestor_manager.managed_tags) == ['slot1']
+    assert list(reader.subscriptions) == ['slot1']
 
 
 @mark.asyncio
@@ -646,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()}

tests/unit/test_gateway_runtime.py

@@ -39,6 +39,7 @@ def mocks(monkeypatch):
     patched['pool'].release_idle = AsyncMock()
     patched['NatsWriteConsumer'].return_value.start = AsyncMock()
     patched['NatsWriteConsumer'].return_value.stop = AsyncMock()
+    patched['NatsWriteConsumer'].return_value.closed = False
     patched['GrpcServer'].return_value.start = AsyncMock()
     patched['GrpcServer'].return_value.stop = AsyncMock()
     return patched
@@ -173,3 +174,47 @@ async def test_idle_sweep_releases_idle_connections(mocks):
     await runtime.idle_sweep()
 
     mocks['pool'].release_idle.assert_awaited_once_with(42.0)
+
+
+def test_write_engine_resolves_servers_from_the_pool_and_mongo(mocks):
+    """SIENTIAPDE-1975 GW1: write commands carry only the server_id."""
+    data_manager = MagicMock()
+    runtime = gr.GatewayRuntime(
+        config=GatewayConfig(),
+        data_manager=data_manager,
+        pool=mocks['pool'],
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(),
+    )
+
+    _, kwargs = mocks['WriteEngine'].call_args
+    assert kwargs['servers'] is runtime.servers
+    assert runtime.servers.pool is mocks['pool']
+    assert runtime.servers.mongo is data_manager.mongo_repository
+
+
+@pytest.mark.asyncio
+async def test_idle_sweep_restarts_a_closed_nats_connection(mocks):
+    runtime = make_runtime(pool=mocks['pool'])
+    nats_mock = mocks['NatsWriteConsumer'].return_value
+    await runtime.start()
+    assert nats_mock.start.await_count == 1
+
+    nats_mock.closed = True  # closed for good (e.g. by the server)
+    await runtime.idle_sweep()
+
+    assert nats_mock.start.await_count == 2
+    assert runtime._nats_started is True
+    runtime.logger.warning.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_idle_sweep_leaves_a_live_nats_connection_alone(mocks):
+    runtime = make_runtime(pool=mocks['pool'])
+    nats_mock = mocks['NatsWriteConsumer'].return_value
+    await runtime.start()
+
+    await runtime.idle_sweep()
+
+    assert nats_mock.start.await_count == 1

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']