feat(SIENTIAPDE-2202): write timestamp and quality per OPC server

- write_timestamp/write_quality come from the command, then the server
  document (connectors/opc-servers, read even when the reads have the
  connection open, same 60 s cache), then the WRITE_TIMESTAMP/WRITE_QUALITY
  env defaults.
- The quality comes from the command's `quality` (the prediction status:
  Good/Bad/Uncertain, Good when absent); an unknown one is a bad_quality error.
- A server that refuses them (BadWriteNotSupported, e.g. Kepware) gets the
  value written again alone and is remembered until restart: one warning and
  opc_write_value_only_fallback_total per server.

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

.env.example

@@ -23,6 +23,7 @@ GRPC_ENABLED="false"
 GRPC_PORT="50051"
 CERT_BUCKET="sientia-dataops"
 CERT_CACHE_DIR="/tmp/opc-certs"
+# Defaults of the write options; the server document and the command override them.
 WRITE_TIMESTAMP="true"
 WRITE_QUALITY="true"
 WRITE_MIN_PAUSE_S="0.0"

README.md

@@ -79,9 +79,21 @@ The Laborious publishes one command per tag (SIENTIAPDE-1975):
 {"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}
+ "quality": "Good"}
 ```
 
+- **Timestamp and quality** (SIENTIAPDE-2202): besides the value, the write carries the
+  prediction `timestamp` as `SourceTimestamp` and the `quality` (the prediction status:
+  `Good`, `Bad` or `Uncertain`; `Good` when absent) as `StatusCode`. Each one is on or
+  off by, in order: `write_timestamp`/`write_quality` in the command, the same boolean
+  fields on the server document (`connectors` or `opc-servers`, cached like the
+  connection), then the `WRITE_TIMESTAMP`/`WRITE_QUALITY` env vars (default `true`).
+  When a server refuses them (`BadWriteNotSupported`, e.g. Kepware), the value is
+  written again alone, the server is remembered until the gateway restarts (its next
+  writes go with the value alone), a warning is logged and
+  `opc_write_value_only_fallback_total` is incremented. Set the two fields to `false` on
+  such a server's document to skip the refused attempt.
+
 - **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

ingestor/connection/servers.py

@@ -4,6 +4,7 @@ The Laborious only knows the `server_id` of a write tag (pipeline `write_tags`);
 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.
+The same document may also set the write options of the server (SIENTIAPDE-2202).
 """
 
 from typing import Any
@@ -15,6 +16,7 @@ from ingestor.connection.pool import OpcConnectionPool
 
 CONNECTORS = 'connectors'
 OPC_SERVERS = 'opc-servers'
+WRITE_OPTIONS = ('write_timestamp', 'write_quality')
 
 
 def _id_forms(server_id: str) -> list[Any]:
@@ -47,8 +49,15 @@ def _opc_server_cfg(server: dict[str, Any]) -> dict[str, Any] | None:
     }
 
 
+def _write_options(document: dict[str, Any]) -> dict[str, bool]:
+    """The write options the server document sets. Only booleans count: a hand-edited
+    string such as "false" must not turn an option on."""
+    return {key: document[key] for key in WRITE_OPTIONS if isinstance(document.get(key), bool)}
+
+
 class ServerConfigResolver:
-    """Resolve and cache the connection settings of an OPC server by `server_id`.
+    """Resolve and cache the connection settings (and write options) 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
@@ -91,17 +100,40 @@ class ServerConfigResolver:
         pooled = self.pool.config_for(server_id)
         if pooled is not None:
             return pooled
+        server = await self._server(server_id)
+        return dict(server['connection']) if server is not None else None
+
+    async def write_options(self, server_id: str) -> dict[str, bool]:
+        """`write_timestamp`/`write_quality` set on the server document (SIENTIAPDE-2202),
+        only the ones it sets: the gateway defaults apply to the others.
+
+        Read from MongoDB even when the read path has the connection open (`resolve`
+        skips MongoDB then), with the same cache. Never raises: when MongoDB cannot be
+        read, the gateway defaults apply and the write goes on.
+        """
+        try:
+            server = await self._server(str(server_id))
+        except Exception as e:
+            self.logger.warning(
+                f'Write options of OPC server {server_id} not read, using the gateway defaults: {e}'
+            )
+            return {}
+        return dict(server['write_options']) if server is not None else {}
+
+    async def _server(self, server_id: str) -> dict[str, Any] | None:
         now = self.clock.elapsed()
         cached = self._cache.get(server_id)
         if cached is not None:
-            cfg, cached_at = cached
+            server, cached_at = cached
-            if now - cached_at < (self.ttl_s if cfg is not None else self.missing_ttl_s):
+            if now - cached_at < (self.ttl_s if server is not None else self.missing_ttl_s):
-                return dict(cfg) if cfg is not None else None
+                return server
-        cfg = await self._lookup(server_id)
+        server = await self._lookup(server_id)
-        self._cache[server_id] = (cfg, now)
+        self._cache[server_id] = (server, now)
-        return dict(cfg) if cfg is not None else None
+        return server
 
     async def _lookup(self, server_id: str) -> dict[str, Any] | None:
+        """The connection settings and write options of the first document that knows
+        the server, or None."""
         ids = {'$in': _id_forms(server_id)}
         connector = await self.mongo.find_one(
             CONNECTORS,
@@ -110,11 +142,12 @@ class ServerConfigResolver:
         )
         cfg = _connector_cfg(connector) if connector else None
         if cfg is not None:
-            return cfg
+            return {'connection': cfg, 'write_options': _write_options(connector)}
         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
+            return None
+        return {'connection': cfg, 'write_options': _write_options(server)}

ingestor/engines/write.py

@@ -9,7 +9,7 @@ import asyncio
 from dataclasses import dataclass
 
 from asyncua import ua
-from asyncua.ua.uaerrors import UaStatusCodeError
+from asyncua.ua.uaerrors import BadWriteNotSupported, 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
@@ -57,6 +57,23 @@ _SESSION_BAD_NAMES = frozenset(
 )
 
 
+# `quality` of the C2 command (the Laborious sends its prediction_status) -> OPC quality.
+_QUALITY_CODES = {
+    'good': ua.StatusCodes.Good,
+    'uncertain': ua.StatusCodes.Uncertain,
+    'bad': ua.StatusCodes.Bad,
+}
+
+
+class _InvalidFieldError(Exception):
+    """A command field needed by an option that is on (timestamp, quality) is bad."""
+
+    def __init__(self, error_type: str, message: str):
+        super().__init__(message)
+        self.error_type = error_type
+        self.message = message
+
+
 @dataclass
 class WriteResult:
     ok: bool
@@ -85,11 +102,14 @@ class WriteEngine(SientiaMonitoring):
         # 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
+        # Defaults (env); the server document and the command override them (SIENTIAPDE-2202).
         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
+        # Servers that refused timestamp/quality: written with the value alone.
+        self._value_only_servers: set[str] = set()
 
         SientiaMonitoring.__init__(
             self,
@@ -110,12 +130,6 @@ class WriteEngine(SientiaMonitoring):
         except ConversionError as e:
             return await self._emit_error(cmd, 'conversion', str(e), 'error')
 
-        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:
@@ -125,11 +139,18 @@ class WriteEngine(SientiaMonitoring):
                 cmd, 'server_not_found', f'OPC server {cmd["server_id"]} is not configured', 'error'
             )
 
+        server_id = str(cmd['server_id'])
+        write_timestamp, write_quality = await self._write_flags(server_id, cmd)
+        try:
+            dv = self._build_datavalue(cmd, typed, write_timestamp, write_quality)
+        except _InvalidFieldError as e:
+            return await self._emit_error(cmd, e.error_type, e.message, 'error')
+
         start = self.clock.elapsed()
         try:
             client = await self.pool.get(cmd['server_id'], server_cfg)
             node = client.get_node(cmd['addr'])
-            await node.write_value(dv)
+            await self._write_node(node, dv, server_id)
         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)
@@ -163,17 +184,76 @@ class WriteEngine(SientiaMonitoring):
         # 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:
+    async def _write_flags(self, server_id: str, cmd: dict) -> tuple[bool, bool]:
+        """`write_timestamp` and `write_quality` of this write (SIENTIAPDE-2202): the
+        command, then the server document, then the gateway env. A server that already
+        refused them (BadWriteNotSupported) gets the value alone."""
+        if server_id in self._value_only_servers:
+            return False, False
+        options = await self.servers.write_options(server_id) if self.servers is not None else {}
+
+        def flag(key: str, default: bool) -> bool:
+            value = cmd.get(key)
+            return bool(options.get(key, default) if value is None else value)
+
+        return flag('write_timestamp', self.write_timestamp), flag(
+            'write_quality', self.write_quality
+        )
+
+    def _build_datavalue(
+        self, cmd: dict, typed_value, write_timestamp: bool, write_quality: bool
+    ) -> ua.DataValue:
+        """The value, with the prediction timestamp and quality when they are on.
+
+        Raises:
+            _InvalidFieldError: When an option is on and its command field is missing or bad.
+        """
         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
+        status_code = self._status_code(cmd) if write_quality else None
-        source_ts = ua.DateTime.fromisoformat(cmd['timestamp']) if write_timestamp else None
+        try:
+            source_ts = ua.DateTime.fromisoformat(cmd['timestamp']) if write_timestamp else None
+        except (KeyError, TypeError, ValueError) as e:
+            raise _InvalidFieldError('bad_timestamp', repr(e)) from e
         return ua.DataValue(variant, StatusCode=status_code, SourceTimestamp=source_ts)
 
+    @staticmethod
+    def _status_code(cmd: dict) -> ua.StatusCode:
+        """The OPC quality of the command's `quality` (the prediction status). Commands
+        without it (before SIENTIAPDE-2202) are written as Good."""
+        quality = cmd.get('quality') or 'good'
+        code = _QUALITY_CODES.get(str(quality).strip().lower())
+        if code is None:
+            raise _InvalidFieldError('bad_quality', f'unknown quality {quality!r}')
+        return ua.StatusCode(ua.UInt32(code))
+
+    async def _write_node(self, node, dv: ua.DataValue, server_id: str) -> None:
+        """Write `dv`. When the server refuses the timestamp/quality it carries
+        (BadWriteNotSupported, e.g. Kepware), write the value alone and remember the
+        server, so its next writes go without them."""
+        try:
+            await node.write_value(dv)
+        except BadWriteNotSupported:
+            if dv.StatusCode is None and dv.SourceTimestamp is None:
+                raise
+            await node.write_value(ua.DataValue(dv.Value, StatusCode=None))
+            await self._remember_value_only(server_id)
+
+    async def _remember_value_only(self, server_id: str) -> None:
+        self._value_only_servers.add(server_id)
+        self.logger.warning(
+            f'OPC server {server_id} refused the timestamp/quality of a write '
+            f'(BadWriteNotSupported): its writes go with the value alone until the gateway '
+            f'restarts. Set write_timestamp/write_quality to false on its document to skip '
+            f'the refused attempt.'
+        )
+        await self.emit_metric(
+            metric_object=metrics.OPC_WRITE_VALUE_ONLY_FALLBACK,
+            tags={'pod_id': self.pod_id, 'server_name': server_id},
+        )
+
     def _is_session_bad(self, exc: Exception) -> bool:
         return isinstance(exc, UaStatusCodeError) and type(exc).__name__ in _SESSION_BAD_NAMES
 

ingestor/metrics.py

@@ -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_VALUE_ONLY_FALLBACK = Counter(
+    'opc_write_value_only_fallback_total',
+    'Servers that refused the timestamp/quality of a write (BadWriteNotSupported); '
+    'their writes go with the value alone',
+    ['pod_id', 'server_name'],
+)
 OPC_WRITE_STATUS_UPDATE_TOTAL = Counter(
     'opc_write_status_update_total',
     'write_status UPDATEs on the prediction rows',

tests/unit/connection/test_servers.py

@@ -149,3 +149,67 @@ async def test_mongo_errors_propagate_to_the_caller():
 
     with pytest.raises(RuntimeError, match='mongo down'):
         await resolver.resolve('1', {})
+
+
+# --- SIENTIAPDE-2202: write options on the server document ---
+
+
+@pytest.mark.asyncio
+async def test_write_options_come_from_the_connector():
+    resolver = make_resolver(
+        connector={**CONNECTOR, 'write_timestamp': False, 'write_quality': True},
+        opc_server={**OPC_SERVER, 'write_timestamp': True},
+    )
+
+    assert await resolver.write_options('1') == {'write_timestamp': False, 'write_quality': True}
+
+
+@pytest.mark.asyncio
+async def test_write_options_come_from_opc_servers_without_tag_database():
+    resolver = make_resolver(opc_server={**OPC_SERVER, 'write_quality': False})
+
+    assert await resolver.write_options('1') == {'write_quality': False}
+
+
+@pytest.mark.asyncio
+async def test_only_boolean_write_options_count():
+    resolver = make_resolver(
+        connector={**CONNECTOR, 'write_timestamp': 'false', 'write_quality': 0}
+    )
+
+    assert await resolver.write_options('1') == {}
+
+
+@pytest.mark.asyncio
+async def test_write_options_are_read_even_with_a_live_pooled_connection():
+    pooled = {'url': 'opc.tcp://read:1', 'server_uri': 'urn:read', 'use_security': True}
+    resolver = make_resolver(connector={**CONNECTOR, 'write_timestamp': False}, pooled=pooled)
+
+    assert await resolver.resolve('1', {}) == pooled
+    assert await resolver.write_options('1') == {'write_timestamp': False}
+
+
+@pytest.mark.asyncio
+async def test_write_options_share_the_lookup_cache():
+    resolver = make_resolver(connector={**CONNECTOR, 'write_quality': False})
+
+    await resolver.resolve('1', {})
+    await resolver.write_options('1')
+
+    assert resolver.mongo.find_one.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_write_options_of_an_unknown_server_are_empty():
+    resolver = make_resolver()
+
+    assert await resolver.write_options('9') == {}
+
+
+@pytest.mark.asyncio
+async def test_write_options_never_raise():
+    resolver = make_resolver()
+    resolver.mongo.find_one = AsyncMock(side_effect=RuntimeError('mongo down'))
+
+    assert await resolver.write_options('1') == {}
+    assert 'mongo down' in resolver.logger.warning.call_args.args[0]

tests/unit/engines/test_write.py

@@ -3,6 +3,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from asyncua import ua
+from asyncua.ua.uaerrors import BadTypeMismatch, BadWriteNotSupported
 
 import ingestor.metrics as metrics
 from ingestor.engines.write import WriteEngine
@@ -10,9 +11,10 @@ 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):
+def make_resolver(cfg=SERVER_CFG, write_options=None):
     resolver = MagicMock()
     resolver.resolve = AsyncMock(return_value=cfg)
+    resolver.write_options = AsyncMock(return_value=write_options or {})
     return resolver
 
 
@@ -367,3 +369,151 @@ async def test_write_bad_timestamp_records_error_without_raising(timestamp):
     assert result.opc_status_code == 'bad_timestamp'
     eng.pool.get.assert_not_awaited()
     eng.sink.update_write_status.assert_awaited_once()
+
+
+# --- SIENTIAPDE-2202: timestamp/quality options, quality from the command, value-only retry ---
+
+
+@pytest.mark.asyncio
+async def test_server_options_override_the_gateway_defaults():
+    eng = make_engine(
+        write_timestamp=True,
+        write_quality=True,
+        servers=make_resolver(write_options={'write_timestamp': False}),
+    )
+    node = make_node()
+    eng.pool.get.return_value = make_client(node)
+
+    await eng.write(base_cmd())
+
+    eng.servers.write_options.assert_awaited_once_with('srv1')
+    dv = node.write_value.await_args.args[0]
+    assert dv.SourceTimestamp is None
+    assert dv.StatusCode is not None  # not set on the server: the gateway default applies
+
+
+@pytest.mark.asyncio
+async def test_command_options_override_the_server_options():
+    eng = make_engine(
+        servers=make_resolver(write_options={'write_timestamp': False, 'write_quality': False})
+    )
+    node = make_node()
+    eng.pool.get.return_value = make_client(node)
+
+    await eng.write(base_cmd(write_timestamp=True))
+
+    dv = node.write_value.await_args.args[0]
+    assert dv.SourceTimestamp == datetime.fromisoformat('2026-07-13T12:00:00Z')
+    assert dv.StatusCode is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ('quality', 'expected'),
+    [
+        ('Good', ua.StatusCodes.Good),
+        ('Bad', ua.StatusCodes.Bad),
+        ('uncertain', ua.StatusCodes.Uncertain),
+        (None, ua.StatusCodes.Good),
+    ],
+)
+async def test_quality_comes_from_the_command(quality, expected):
+    eng = make_engine(write_quality=True)
+    node = make_node()
+    eng.pool.get.return_value = make_client(node)
+
+    result = await eng.write(base_cmd(quality=quality))
+
+    assert result.status == 'ok'
+    assert node.write_value.await_args.args[0].StatusCode.value == expected
+
+
+@pytest.mark.asyncio
+async def test_unknown_quality_records_error_without_writing():
+    eng = make_engine(write_quality=True)
+
+    result = await eng.write(base_cmd(quality='Excellent'))
+
+    assert result.status == 'error'
+    assert result.opc_status_code == 'bad_quality'
+    eng.pool.get.assert_not_awaited()
+    eng.sink.update_write_status.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_unknown_quality_is_ignored_when_quality_is_off():
+    eng = make_engine(write_quality=False)
+    node = make_node()
+    eng.pool.get.return_value = make_client(node)
+
+    result = await eng.write(base_cmd(quality='Excellent'))
+
+    assert result.status == 'ok'
+    assert node.write_value.await_args.args[0].StatusCode is None
+
+
+@pytest.mark.asyncio
+async def test_refused_timestamp_and_quality_retry_with_the_value_alone():
+    eng = make_engine(write_timestamp=True, write_quality=True)
+    node = make_node()
+    node.write_value.side_effect = [BadWriteNotSupported(), None]
+    eng.pool.get.return_value = make_client(node)
+
+    result = await eng.write(base_cmd())
+
+    assert result.status == 'ok'
+    first, retry = (c.args[0] for c in node.write_value.await_args_list)
+    assert first.SourceTimestamp is not None and first.StatusCode is not None
+    assert retry.SourceTimestamp is None and retry.StatusCode is None
+    assert retry.Value.Value == 42.0
+    assert 'srv1' in eng.logger.warning.call_args.args[0]
+    eng.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_WRITE_VALUE_ONLY_FALLBACK,
+        tags={'pod_id': eng.pod_id, 'server_name': 'srv1'},
+    )
+    eng.sink.update_write_status.assert_awaited_once_with(base_cmd(), result)
+
+
+@pytest.mark.asyncio
+async def test_server_that_refused_gets_the_value_alone_next_time():
+    eng = make_engine(write_timestamp=True, write_quality=True)
+    node = make_node()
+    node.write_value.side_effect = [BadWriteNotSupported(), None, None]
+    eng.pool.get.return_value = make_client(node)
+
+    await eng.write(base_cmd())
+    result = await eng.write(base_cmd())
+
+    assert result.status == 'ok'
+    assert node.write_value.await_count == 3  # refused + retry, then one value-only write
+    last = node.write_value.await_args.args[0]
+    assert last.SourceTimestamp is None and last.StatusCode is None
+    eng.servers.write_options.assert_awaited_once()  # not looked up once remembered
+    assert eng.logger.warning.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_refused_value_alone_is_an_error():
+    eng = make_engine(write_timestamp=False, write_quality=False)
+    node = make_node()
+    node.write_value.side_effect = BadWriteNotSupported()
+    eng.pool.get.return_value = make_client(node)
+
+    result = await eng.write(base_cmd())
+
+    assert result.status == 'error'
+    assert result.opc_status_code == 'BadWriteNotSupported'
+    assert node.write_value.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_failed_value_only_retry_is_reported():
+    eng = make_engine(write_timestamp=True, write_quality=True)
+    node = make_node()
+    node.write_value.side_effect = [BadWriteNotSupported(), BadTypeMismatch()]
+    eng.pool.get.return_value = make_client(node)
+
+    result = await eng.write(base_cmd())
+
+    assert result.status == 'error'
+    assert result.opc_status_code == 'BadTypeMismatch'