Merge pull request #37 from Aignosi/release-rc/ingestor1

SIENTIAPDE-1973: OPC Gateway — RC1 (write via NATS, active read, gRPC)
2026-10-01 12:02:00 -03:00
parent e76df05e7a ff9339d408
commit a083099e2d
22 changed files with 1407 additions and 6 deletions
--- a/ingestor/adapters/init.py
+++ b/ingestor/adapters/init.py
--- a/ingestor/adapters/grpc_server.py
+++ b/ingestor/adapters/grpc_server.py
@@ -0,0 +1,68 @@
+"""gRPC input adapter for on-demand tag reads (contract C4).
+
+Thin adapter: unpack the request, call ActiveReadEngine.read_now, map the
+result onto ReadTagResponse. Internal-only for now -- no auth/TLS, per the
+feature owner (no consumer exists yet, likely a frontend later).
+"""
+
+import grpc
+from sientia_do.observability.logger import Logger
+
+from ingestor.engines.active_read import ActiveReadEngine
+from ingestor.proto import opc_gateway_pb2, opc_gateway_pb2_grpc
+
+
+class ReadTagServicer(opc_gateway_pb2_grpc.OpcGatewayServicer):
+    def __init__(self, active_read: ActiveReadEngine, logger: Logger, slot_lookup=None):
+        self.active_read = active_read
+        self.logger = logger
+        # ponytail: slot_lookup wiring (address -> names from the current slot)
+        # lands in Task 18 (app wiring); optional callable keeps this servicer usable now.
+        self.slot_lookup = slot_lookup
+
+    async def ReadTag(self, request, context):
+        names = None
+        if self.slot_lookup is not None:
+            names = self.slot_lookup(request.server_id, request.addr)
+        try:
+            result = await self.active_read.read_now(request.server_id, request.addr, names=names)
+        except Exception as e:
+            self.logger.error(f'ReadTag failed for {request.server_id}:{request.addr}: {e}')
+            await context.abort(grpc.StatusCode.INTERNAL, f'read failed: {e}')
+            return
+        return opc_gateway_pb2.ReadTagResponse(
+            value=str(result['value']),
+            opc_status_code=result['opc_status_code'] or '',
+            source_timestamp=result['source_timestamp'] or '',
+            server_timestamp=result['server_timestamp'] or '',
+            persisted=result['persisted'],
+        )
+
+
+class GrpcServer:
+    def __init__(
+        self,
+        active_read: ActiveReadEngine,
+        logger: Logger,
+        port: int = 50051,
+        slot_lookup=None,
+    ):
+        self.active_read = active_read
+        self.logger = logger
+        self.port = port
+        self.slot_lookup = slot_lookup
+        self.server: grpc.aio.Server | None = None
+
+    async def start(self) -> None:
+        self.server = grpc.aio.server()
+        opc_gateway_pb2_grpc.add_OpcGatewayServicer_to_server(
+            ReadTagServicer(self.active_read, self.logger, self.slot_lookup), self.server
+        )
+        self.server.add_insecure_port(f'[::]:{self.port}')
+        await self.server.start()
+        self.logger.info(f'GrpcServer started on port {self.port}')
+
+    async def stop(self) -> None:
+        if self.server is None:
+            return
+        await self.server.stop(grace=5)
--- a/ingestor/adapters/nats_consumer.py
+++ b/ingestor/adapters/nats_consumer.py
@@ -0,0 +1,55 @@
+"""NATS input adapter for OPC write commands (contract C2).
+
+Thin adapter: decode the message JSON and hand the dict to WriteEngine,
+which is already raise-safe against malformed commands. No metrics of its
+own — the engine records write metrics; a malformed-JSON drop is logged
+here. No reconnect logic — nats-py handles reconnection internally.
+"""
+
+import json
+
+import nats
+from nats.aio.client import Client
+from nats.aio.subscription import Subscription
+from sientia_do.observability.logger import Logger
+
+from ingestor.engines.write import WriteEngine
+
+
+class NatsWriteConsumer:
+    def __init__(
+        self,
+        nats_url: str,
+        write_engine: WriteEngine,
+        logger: Logger,
+        subject: str = 'opc.write',
+        queue: str = 'gateway-writers',
+    ):
+        self.nats_url = nats_url
+        self.write_engine = write_engine
+        self.logger = logger
+        self.subject = subject
+        self.queue = queue
+        self.nc: Client | None = None
+        self.sub: Subscription | None = None
+
+    async def start(self) -> None:
+        self.nc = await nats.connect(servers=self.nats_url)
+        self.sub = await self.nc.subscribe(self.subject, queue=self.queue, cb=self._handle)
+        self.logger.info(f'NatsWriteConsumer started: subject={self.subject} queue={self.queue}')
+
+    async def _handle(self, msg) -> None:
+        try:
+            payload = json.loads(msg.data)
+        except ValueError as e:  # includes JSONDecodeError and UnicodeDecodeError
+            preview = msg.data[:200]
+            self.logger.error(f'NatsWriteConsumer dropped malformed JSON ({e}): {preview!r}')
+            return
+        await self.write_engine.write(payload)
+
+    async def stop(self) -> None:
+        if self.nc is None:
+            return
+        if self.sub is not None:
+            await self.sub.unsubscribe()
+        await self.nc.drain()
--- a/ingestor/converters.py
+++ b/ingestor/converters.py
@@ -25,6 +25,8 @@ def to_typed_value(raw, data_type: str):
    data_type: 'boolean' | 'int' | 'float' | 'double' | 'string'.
    Raises ConversionError on any invalid input (never a bare TypeError).
    """
+    if not isinstance(data_type, str):
+        raise ConversionError(f'data_type must be a string, got {type(data_type).__name__}')
    dtype = data_type.strip().lower()
    try:
        if dtype in ('bool', 'boolean'):
--- a/ingestor/engines/active_read.py
+++ b/ingestor/engines/active_read.py
@@ -0,0 +1,132 @@
+"""Active OPC reads (F3): on-demand (`read_now`, used by the gRPC server -
+Task 16) and scheduled (`scheduled_tick`, driven by the app main loop each
+collect cycle). Emits metrics only -- unlike WriteEngine there is no
+notification path here."""
+
+from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+from sientia_do.observability.logger import Logger
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, OPC_TIMEZONE
+
+import ingestor.metrics as metrics
+from ingestor.connection.clock import MonotonicClock
+from ingestor.connection.pool import OpcConnectionPool
+from ingestor.io.data_sink import DataSink
+
+
+class ActiveReadEngine(SientiaMonitoring):
+    """On-demand + scheduled OPC reads, sharing the pool/sink with WriteEngine
+    and SubscriptionEngine."""
+
+    def __init__(
+        self,
+        pool: OpcConnectionPool,
+        sink: DataSink,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+    ):
+        self.pool = pool
+        self.sink = sink
+        self.clock = MonotonicClock()
+        self._ticks: dict[tuple[str, str], int] = {}
+
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+    async def read_now(
+        self,
+        server_id: str,
+        addr: str,
+        names: list[dict] | None = None,
+        server_cfg: dict | None = None,
+    ) -> dict:
+        """Read `addr` on `server_id` right now; persist via sink only if `names` given."""
+        tags = {'pod_id': self.pod_id, 'server_name': server_id}
+        start = self.clock.elapsed()
+        try:
+            client = await self.pool.get(server_id, server_cfg or {})
+            dv = await client.get_node(addr).read_data_value()
+        except Exception as e:
+            await self.emit_metric(
+                metric_object=metrics.OPC_ACTIVE_READ_TOTAL, tags={**tags, 'result': 'error'}
+            )
+            self.logger.error(f'Active read failed for {server_id}:{addr}: {e}')
+            raise
+
+        await self.emit_metric(
+            metric_object=metrics.OPC_ACTIVE_READ_LATENCY,
+            method='observe',
+            value=self.clock.elapsed() - start,
+            tags=tags,
+        )
+        await self.emit_metric(
+            metric_object=metrics.OPC_ACTIVE_READ_TOTAL, tags={**tags, 'result': 'success'}
+        )
+        return await self._build_result(dv, addr, names, tags)
+
+    async def _build_result(self, dv, addr: str, names: list[dict] | None, tags: dict) -> dict:
+        value, status_name, source_ts, server_ts = self._extract(dv)
+        await self._record_read_lag(source_ts, tags)
+        source_str = self._format_ts(source_ts)
+        persisted = bool(names)
+        if names:
+            await self.sink.publish_reading(
+                tag=addr, value=value, timestamp=source_str, names=names
+            )
+        return {
+            'value': value,
+            'opc_status_code': status_name,
+            'source_timestamp': source_str,
+            'server_timestamp': self._format_ts(server_ts),
+            'persisted': persisted,
+        }
+
+    @staticmethod
+    def _extract(dv):
+        status_name = dv.StatusCode.name if dv.StatusCode is not None else None
+        return dv.Value.Value, status_name, dv.SourceTimestamp, dv.ServerTimestamp
+
+    @staticmethod
+    def _format_ts(ts) -> str | None:
+        if ts is None:
+            return None
+        return ts.replace(tzinfo=OPC_TIMEZONE).strftime(DATETIME_FORMAT_WITH_TZ)
+
+    async def _record_read_lag(self, source_ts, tags: dict) -> None:
+        if source_ts is None:
+            return
+        lag = MonotonicClock.now_wall() - source_ts.replace(tzinfo=OPC_TIMEZONE).timestamp()
+        await self.emit_metric(
+            metric_object=metrics.OPC_READ_LAG, method='set', value=lag, tags=tags
+        )
+
+    async def scheduled_tick(self, slot: dict) -> None:
+        """Sweep every server/tag in a parsed slot, reading tags due for an active read."""
+        for server in slot.values():
+            server_id = server['server_id']
+            server_cfg = {k: v for k, v in server.items() if k != 'tags'}
+            for addr, tag in server['tags'].items():
+                await self._tick_tag(server_id, addr, tag, server_cfg)
+
+    async def _tick_tag(self, server_id: str, addr: str, tag: dict, server_cfg: dict) -> None:
+        active_read = tag.get('active_read') or {}
+        if not active_read.get('enabled'):
+            return
+        if not self._should_read(server_id, addr, active_read.get('cycle_multiple', 1)):
+            return
+        try:
+            await self.read_now(server_id, addr, names=tag['names'], server_cfg=server_cfg)
+        except Exception as e:
+            self.logger.error(f'Scheduled active read failed for {server_id}:{addr}: {e}')
+
+    def _should_read(self, server_id: str, addr: str, cycle_multiple: int) -> bool:
+        key = (server_id, addr)
+        count = self._ticks.get(key, 0) + 1
+        self._ticks[key] = count
+        return count % max(cycle_multiple, 1) == 0
--- a/ingestor/engines/write.py
+++ b/ingestor/engines/write.py
@@ -0,0 +1,195 @@
+"""OPC write execution (F2 capstone) from a NATS C2 command dict.
+
+Fixes 1879 correctness gaps: safe value coercion (never raises on bad
+input), StatusCode/SourceTimestamp fidelity on the DataValue actually sent
+to the server, a rate-limit floor between writes, and OPC error typing.
+"""
+
+import asyncio
+from dataclasses import dataclass
+
+from asyncua import ua
+from asyncua.ua.uaerrors import UaStatusCodeError
+from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+from sientia_do.observability.logger import Logger
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+
+import ingestor.metrics as metrics
+from ingestor.connection.clock import MonotonicClock
+from ingestor.connection.pool import OpcConnectionPool
+from ingestor.converters import ConversionError, to_typed_value
+from ingestor.io.data_sink import DataSink
+
+_VARIANT_TYPES = {
+    'float': ua.VariantType.Float,
+    'double': ua.VariantType.Double,
+    'real': ua.VariantType.Double,
+    'int': ua.VariantType.Int32,
+    'integer': ua.VariantType.Int32,
+    'bool': ua.VariantType.Boolean,
+    'boolean': ua.VariantType.Boolean,
+    'str': ua.VariantType.String,
+    'string': ua.VariantType.String,
+}
+
+# ponytail: same reconnect-worthy Bad*-status subset as laborious's
+# opc_repository.py (different repo, not importable here); keep in sync by
+# hand or promote to a shared lib if it drifts.
+_SESSION_BAD_NAMES = frozenset(
+    {
+        'BadSessionIdInvalid',
+        'BadSessionClosed',
+        'BadSessionNotActivated',
+        'BadSecureChannelIdInvalid',
+        'BadSecureChannelClosed',
+        'BadSecureChannelTokenUnknown',
+        'BadTcpSecureChannelUnknown',
+        'BadServerNotConnected',
+        'BadConnectionClosed',
+        'BadConnectionRejected',
+        'BadNotConnected',
+        'BadTimeout',
+        'BadRequestTimeout',
+        'BadTooManySessions',
+    }
+)
+
+
+@dataclass
+class WriteResult:
+    ok: bool
+    status: str  # 'ok' | 'error' | 'session_bad'
+    opc_status_code: str
+    message: str
+
+
+class WriteEngine(SientiaMonitoring):
+    """Executes one OPC write from a NATS C2 command."""
+
+    def __init__(
+        self,
+        pool: OpcConnectionPool,
+        sink: DataSink,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+        write_timestamp: bool = True,
+        write_quality: bool = True,
+        min_pause_s: float = 0.0,
+    ):
+        self.pool = pool
+        self.sink = sink
+        self.write_timestamp = write_timestamp
+        self.write_quality = write_quality
+        self.min_pause_s = min_pause_s
+        self.clock = MonotonicClock()
+        self._last_write_at: float | None = None
+
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+    async def write(self, cmd: dict) -> WriteResult:
+        await self._rate_limit()
+
+        missing = [k for k in ('value', 'data_type', 'server_id', 'addr') if k not in cmd]
+        if missing:
+            return await self._emit_error(cmd, 'bad_command', f'missing fields: {missing}', 'error')
+
+        try:
+            typed = to_typed_value(cmd['value'], cmd['data_type'])
+        except ConversionError as e:
+            return await self._emit_error(cmd, 'conversion', str(e), 'error')
+
+        dv = self._build_datavalue(cmd, typed)
+
+        start = self.clock.elapsed()
+        try:
+            client = await self.pool.get(cmd['server_id'], self._server_cfg(cmd))
+            node = client.get_node(cmd['addr'])
+            await node.write_value(dv)
+        except Exception as e:
+            status = 'session_bad' if self._is_session_bad(e) else 'error'
+            return await self._emit_error(cmd, type(e).__name__, str(e), status)
+
+        return await self._emit_success(cmd, start)
+
+    async def _rate_limit(self) -> None:
+        now = self.clock.elapsed()
+        if self.min_pause_s > 0 and self._last_write_at is not None:
+            remaining = self.min_pause_s - (now - self._last_write_at)
+            if remaining > 0:
+                await asyncio.sleep(remaining)
+                now = self.clock.elapsed()
+        self._last_write_at = now
+
+    def _server_cfg(self, cmd: dict) -> dict:
+        # ponytail: minimal cfg built straight from the C2 message. Resolving
+        # the full connector config (url/security) from the connectors DB by
+        # server_id is F4 wiring, not this task — pool is mocked in tests.
+        return {
+            'url': cmd.get('url'),
+            'use_security': cmd.get('use_security', False),
+            'server_uri': cmd.get('server_uri'),
+        }
+
+    def _variant_type(self, data_type: str) -> ua.VariantType:
+        # direct index: unknown dtypes already became ConversionError in to_typed_value
+        return _VARIANT_TYPES[data_type.strip().lower()]
+
+    def _build_datavalue(self, cmd: dict, typed_value) -> ua.DataValue:
+        variant = ua.Variant(typed_value, self._variant_type(cmd['data_type']))
+        write_quality = cmd.get('write_quality', self.write_quality)
+        write_timestamp = cmd.get('write_timestamp', self.write_timestamp)
+        # NOTE: DataValue.StatusCode default_factory populates a Good status
+        # even when omitted, so "off" must pass StatusCode=None explicitly
+        # or the flag silently no-ops.
+        status_code = ua.StatusCode(ua.UInt32(ua.StatusCodes.Good)) if write_quality else None
+        source_ts = ua.DateTime.fromisoformat(cmd['timestamp']) if write_timestamp else None
+        return ua.DataValue(variant, StatusCode=status_code, SourceTimestamp=source_ts)
+
+    def _is_session_bad(self, exc: Exception) -> bool:
+        return isinstance(exc, UaStatusCodeError) and type(exc).__name__ in _SESSION_BAD_NAMES
+
+    async def _emit_error(self, cmd: dict, error_type: str, message: str, status: str):
+        tags = {'pod_id': self.pod_id, 'server_name': cmd.get('server_id', 'unknown')}
+        await self.emit_metric(
+            metric_object=metrics.OPC_ERROR_BY_TYPE, tags={**tags, 'error_type': error_type}
+        )
+        await self.emit_metric(
+            metric_object=metrics.OPC_WRITE_TOTAL,
+            tags={**tags, 'model_id': cmd.get('model_id', 'unknown'), 'result': 'error'},
+        )
+        result = WriteResult(ok=False, status=status, opc_status_code=error_type, message=message)
+        await self.sink.update_write_status(cmd, result)
+        return result
+
+    async def _emit_success(self, cmd: dict, start: float):
+        tags = {'pod_id': self.pod_id, 'server_name': cmd.get('server_id', 'unknown')}
+        await self.emit_metric(
+            metric_object=metrics.OPC_WRITE_TOTAL,
+            tags={**tags, 'model_id': cmd.get('model_id', 'unknown'), 'result': 'success'},
+        )
+        await self.emit_metric(
+            metric_object=metrics.OPC_WRITE_LATENCY,
+            method='observe',
+            value=self.clock.elapsed() - start,
+            tags=tags,
+        )
+        await self.emit_metric(
+            metric_object=metrics.OPC_WRITE_STATUS_BY_MODEL,
+            method='set',
+            value=1,
+            tags={
+                'pod_id': self.pod_id,
+                'model_id': cmd.get('model_id', 'unknown'),
+                'status': 'ok',
+            },
+        )
+        result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='')
+        await self.sink.update_write_status(cmd, result)
+        return result
--- a/ingestor/io/data_sink.py
+++ b/ingestor/io/data_sink.py
@@ -1,16 +1,20 @@
+import re
+
+from sqlalchemy import text
+
 import ingestor.metrics as metrics

+# fullmatch (no ^/$ anchors) so a trailing newline can't slip through — `$` would
+# match before a trailing '\n', which is not what we want for a SQL identifier.
+_IDENTIFIER_RE = re.compile(r'\w+', re.ASCII)
+

 class DataSink:
-    """Gateway egress wrapper: fans a reading out to every configured name/collection.
+    """Gateway egress wrapper: fans a reading out to every configured name/collection."""

-    `postgres` is unused for now — task 12 adds `update_write_status` on top of it.
-    """
-
-    def __init__(self, data_manager, postgres, metadata: dict | None = None) -> None:
+    def __init__(self, data_manager, postgres) -> None:
        self.data_manager = data_manager
        self.postgres = postgres
-        self.metadata = metadata

    async def emit_metric(self, **kwargs) -> None:
        # ponytail: delegate to data_manager, it's already a SientiaMonitoring
@@ -37,3 +41,34 @@ class DataSink:
                    'model_id': name['model_id'],
                },
            )
+
+    async def update_write_status(self, cmd: dict, result) -> None:  # NOSONAR
+        """UPDATE the per-model prediction row's write_status (C3), by timestamp+model_id.
+
+        schema/table_name are per-model SQL identifiers, not values — they can't be
+        bound params, so they're checked against an allowlist before interpolation.
+        status/timestamp/model_id ARE bound params, never f-stringed.
+        """
+        schema = self._validate_identifier(cmd['schema'])
+        table = self._validate_identifier(cmd['table_name'])
+        # schema/table are allowlist-validated identifiers above; status/timestamp/model_id
+        # are bound params below, never interpolated — not a real injection vector.
+        stmt = text(
+            f'UPDATE {schema}.{table} SET write_status = :status '  # noqa: S608 # nosec B608
+            'WHERE timestamp = :timestamp AND model_id = :model_id'
+        )
+        params = {
+            'status': result.status,
+            'timestamp': cmd['timestamp'],
+            'model_id': cmd['model_id'],
+        }
+        # ponytail: sync engine call inside async def — same pattern sientia_do's own
+        # Postgres activities use (e.g. export_data_to_postgres); no async pg driver here.
+        with self.postgres.engine.begin() as connection:
+            connection.execute(stmt, params)
+
+    @staticmethod
+    def _validate_identifier(name: str) -> str:
+        if not isinstance(name, str) or not _IDENTIFIER_RE.fullmatch(name):
+            raise ValueError(f'invalid SQL identifier: {name!r}')
+        return name
--- a/ingestor/proto/init.py
+++ b/ingestor/proto/init.py
--- a/ingestor/proto/opc_gateway.proto
+++ b/ingestor/proto/opc_gateway.proto
@@ -0,0 +1,18 @@
+syntax = "proto3";
+package opcgateway;
+
+service OpcGateway {
+  rpc ReadTag(ReadTagRequest) returns (ReadTagResponse);
+}
+message ReadTagRequest {
+  string server_id = 1;
+  string addr = 2;
+  string data_type = 3;
+}
+message ReadTagResponse {
+  string value = 1;
+  string opc_status_code = 2;
+  string source_timestamp = 3;
+  string server_timestamp = 4;
+  bool persisted = 5;
+}
--- a/ingestor/proto/opc_gateway_pb2.py
+++ b/ingestor/proto/opc_gateway_pb2.py
@@ -0,0 +1,40 @@
+# -*- coding: utf-8 -*-
+# Generated by the protocol buffer compiler.  DO NOT EDIT!
+# NO CHECKED-IN PROTOBUF GENCODE
+# source: opc_gateway.proto
+# Protobuf Python Version: 7.35.0
+"""Generated protocol buffer code."""
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import descriptor_pool as _descriptor_pool
+from google.protobuf import runtime_version as _runtime_version
+from google.protobuf import symbol_database as _symbol_database
+from google.protobuf.internal import builder as _builder
+_runtime_version.ValidateProtobufRuntimeVersion(
+    _runtime_version.Domain.PUBLIC,
+    7,
+    35,
+    0,
+    '',
+    'opc_gateway.proto'
+)
+# @@protoc_insertion_point(imports)
+
+_sym_db = _symbol_database.Default()
+
+
+
+
+DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x11opc_gateway.proto\x12\nopcgateway\"D\n\x0eReadTagRequest\x12\x11\n\tserver_id\x18\x01 \x01(\t\x12\x0c\n\x04\x61\x64\x64r\x18\x02 \x01(\t\x12\x11\n\tdata_type\x18\x03 \x01(\t\"\x80\x01\n\x0fReadTagResponse\x12\r\n\x05value\x18\x01 \x01(\t\x12\x17\n\x0fopc_status_code\x18\x02 \x01(\t\x12\x18\n\x10source_timestamp\x18\x03 \x01(\t\x12\x18\n\x10server_timestamp\x18\x04 \x01(\t\x12\x11\n\tpersisted\x18\x05 \x01(\x08\x32P\n\nOpcGateway\x12\x42\n\x07ReadTag\x12\x1a.opcgateway.ReadTagRequest\x1a\x1b.opcgateway.ReadTagResponseb\x06proto3')
+
+_globals = globals()
+_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
+_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'opc_gateway_pb2', _globals)
+if not _descriptor._USE_C_DESCRIPTORS:
+  DESCRIPTOR._loaded_options = None
+  _globals['_READTAGREQUEST']._serialized_start=33
+  _globals['_READTAGREQUEST']._serialized_end=101
+  _globals['_READTAGRESPONSE']._serialized_start=104
+  _globals['_READTAGRESPONSE']._serialized_end=232
+  _globals['_OPCGATEWAY']._serialized_start=234
+  _globals['_OPCGATEWAY']._serialized_end=314
+# @@protoc_insertion_point(module_scope)
--- a/ingestor/proto/opc_gateway_pb2_grpc.py
+++ b/ingestor/proto/opc_gateway_pb2_grpc.py
@@ -0,0 +1,97 @@
+# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
+"""Client and server classes corresponding to protobuf-defined services."""
+import grpc
+import warnings
+
+from . import opc_gateway_pb2 as opc__gateway__pb2
+
+GRPC_GENERATED_VERSION = '1.82.1'
+GRPC_VERSION = grpc.__version__
+_version_not_supported = False
+
+try:
+    from grpc._utilities import first_version_is_lower
+    _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
+except ImportError:
+    _version_not_supported = True
+
+if _version_not_supported:
+    raise RuntimeError(
+        f'The grpc package installed is at version {GRPC_VERSION},'
+        + ' but the generated code in opc_gateway_pb2_grpc.py depends on'
+        + f' grpcio>={GRPC_GENERATED_VERSION}.'
+        + f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}'
+        + f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.'
+    )
+
+
+class OpcGatewayStub:
+    """Missing associated documentation comment in .proto file."""
+
+    def __init__(self, channel):
+        """Constructor.
+
+        Args:
+            channel: A grpc.Channel.
+        """
+        self.ReadTag = channel.unary_unary(
+                '/opcgateway.OpcGateway/ReadTag',
+                request_serializer=opc__gateway__pb2.ReadTagRequest.SerializeToString,
+                response_deserializer=opc__gateway__pb2.ReadTagResponse.FromString,
+                _registered_method=True)
+
+
+class OpcGatewayServicer:
+    """Missing associated documentation comment in .proto file."""
+
+    def ReadTag(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+
+def add_OpcGatewayServicer_to_server(servicer, server):
+    rpc_method_handlers = {
+            'ReadTag': grpc.unary_unary_rpc_method_handler(
+                    servicer.ReadTag,
+                    request_deserializer=opc__gateway__pb2.ReadTagRequest.FromString,
+                    response_serializer=opc__gateway__pb2.ReadTagResponse.SerializeToString,
+            ),
+    }
+    generic_handler = grpc.method_handlers_generic_handler(
+            'opcgateway.OpcGateway', rpc_method_handlers)
+    server.add_generic_rpc_handlers((generic_handler,))
+    server.add_registered_method_handlers('opcgateway.OpcGateway', rpc_method_handlers)
+
+
+ # This class is part of an EXPERIMENTAL API.
+class OpcGateway:
+    """Missing associated documentation comment in .proto file."""
+
+    @staticmethod
+    def ReadTag(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(
+            request,
+            target,
+            '/opcgateway.OpcGateway/ReadTag',
+            opc__gateway__pb2.ReadTagRequest.SerializeToString,
+            opc__gateway__pb2.ReadTagResponse.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+            _registered_method=True)
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -23,6 +23,7 @@ exclude = [
    "*.pyc",
    ".pytest_cache",
    "htmlcov",
+    "ingestor/proto/*_pb2*.py",
 ]

 [tool.ruff.lint]
@@ -80,6 +81,9 @@ warn_unused_ignores = false
 warn_no_return = true
 strict_equality = true
 ignore_missing_imports = true
+# Generated protoc/grpcio-tools output - not a checking target, dynamic attrs
+# mypy can't see (message classes built at runtime from DESCRIPTOR).
+exclude = ["^ingestor/proto/"]

 # Ignore missing imports for external packages
 [[tool.mypy.overrides]]
@@ -106,6 +110,14 @@ ignore_missing_imports = true
 module = "pandas.*"
 ignore_missing_imports = true

+# Generated protoc/grpcio-tools output - not hand-maintained, dynamic attrs
+# mypy can't see (message classes built at runtime from DESCRIPTOR). Skip
+# following it entirely so importers see Any instead of false attr-defined errors.
+[[tool.mypy.overrides]]
+module = "ingestor.proto.*"
+ignore_errors = true
+follow_imports = "skip"
+
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 python_files = ["test_*.py"]
--- a/sonar-project.properties
+++ b/sonar-project.properties
@@ -8,3 +8,7 @@ sonar.python.coverage.reportPaths=coverage.xml
 sonar.python.xunit.reportPath=pytest.xml
 sonar.python.version=3.11
 sonar.coverage.exclusions=ingestor/app.py
+# Generated protoc/grpcio-tools output - not hand-maintained; renaming its
+# ReadTag/etc. methods to satisfy naming-convention rules would break gRPC
+# wire compatibility (grpcio-tools dictates these names, not style choice).
+sonar.exclusions=ingestor/proto/**
--- a/tests/unit/adapters/init.py
+++ b/tests/unit/adapters/init.py
--- a/tests/unit/adapters/test_grpc_server.py
+++ b/tests/unit/adapters/test_grpc_server.py
@@ -0,0 +1,97 @@
+from unittest.mock import AsyncMock, MagicMock
+
+import grpc
+import pytest
+
+from ingestor.adapters.grpc_server import GrpcServer, ReadTagServicer
+from ingestor.proto.opc_gateway_pb2 import ReadTagRequest
+
+
+def make_request(server_id='srv1', addr='ns=2;i=5'):
+    return ReadTagRequest(server_id=server_id, addr=addr, data_type='float')
+
+
+@pytest.mark.asyncio
+async def test_read_tag_maps_full_result_onto_response():
+    active_read = AsyncMock()
+    active_read.read_now.return_value = {
+        'value': 1.5,
+        'opc_status_code': 'Good',
+        'source_timestamp': '2026-07-15T10:00:00-03:00',
+        'server_timestamp': '2026-07-15T10:00:01-03:00',
+        'persisted': False,
+    }
+    servicer = ReadTagServicer(active_read, MagicMock())
+
+    response = await servicer.ReadTag(make_request(), AsyncMock())
+
+    assert response.value == '1.5'
+    assert response.opc_status_code == 'Good'
+    assert response.source_timestamp == '2026-07-15T10:00:00-03:00'
+    assert response.server_timestamp == '2026-07-15T10:00:01-03:00'
+    assert response.persisted is False
+
+
+@pytest.mark.asyncio
+async def test_read_tag_none_timestamps_become_empty_string():
+    active_read = AsyncMock()
+    active_read.read_now.return_value = {
+        'value': 1.5,
+        'opc_status_code': None,
+        'source_timestamp': None,
+        'server_timestamp': None,
+        'persisted': False,
+    }
+    servicer = ReadTagServicer(active_read, MagicMock())
+
+    response = await servicer.ReadTag(make_request(), AsyncMock())
+
+    assert response.opc_status_code == ''
+    assert response.source_timestamp == ''
+    assert response.server_timestamp == ''
+
+
+@pytest.mark.asyncio
+async def test_read_tag_aborts_with_internal_on_engine_failure():
+    active_read = AsyncMock()
+    active_read.read_now.side_effect = RuntimeError('boom')
+    logger = MagicMock()
+    servicer = ReadTagServicer(active_read, logger)
+    context = AsyncMock()
+    context.abort.side_effect = grpc.RpcError('aborted')
+
+    request = make_request()
+    with pytest.raises(grpc.RpcError):
+        await servicer.ReadTag(request, context)
+
+    context.abort.assert_awaited_once()
+    args = context.abort.call_args.args
+    assert args[0] == grpc.StatusCode.INTERNAL
+    logger.error.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_read_tag_uses_slot_lookup_names_when_provided():
+    active_read = AsyncMock()
+    active_read.read_now.return_value = {
+        'value': 1,
+        'opc_status_code': 'Good',
+        'source_timestamp': None,
+        'server_timestamp': None,
+        'persisted': True,
+    }
+    names = [{'collection': 'c1', 'name': 't1'}]
+    slot_lookup = MagicMock(return_value=names)
+    servicer = ReadTagServicer(active_read, MagicMock(), slot_lookup=slot_lookup)
+
+    await servicer.ReadTag(make_request(server_id='srv1', addr='ns=2;i=5'), AsyncMock())
+
+    slot_lookup.assert_called_once_with('srv1', 'ns=2;i=5')
+    active_read.read_now.assert_awaited_once_with('srv1', 'ns=2;i=5', names=names)
+
+
+@pytest.mark.asyncio
+async def test_grpc_server_stop_before_start_does_not_raise():
+    server = GrpcServer(active_read=AsyncMock(), logger=MagicMock())
+
+    await server.stop()  # no raise
--- a/tests/unit/adapters/test_nats_consumer.py
+++ b/tests/unit/adapters/test_nats_consumer.py
@@ -0,0 +1,76 @@
+import json
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from ingestor.adapters.nats_consumer import NatsWriteConsumer
+
+
+def make_consumer(**kwargs):
+    return NatsWriteConsumer(
+        nats_url='nats://localhost:4222',
+        write_engine=AsyncMock(),
+        logger=MagicMock(),
+        **kwargs,
+    )
+
+
+@pytest.mark.asyncio
+async def test_handle_valid_json_dispatches_to_write_engine():
+    consumer = make_consumer()
+    payload = {'server_id': 'srv1', 'addr': 'ns=2;i=5', 'value': 1.0, 'data_type': 'float'}
+    msg = MagicMock(data=json.dumps(payload).encode())
+
+    await consumer._handle(msg)
+
+    consumer.write_engine.write.assert_awaited_once_with(payload)
+
+
+@pytest.mark.asyncio
+async def test_handle_invalid_json_drops_message_and_logs_error():
+    consumer = make_consumer()
+    msg = MagicMock(data=b'not-json')
+
+    await consumer._handle(msg)
+
+    consumer.write_engine.write.assert_not_awaited()
+    consumer.logger.error.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_start_subscribes_with_subject_and_queue():
+    consumer = make_consumer(subject='opc.write', queue='gateway-writers')
+    mock_nc = AsyncMock()
+    mock_nc.subscribe = AsyncMock()
+
+    with patch('ingestor.adapters.nats_consumer.nats') as mock_nats:
+        mock_nats.connect = AsyncMock(return_value=mock_nc)
+        await consumer.start()
+
+    mock_nats.connect.assert_awaited_once_with(servers='nats://localhost:4222')
+    mock_nc.subscribe.assert_awaited_once_with(
+        'opc.write', queue='gateway-writers', cb=consumer._handle
+    )
+
+
+@pytest.mark.asyncio
+async def test_stop_before_start_does_not_raise():
+    consumer = make_consumer()
+
+    await consumer.stop()  # no raise
+
+
+@pytest.mark.asyncio
+async def test_stop_after_start_drains_and_unsubscribes():
+    consumer = make_consumer()
+    mock_nc = AsyncMock()
+    mock_sub = AsyncMock()
+    mock_nc.subscribe = AsyncMock(return_value=mock_sub)
+
+    with patch('ingestor.adapters.nats_consumer.nats') as mock_nats:
+        mock_nats.connect = AsyncMock(return_value=mock_nc)
+        await consumer.start()
+        await consumer.stop()
+
+    mock_sub.unsubscribe.assert_awaited_once()
+    mock_nc.drain.assert_awaited_once()
--- a/tests/unit/engines/test_active_read.py
+++ b/tests/unit/engines/test_active_read.py
@@ -0,0 +1,179 @@
+from datetime import datetime
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from asyncua import ua
+
+import ingestor.metrics as metrics
+from ingestor.engines.active_read import ActiveReadEngine
+
+
+def make_engine():
+    eng = ActiveReadEngine(
+        pool=AsyncMock(),
+        sink=AsyncMock(),
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=AsyncMock(),
+    )
+    eng.emit_metric = AsyncMock()
+    return eng
+
+
+def make_dv(value=42.0, status='Good', source_ts=None, server_ts=None):
+    source_ts = source_ts or datetime(2026, 7, 13, 12, 0, 0)
+    server_ts = server_ts or source_ts
+    return ua.DataValue(
+        ua.Variant(value),
+        StatusCode=ua.StatusCode(getattr(ua.StatusCodes, status)),
+        SourceTimestamp=source_ts,
+        ServerTimestamp=server_ts,
+    )
+
+
+def make_client(dv=None):
+    node = MagicMock()
+    node.read_data_value = AsyncMock(return_value=dv or make_dv())
+    client = MagicMock()
+    client.get_node = MagicMock(return_value=node)
+    return client, node
+
+
+NAMES = [{'tag_name': 'a', 'model_id': 'm1', 'collection': 'coll'}]
+
+
+@pytest.mark.asyncio
+async def test_read_now_persists_and_returns_fields_when_names_given():
+    eng = make_engine()
+    client, node = make_client()
+    eng.pool.get.return_value = client
+
+    result = await eng.read_now('srv1', 'ns=2;i=5', names=NAMES)
+
+    client.get_node.assert_called_once_with('ns=2;i=5')
+    node.read_data_value.assert_awaited_once()
+    eng.sink.publish_reading.assert_awaited_once()
+    _, kwargs = eng.sink.publish_reading.call_args
+    assert kwargs['tag'] == 'ns=2;i=5'
+    assert kwargs['value'] == 42.0
+    assert kwargs['names'] == NAMES
+    assert result['value'] == 42.0
+    assert result['opc_status_code'] == 'Good'
+    assert result['persisted'] is True
+    assert result['source_timestamp'].startswith('2026-07-13 12:00:00')
+    assert result['server_timestamp'].startswith('2026-07-13 12:00:00')
+    eng.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_ACTIVE_READ_TOTAL,
+        tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'result': 'success'},
+    )
+    latency_call = next(
+        c
+        for c in eng.emit_metric.await_args_list
+        if c.kwargs['metric_object'] is metrics.OPC_ACTIVE_READ_LATENCY
+    )
+    assert latency_call.kwargs['method'] == 'observe'
+    assert latency_call.kwargs['value'] >= 0
+    lag_call = next(
+        c
+        for c in eng.emit_metric.await_args_list
+        if c.kwargs['metric_object'] is metrics.OPC_READ_LAG
+    )
+    assert lag_call.kwargs['method'] == 'set'
+
+
+@pytest.mark.asyncio
+async def test_read_now_without_names_does_not_persist():
+    eng = make_engine()
+    client, _ = make_client()
+    eng.pool.get.return_value = client
+
+    result = await eng.read_now('srv1', 'ns=2;i=5')
+
+    eng.sink.publish_reading.assert_not_awaited()
+    assert result['persisted'] is False
+
+
+@pytest.mark.asyncio
+async def test_read_now_error_emits_metric_and_reraises():
+    eng = make_engine()
+    node = MagicMock()
+    node.read_data_value = AsyncMock(side_effect=RuntimeError('boom'))
+    client = MagicMock()
+    client.get_node = MagicMock(return_value=node)
+    eng.pool.get.return_value = client
+
+    with pytest.raises(RuntimeError, match='boom'):
+        await eng.read_now('srv1', 'ns=2;i=5')
+
+    eng.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_ACTIVE_READ_TOTAL,
+        tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'result': 'error'},
+    )
+    eng.sink.publish_reading.assert_not_awaited()
+
+
+def make_slot(cycle_multiple=None, enabled=True):
+    active_read = {'enabled': enabled}
+    if cycle_multiple is not None:
+        active_read['cycle_multiple'] = cycle_multiple
+    return {
+        'srv_name': {
+            'server_id': 'srv1',
+            'url': 'opc.tcp://x',
+            'tags': {
+                'ns=2;i=5': {'frequency': 1000, 'names': NAMES, 'active_read': active_read},
+            },
+        }
+    }
+
+
+@pytest.mark.asyncio
+async def test_scheduled_tick_respects_cycle_multiple():
+    eng = make_engine()
+    eng.read_now = AsyncMock()
+    slot = make_slot(cycle_multiple=3)
+
+    await eng.scheduled_tick(slot)
+    await eng.scheduled_tick(slot)
+    eng.read_now.assert_not_awaited()
+    await eng.scheduled_tick(slot)
+    eng.read_now.assert_awaited_once_with(
+        'srv1', 'ns=2;i=5', names=NAMES, server_cfg={'server_id': 'srv1', 'url': 'opc.tcp://x'}
+    )
+
+
+@pytest.mark.asyncio
+async def test_scheduled_tick_skips_disabled_tags():
+    eng = make_engine()
+    eng.read_now = AsyncMock()
+    slot = make_slot(enabled=False)
+
+    await eng.scheduled_tick(slot)
+
+    eng.read_now.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_scheduled_tick_continues_past_tag_error():
+    eng = make_engine()
+    bad_slot = make_slot(cycle_multiple=1)
+    good_names = [{'tag_name': 'b', 'model_id': 'm2', 'collection': 'coll'}]
+    bad_slot['srv_name']['tags']['ns=2;i=6'] = {
+        'frequency': 1000,
+        'names': good_names,
+        'active_read': {'enabled': True, 'cycle_multiple': 1},
+    }
+
+    calls = []
+
+    async def fake_read_now(server_id, addr, names=None, server_cfg=None):
+        calls.append(addr)
+        if addr == 'ns=2;i=5':
+            raise RuntimeError('boom')
+
+    eng.read_now = AsyncMock(side_effect=fake_read_now)
+
+    await eng.scheduled_tick(bad_slot)
+
+    assert set(calls) == {'ns=2;i=5', 'ns=2;i=6'}
+    eng.logger.error.assert_called_once()
--- a/tests/unit/engines/test_write.py
+++ b/tests/unit/engines/test_write.py
@@ -0,0 +1,259 @@
+from datetime import datetime
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from asyncua import ua
+
+import ingestor.metrics as metrics
+from ingestor.engines.write import WriteEngine
+
+
+def make_engine(**kwargs):
+    eng = WriteEngine(
+        pool=AsyncMock(),
+        sink=AsyncMock(),
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=AsyncMock(),
+        **kwargs,
+    )
+    eng.emit_metric = AsyncMock()
+    return eng
+
+
+def make_client(node=None):
+    node = node or make_node()
+    client = MagicMock()
+    client.get_node = MagicMock(return_value=node)
+    return client
+
+
+def make_node():
+    node = MagicMock()
+    node.write_value = AsyncMock()
+    return node
+
+
+def base_cmd(**overrides):
+    cmd = {
+        'server_id': 'srv1',
+        'addr': 'ns=2;i=5',
+        'value': 42.0,
+        'data_type': 'float',
+        'tag_type': 'prediction',
+        'model_id': 'm1',
+        'model_name': 'model-1',
+        'timestamp': '2026-07-13T12:00:00Z',
+        'schema': 'sientia_data',
+        'table_name': 'predictions_m1',
+    }
+    cmd.update(overrides)
+    return cmd
+
+
+@pytest.mark.asyncio
+async def test_write_success_writes_datavalue_and_returns_ok():
+    eng = make_engine()
+    node = make_node()
+    client = make_client(node)
+    eng.pool.get.return_value = client
+
+    result = await eng.write(base_cmd())
+
+    client.get_node.assert_called_once_with('ns=2;i=5')
+    node.write_value.assert_awaited_once()
+    dv = node.write_value.await_args.args[0]
+    assert isinstance(dv, ua.DataValue)
+    assert dv.Value.Value == 42.0
+    assert dv.Value.VariantType == ua.VariantType.Float
+    assert result.ok is True
+    assert result.status == 'ok'
+    assert result.opc_status_code == 'Good'
+    eng.sink.update_write_status.assert_awaited_once_with(base_cmd(), result)
+
+
+@pytest.mark.asyncio
+async def test_write_success_emits_metrics():
+    eng = make_engine()
+    eng.pool.get.return_value = make_client()
+
+    await eng.write(base_cmd())
+
+    eng.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_WRITE_TOTAL,
+        tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'model_id': 'm1', 'result': 'success'},
+    )
+    status_call = next(
+        c
+        for c in eng.emit_metric.await_args_list
+        if c.kwargs['metric_object'] is metrics.OPC_WRITE_STATUS_BY_MODEL
+    )
+    assert status_call.kwargs['value'] == 1
+    assert status_call.kwargs['tags'] == {'pod_id': eng.pod_id, 'model_id': 'm1', 'status': 'ok'}
+    latency_call = next(
+        c
+        for c in eng.emit_metric.await_args_list
+        if c.kwargs['metric_object'] is metrics.OPC_WRITE_LATENCY
+    )
+    assert latency_call.kwargs['method'] == 'observe'
+    assert latency_call.kwargs['value'] >= 0
+
+
+@pytest.mark.asyncio
+async def test_write_defaults_carry_statuscode_and_timestamp():
+    eng = make_engine(write_timestamp=True, write_quality=True)
+    node = make_node()
+    eng.pool.get.return_value = make_client(node)
+
+    await eng.write(base_cmd())
+
+    dv = node.write_value.await_args.args[0]
+    assert dv.StatusCode is not None
+    assert dv.StatusCode.is_good()
+    assert dv.SourceTimestamp == datetime.fromisoformat('2026-07-13T12:00:00Z')
+
+
+@pytest.mark.asyncio
+async def test_write_engine_flags_off_omit_statuscode_and_timestamp():
+    eng = make_engine(write_timestamp=False, write_quality=False)
+    node = make_node()
+    eng.pool.get.return_value = make_client(node)
+
+    await eng.write(base_cmd())
+
+    dv = node.write_value.await_args.args[0]
+    assert dv.StatusCode is None
+    assert dv.SourceTimestamp is None
+
+
+@pytest.mark.asyncio
+async def test_per_message_flag_override_disables_timestamp():
+    # Engine default is True, but the command explicitly turns it off.
+    eng = make_engine(write_timestamp=True, write_quality=True)
+    node = make_node()
+    eng.pool.get.return_value = make_client(node)
+
+    await eng.write(base_cmd(write_timestamp=False))
+
+    dv = node.write_value.await_args.args[0]
+    assert dv.SourceTimestamp is None
+    assert dv.StatusCode is not None  # write_quality untouched, stays on
+
+
+@pytest.mark.asyncio
+async def test_write_bad_type_returns_error_without_raising():
+    eng = make_engine()
+
+    result = await eng.write(base_cmd(value='abc', data_type='int'))
+
+    assert result.ok is False
+    assert result.status == 'error'
+    eng.sink.update_write_status.assert_awaited_once()
+    eng.pool.get.assert_not_awaited()
+    eng.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_ERROR_BY_TYPE,
+        tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'error_type': 'conversion'},
+    )
+    eng.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_WRITE_TOTAL,
+        tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'model_id': 'm1', 'result': 'error'},
+    )
+
+
+@pytest.mark.asyncio
+async def test_write_none_data_type_returns_error_without_raising():
+    eng = make_engine()
+
+    result = await eng.write(base_cmd(data_type=None))
+
+    assert result.ok is False
+    assert result.status == 'error'
+    eng.sink.update_write_status.assert_awaited_once()
+    eng.pool.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_write_missing_value_key_returns_error_without_raising():
+    eng = make_engine()
+    cmd = base_cmd()
+    del cmd['value']
+
+    result = await eng.write(cmd)
+
+    assert result.ok is False
+    assert result.status == 'error'
+    eng.sink.update_write_status.assert_awaited_once()
+    eng.pool.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_write_missing_data_type_key_returns_error_without_raising():
+    eng = make_engine()
+    cmd = base_cmd()
+    del cmd['data_type']
+
+    result = await eng.write(cmd)
+
+    assert result.ok is False
+    assert result.status == 'error'
+    eng.sink.update_write_status.assert_awaited_once()
+    eng.pool.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_write_value_raises_maps_to_error_result_not_raised():
+    eng = make_engine()
+    node = make_node()
+    node.write_value.side_effect = RuntimeError('boom')
+    eng.pool.get.return_value = make_client(node)
+
+    result = await eng.write(base_cmd())
+
+    assert result.ok is False
+    assert result.status == 'error'
+    assert result.opc_status_code == 'RuntimeError'
+    eng.sink.update_write_status.assert_awaited_once()
+    eng.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_ERROR_BY_TYPE,
+        tags={'pod_id': eng.pod_id, 'server_name': 'srv1', 'error_type': 'RuntimeError'},
+    )
+
+
+@pytest.mark.asyncio
+async def test_write_session_bad_status_code_error_maps_to_session_bad():
+    from asyncua.ua.uaerrors import BadSessionClosed
+
+    eng = make_engine()
+    node = make_node()
+    node.write_value.side_effect = BadSessionClosed('closed')
+    eng.pool.get.return_value = make_client(node)
+
+    result = await eng.write(base_cmd())
+
+    assert result.ok is False
+    assert result.status == 'session_bad'
+
+
+@pytest.mark.asyncio
+async def test_rate_limit_sleeps_between_back_to_back_writes():
+    eng = make_engine(min_pause_s=0.05)
+    eng.pool.get.return_value = make_client()
+
+    with patch('ingestor.engines.write.asyncio.sleep', new_callable=AsyncMock) as sleep_mock:
+        await eng.write(base_cmd())
+        sleep_mock.assert_not_awaited()
+        await eng.write(base_cmd())
+        sleep_mock.assert_awaited_once()
+        remaining = sleep_mock.await_args.args[0]
+        assert 0 < remaining <= 0.05
+
+
+@pytest.mark.asyncio
+async def test_no_rate_limit_when_min_pause_is_zero():
+    eng = make_engine(min_pause_s=0.0)
+    eng.pool.get.return_value = make_client()
+
+    with patch('ingestor.engines.write.asyncio.sleep', new_callable=AsyncMock) as sleep_mock:
+        await eng.write(base_cmd())
+        await eng.write(base_cmd())
+        sleep_mock.assert_not_awaited()
--- a/tests/unit/io/test_data_sink_status.py
+++ b/tests/unit/io/test_data_sink_status.py
@@ -0,0 +1,89 @@
+from unittest.mock import MagicMock
+
+import pytest
+
+from ingestor.engines.write import WriteResult
+from ingestor.io.data_sink import DataSink
+
+
+def make_sink():
+    postgres = MagicMock()
+    connection = MagicMock()
+    postgres.engine.begin.return_value.__enter__.return_value = connection
+    sink = DataSink(data_manager=MagicMock(), postgres=postgres)
+    return sink, connection
+
+
+@pytest.mark.asyncio
+async def test_update_write_status_targets_model_row():
+    sink, connection = make_sink()
+    cmd = {
+        'schema': 'sientia_data',
+        'table_name': 'pred_m1',
+        'model_id': '1',
+        'timestamp': '2026-07-13T00:00:00Z',
+    }
+    result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='')
+
+    await sink.update_write_status(cmd, result)
+
+    assert connection.execute.call_count == 1
+    stmt, params = connection.execute.call_args[0]
+    sql = str(stmt)
+    assert 'update' in sql.lower() and 'write_status' in sql.lower()
+    assert 'sientia_data.pred_m1' in sql
+    # values must be bound params, never interpolated into the SQL text itself
+    assert params == {'status': 'ok', 'timestamp': '2026-07-13T00:00:00Z', 'model_id': '1'}
+    assert 'ok' not in sql
+    assert '2026-07-13T00:00:00Z' not in sql
+
+
+@pytest.mark.asyncio
+async def test_update_write_status_rejects_bad_schema():
+    sink, connection = make_sink()
+    cmd = {
+        'schema': 'a; DROP TABLE x',
+        'table_name': 'pred_m1',
+        'model_id': '1',
+        'timestamp': '2026-07-13T00:00:00Z',
+    }
+    result = WriteResult(ok=False, status='error', opc_status_code='Bad', message='x')
+
+    with pytest.raises(ValueError):
+        await sink.update_write_status(cmd, result)
+
+    connection.execute.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_update_write_status_rejects_trailing_newline_identifier():
+    sink, connection = make_sink()
+    cmd = {
+        'schema': 'sientia_data\n',
+        'table_name': 'pred_m1',
+        'model_id': '1',
+        'timestamp': '2026-07-13T00:00:00Z',
+    }
+    result = WriteResult(ok=True, status='ok', opc_status_code='Good', message='')
+
+    with pytest.raises(ValueError):
+        await sink.update_write_status(cmd, result)
+
+    connection.execute.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_update_write_status_rejects_bad_table_name():
+    sink, connection = make_sink()
+    cmd = {
+        'schema': 'sientia_data',
+        'table_name': 'pred_m1; DROP TABLE x',
+        'model_id': '1',
+        'timestamp': '2026-07-13T00:00:00Z',
+    }
+    result = WriteResult(ok=False, status='error', opc_status_code='Bad', message='x')
+
+    with pytest.raises(ValueError):
+        await sink.update_write_status(cmd, result)
+
+    connection.execute.assert_not_called()
--- a/tests/unit/proto/init.py
+++ b/tests/unit/proto/init.py
--- a/tests/unit/proto/test_proto.py
+++ b/tests/unit/proto/test_proto.py
@@ -0,0 +1,36 @@
+from ingestor.proto import opc_gateway_pb2, opc_gateway_pb2_grpc
+
+
+def test_read_tag_request_roundtrip():
+    req = opc_gateway_pb2.ReadTagRequest(server_id='s1', addr='ns=2;s=Tag1', data_type='Float')
+
+    data = req.SerializeToString()
+    parsed = opc_gateway_pb2.ReadTagRequest()
+    parsed.ParseFromString(data)
+
+    assert parsed.server_id == 's1'
+    assert parsed.addr == 'ns=2;s=Tag1'
+    assert parsed.data_type == 'Float'
+
+
+def test_read_tag_response_roundtrip():
+    resp = opc_gateway_pb2.ReadTagResponse(
+        value='42.0',
+        opc_status_code='Good',
+        source_timestamp='2026-07-14T00:00:00Z',
+        server_timestamp='2026-07-14T00:00:01Z',
+        persisted=True,
+    )
+
+    data = resp.SerializeToString()
+    parsed = opc_gateway_pb2.ReadTagResponse()
+    parsed.ParseFromString(data)
+
+    assert parsed.value == '42.0'
+    assert parsed.opc_status_code == 'Good'
+    assert parsed.persisted is True
+
+
+def test_grpc_stub_module_loads():
+    assert hasattr(opc_gateway_pb2_grpc, 'OpcGatewayStub')
+    assert hasattr(opc_gateway_pb2_grpc, 'OpcGatewayServicer')
--- a/tests/unit/test_converters.py
+++ b/tests/unit/test_converters.py
@@ -38,3 +38,10 @@ def test_float_and_int():
 def test_invalid_raises_conversion_error(raw, dtype):
    with pytest.raises(ConversionError):
        to_typed_value(raw, dtype)
+
+
+@pytest.mark.parametrize('dtype', [None, 123])
+def test_non_string_data_type_raises_conversion_error(dtype):
+    # contract: never a bare AttributeError/TypeError, always ConversionError
+    with pytest.raises(ConversionError):
+        to_typed_value('5', dtype)