sientia-dataops-opc-ingestor/docs/superpowers/plans/2026-07-13-opc-gateway.md

OPC Gateway Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Evolve sientia-dataops-opc-ingestor into an OPC Gateway that reads by subscription (multi-name fan-out), reads actively (scheduled + on-demand via gRPC), writes OPC via NATS, manages certs via MinIO+Mongo, and emits detailed metrics — following the OPC best practices of SIENTIAPDE-1879.

Architecture: Approach B — engines per capability over a shared OpcConnectionPool (connections keyed by server, not by slot). Input adapters (NatsWriteConsumer, GrpcServer) feed the engines; a DataSink handles egress (Mongo/Kafka reads + Postgres write_status). Coordination (slot lease/heartbeat) is reused from the existing Redis layer.

Tech Stack: Python 3.11, asyncua, nats-py (Core NATS), grpcio/grpcio-tools, prometheus_client, sientia_do dataops-library (Mongo/Redis/MinIO/Postgres), pytest + pytest-asyncio, ruff (line-length 100, single quotes).

Spec: docs/superpowers/specs/2026-07-13-opc-gateway-design.md (read it first — contracts C1–C4, metrics, §13 1879 checklist).

Scope: This plan is the Gateway only (repo sientia-dataops-opc-ingestor). Orchestrator (1974) and Laborious (1975) are parallel tracks with their own plans in their own repos; here they appear only as the contracts C1 (slot v2), C2 (NATS message), C3 (write_status).

Conventions (match the repo):

• Managers/engines extend SientiaMonitoring (sientia_do.observability.sientia_monitoring) → gives self.logger, self.pod_id, await self.emit_metric(...), await self.send_notification_async(...).
• Metrics defined in ingestor/metrics.py via prometheus_client.
• Package stays ingestor/ (do not rename — the Helm health probe and Gitea PYTHON_APP entrypoint depend on ingestor.app). New capabilities go in new subpackages.
• Test command everywhere: python -m pytest <path> -v. Lint: ruff check ingestor tests.

---

File Structure

New subpackages under ingestor/ (one clear responsibility each; keep files < ~500 lines, functions < ~20 lines per 1879):

• ingestor/connection/pool.py — OpcConnectionPool: asyncua clients keyed by server_id, lazy open, reuse, idle-eviction, reconnect.
• ingestor/connection/security.py — build_security(...): Basic256Sha256, secure_channel_timeout = int(session_timeout*1.25), cert file paths.
• ingestor/connection/certs.py — CertProvider: read platform_certificates (Mongo) + download cert/key from MinIO (sientia-dataops bucket) to an emptyDir/tmpfs path.
• ingestor/connection/clock.py — MonotonicClock: single time.monotonic() reference + optional NTP; elapsed(), now_wall().
• ingestor/converters.py — to_variant_value(raw, data_type): safe type coercion (bool-string bug fix + try/except).
• ingestor/engines/subscription.py — SubscriptionEngine: slot v2 → subscription → multi-name fan-out to DataSink.
• ingestor/engines/active_read.py — ActiveReadEngine: scheduled reads + read_now() for on-demand.
• ingestor/engines/write.py — WriteEngine: OPC write with status_code/timestamp/quality flags, rate-limit, error mapping, → DataSink.update_write_status.
• ingestor/io/slots.py — load_slot(redis, slot_id): parse slot v2 (C1) into typed structures.
• ingestor/io/data_sink.py — DataSink: publish_reading(...) (Mongo/Kafka fan-out) + update_write_status(...) (Postgres UPDATE by timestamp+model_id).
• ingestor/adapters/nats_consumer.py — NatsWriteConsumer: subscribe opc.write (queue group gateway-writers) → WriteEngine.
• ingestor/adapters/grpc_server.py — GrpcServer: ReadTag RPC → ActiveReadEngine.read_now.
• ingestor/proto/opc_gateway.proto (+ generated *_pb2.py, *_pb2_grpc.py).
• ingestor/config.py — extend with new env (NATS, gRPC, cert bucket, NTP).
• ingestor/metrics.py — extend with new metrics (§9 of spec).

Reused as-is / lightly touched: ingestor/managers/resource_manager.py (lease/heartbeat), ingestor/managers/data_manager.py (Kafka/Mongo egress — wrapped by DataSink), ingestor/app.py (wire new engines in F4).

Tests mirror source: tests/unit/connection/…, tests/unit/engines/…, tests/unit/io/…, tests/unit/adapters/…, tests/unit/test_converters.py. Add an __init__.py to each new test dir (the repo's existing test dirs use them for collection, e.g. tests/unit/managers/__init__.py).

---

Phase F0 — Foundations (deps, config, metrics, connection pool)

Task 1: Add dependencies

Files:

• Modify: requirements.txt (and pyproject.toml if deps are declared there)

• Step 1: Add libs

Add to requirements.txt:

nats-py>=2.6
grpcio>=1.60
grpcio-tools>=1.60

(asyncua, prometheus_client, sientia_do, pandas already present. Bump sientia_do to the version that ships the generic Mongo methods — feature/SIENTIAPDE-1590 / ≥ current pin.)

• Step 2: Install & verify import

Run: pip install -r requirements.txt && python -c "import nats, grpc; print('ok')" Expected: ok

• Step 3: Commit

git add requirements.txt pyproject.toml
git commit -m "chore(gateway): add nats-py, grpcio deps"

---

Task 2: New metrics (spec §9)

Files:

• Modify: ingestor/metrics.py

• Test: tests/unit/test_metrics.py (extend existing)

• Step 1: Write failing test — assert the new metric objects exist and accept labels.

# tests/unit/test_metrics.py
import ingestor.metrics as m

def test_gateway_metrics_exist():
    m.OPC_WRITE_TOTAL.labels(pod_id='p', server_name='s', model_id='1', result='success')
    m.OPC_WRITE_LATENCY.labels(pod_id='p', server_name='s')
    m.OPC_ACTIVE_READ_TOTAL.labels(pod_id='p', server_name='s', result='success')
    m.OPC_ACTIVE_READ_LATENCY.labels(pod_id='p', server_name='s')
    m.OPC_WRITE_STATUS_BY_MODEL.labels(pod_id='p', model_id='1', status='ok')
    m.OPC_TAG_VALUE.labels(pod_id='p', tag_name='t', model_id='1')
    m.OPC_READ_LAG.labels(pod_id='p', server_name='s')
    m.OPC_ERROR_BY_TYPE.labels(pod_id='p', server_name='s', error_type='BadTimeout')
    m.OPC_DISCONNECTIONS_TOTAL.labels(pod_id='p', server_name='s')

• Step 2: Run — expect FAIL (AttributeError). Run: python -m pytest tests/unit/test_metrics.py -v

• Step 3: Implement — append to ingestor/metrics.py:

# --- Gateway: Write metrics ---
OPC_WRITE_TOTAL = Counter(
    'opc_write_total', 'OPC active write attempts',
    ['pod_id', 'server_name', 'model_id', 'result'],  # result: success|error
)
OPC_WRITE_LATENCY = Histogram(
    'opc_write_latency_seconds', 'OPC active write latency', ['pod_id', 'server_name'],
)
OPC_WRITE_STATUS_BY_MODEL = Gauge(
    'opc_write_status_by_model', 'Last write status per model (1=ok,0=error)',
    ['pod_id', 'model_id', 'status'],
)
# --- Gateway: Active read metrics ---
OPC_ACTIVE_READ_TOTAL = Counter(
    'opc_active_read_total', 'OPC active read attempts',
    ['pod_id', 'server_name', 'result'],
)
OPC_ACTIVE_READ_LATENCY = Histogram(
    'opc_active_read_latency_seconds', 'OPC active read latency', ['pod_id', 'server_name'],
)
OPC_READ_LAG = Gauge(
    'opc_read_lag_seconds', 'Lag between OPC source timestamp and machine time',
    ['pod_id', 'server_name'],
)
OPC_TAG_VALUE = Gauge(
    'opc_tag_value', 'Last value read per tag/model', ['pod_id', 'tag_name', 'model_id'],
)
# --- Gateway: connection/error metrics ---
OPC_DISCONNECTIONS_TOTAL = Counter(
    'opc_disconnections_total', 'OPC disconnections', ['pod_id', 'server_name'],
)
OPC_ERROR_BY_TYPE = Counter(
    'opc_error_by_type_total', 'OPC errors by type', ['pod_id', 'server_name', 'error_type'],
)

• Step 4: Run — expect PASS. python -m pytest tests/unit/test_metrics.py -v

• Step 5: Commit — git commit -am "feat(gateway): add write/active-read/error metrics"

---

Task 3: Safe type converter (1879 BUG-1 & BUG-2)

Files:

• Create: ingestor/converters.py

• Test: tests/unit/test_converters.py

• Step 1: Write failing tests (the bool-string bug is the whole point):

# tests/unit/test_converters.py
import pytest
from ingestor.converters import to_typed_value, ConversionError

@pytest.mark.parametrize('raw,expected', [
    ('false', False), ('False', False), ('0', False), ('no', False),
    ('off', False), ('', False), ('true', True), ('1', True), (True, True),
])
def test_bool_from_json_string(raw, expected):
    assert to_typed_value(raw, 'boolean') is expected

def test_float_and_int():
    assert to_typed_value('42.5', 'float') == 42.5
    assert to_typed_value('7', 'int') == 7

@pytest.mark.parametrize('raw,dtype', [
    (None, 'int'), ('abc', 'float'), ('123.123', 'int'), (None, 'float'),
])
def test_invalid_raises_conversion_error(raw, dtype):
    with pytest.raises(ConversionError):
        to_typed_value(raw, dtype)

• Step 2: Run — expect FAIL (module missing). python -m pytest tests/unit/test_converters.py -v

• Step 3: Implement ingestor/converters.py:

"""Safe coercion of JSON-transported values into typed OPC write values.

Fixes 1879 BUG-1 (bool('false') is True in Python) and BUG-2 (uncaught
TypeError/ValueError from int/float on bad input).
"""

_FALSE_STRINGS = {'false', '0', 'no', 'off', ''}


class ConversionError(ValueError):
    """Raised when a raw value cannot be coerced to the requested type."""


def _to_bool(raw) -> bool:
    if isinstance(raw, bool):
        return raw
    if isinstance(raw, (int, float)):
        return bool(raw)
    return str(raw).strip().lower() not in _FALSE_STRINGS


def to_typed_value(raw, data_type: str):
    """Coerce raw (often a JSON string) to the given data_type.

    data_type: 'boolean' | 'int' | 'float' | 'double' | 'string'.
    Raises ConversionError on any invalid input (never a bare TypeError).
    """
    dtype = data_type.strip().lower()
    try:
        if dtype in ('bool', 'boolean'):
            return _to_bool(raw)
        if dtype in ('int', 'integer'):
            if isinstance(raw, str) and ('.' in raw or raw.strip() == ''):
                raise ConversionError(f'invalid int: {raw!r}')
            return int(raw)
        if dtype in ('float', 'double', 'real'):
            return float(raw)
        if dtype in ('str', 'string'):
            return str(raw)
    except (TypeError, ValueError) as e:
        raise ConversionError(f'cannot convert {raw!r} to {data_type}: {e}') from e
    raise ConversionError(f'unknown data_type: {data_type!r}')

• Step 4: Run — expect PASS. python -m pytest tests/unit/test_converters.py -v

• Step 5: Commit — git commit -am "feat(gateway): safe type converter (1879 BUG-1/BUG-2)"

---

Task 4: Monotonic clock helper (1879 §2)

Files:

• Create: ingestor/connection/clock.py, ingestor/connection/__init__.py

• Test: tests/unit/connection/test_clock.py

• Step 1: Failing test

# tests/unit/connection/test_clock.py
from ingestor.connection.clock import MonotonicClock

def test_elapsed_is_non_negative_and_monotonic():
    clk = MonotonicClock()
    a = clk.elapsed()
    b = clk.elapsed()
    assert a >= 0 and b >= a  # never negative even if wall clock jumps back

• Step 2: Run — expect FAIL.

• Step 3: Implement ingestor/connection/clock.py:

import time


class MonotonicClock:
    """Single monotonic reference for latency/timeout math (1879 §2).

    Wall-clock (`now_wall`) is only for human-facing timestamps; all elapsed/
    timeout math must use `elapsed()` so NTP re-sync can't produce negatives.
    """

    def __init__(self):
        self._start = time.monotonic()

    def elapsed(self) -> float:
        return time.monotonic() - self._start

    @staticmethod
    def now_wall() -> float:
        return time.time()

• Step 4: Run — expect PASS.
• Step 5: Commit — git commit -am "feat(gateway): monotonic clock helper (1879)"

Note (calibration): NTP sync + wall-clock quality downgrade (Good_Local_Override 0xD8) are flag-driven and belong to the WriteEngine (Task 11). Leave a # ponytail: marker here noting NTP option is deferred until a real clock-drift case appears.

---

Task 5: Security builder (1879 §1, §3)

Files:

• Create: ingestor/connection/security.py

• Test: tests/unit/connection/test_security.py

• Step 1: Failing test — assert channel timeout is 1.25× session and policy is Sha256.

# tests/unit/connection/test_security.py
from unittest.mock import AsyncMock, MagicMock
import pytest
from ingestor.connection.security import apply_security, SESSION_TIMEOUT_MS

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

• Step 2: Run — expect FAIL.

• Step 3: Implement ingestor/connection/security.py:

from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256

SESSION_TIMEOUT_MS = 600_000  # 10 min; tune per server (Samarco: long sessions degrade)


async def apply_security(client, server_uri, cert_path, private_key_path, server_cert_path):
    """Apply Basic256Sha256 + certificate auth (1879 §1) and set channel > session
    lifetime (1879 §3: secure_channel_timeout = session_timeout * 1.25)."""
    client.application_uri = server_uri
    await client.set_security(
        SecurityPolicyBasic256Sha256,
        certificate=cert_path,
        private_key=private_key_path,
        server_certificate=server_cert_path,
    )
    client.session_timeout = SESSION_TIMEOUT_MS
    client.secure_channel_timeout = int(SESSION_TIMEOUT_MS * 1.25)

• Step 4: Run — expect PASS.
• Step 5: Commit — git commit -am "feat(gateway): Basic256Sha256 + channel>session lifetime (1879)"

---

Task 6: Certificate provider (MinIO + Mongo)

Files:

• Create: ingestor/connection/certs.py

• Test: tests/unit/connection/test_certs.py

• Step 1: Failing test — reads platform_certificates doc, downloads both objects, returns local paths.

# tests/unit/connection/test_certs.py
from unittest.mock import AsyncMock
import pytest
from ingestor.connection.certs import CertProvider

@pytest.mark.asyncio
async def test_fetch_returns_local_paths(tmp_path):
    mongo = AsyncMock()
    mongo.find_one.return_value = {
        'client_cert': {'minio_object_name': 'platform/certificates/client_cert_c.pem'},
        'private_key': {'minio_object_name': 'platform/certificates/private_key_k.pem'},
    }
    minio = AsyncMock()
    provider = CertProvider(mongo, minio, cache_dir=str(tmp_path), bucket='sientia-dataops')
    cert, key = await provider.fetch()
    assert cert.endswith('c.pem') and key.endswith('k.pem')
    assert minio.download_file.await_count == 2

• Step 2: Run — expect FAIL.

• Step 3: Implement ingestor/connection/certs.py — read the single platform_certificates doc via mongo.find_one({}, 'platform_certificates'), await minio.download_file(bucket, object_name, local_path) into cache_dir (must be a writable emptyDir/tmpfs — root FS is read-only in the pod, spec §12). Return (cert_path, key_path). Keep it < 20 lines.

• Step 4: Run — expect PASS.

• Step 5: Commit — git commit -am "feat(gateway): cert provider from MinIO+Mongo (sientia-dataops)"

Confirm the exact MongoDBRepository.find_one/MinioRepository.download_file signatures against the pinned sientia_do; adapt the calls if they differ.

---

Task 7: OpcConnectionPool

Files:

• Create: ingestor/connection/pool.py

• Test: tests/unit/connection/test_pool.py

• Step 1: Failing tests — (a) same server_id returns the same client (reuse); (b) a connect failure increments OPC_CONNECTIONS_FAILED and re-raises.

# tests/unit/connection/test_pool.py
from unittest.mock import AsyncMock, patch
import pytest
from ingestor.connection.pool import OpcConnectionPool

@pytest.mark.asyncio
async def test_get_reuses_connection_per_server(monkeypatch):
    pool = OpcConnectionPool(cert_provider=AsyncMock(), metadata={}, logger=AsyncMock())
    fake_client = AsyncMock()
    with patch('ingestor.connection.pool.Client', return_value=fake_client):
        c1 = await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
        c2 = await pool.get('srv', {'url': 'opc.tcp://h:4840', 'use_security': False})
    assert c1 is c2
    assert fake_client.connect.await_count == 1

• Step 2: Run — expect FAIL.

• Step 3: Implement ingestor/connection/pool.py — OpcConnectionPool keyed by server_id:

  • async get(server_id, server_cfg): if a live client exists, return it; else create Client(url, timeout=10), if use_security call apply_security(...) with certs from cert_provider.fetch(), await client.connect(), store, emit OPC_CONNECTIONS_TOTAL/OPC_CONNECTION_STATUS. On failure emit OPC_CONNECTIONS_FAILED, drop entry, re-raise.
  • async release_idle(max_idle_s): disconnect clients unused past TTL (track last-use via MonotonicClock), emit OPC_DISCONNECTIONS_TOTAL.
  • async close_all().
  • Reuse the existing OpcManager.disconnection_fallback pattern for robust disconnect.

• Step 4: Run — expect PASS.

• Step 5: Commit — git commit -am "feat(gateway): OpcConnectionPool (per-server, reuse, idle-evict)"

---

Phase F1 — Subscription reads with multi-name fan-out (contract C1)

Task 8: Slot v2 loader

Files:

• Create: ingestor/io/slots.py, ingestor/io/__init__.py

• Test: tests/unit/io/test_slots.py

• Step 1: Failing test — parse the C1 structure incl. names[] and active_read.

# tests/unit/io/test_slots.py
from ingestor.io.slots import parse_slot

RAW = {'server1': {'server_id': '1', 'url': 'opc.tcp://h:4840', 'use_security': True,
    'subscription_period_ms': 1000, 'tags': {'ns=2;i=2': {'frequency': '15000',
    'active_read': {'enabled': True, 'cycle_multiple': 3},
    'names': [{'collection': 'raw_x', 'tag_name': 'Counter', 'model_name': 'm1', 'model_id': '1'},
              {'collection': 'raw_y', 'tag_name': 'Contador', 'model_name': 'm2', 'model_id': '2'}]}}}}

def test_parse_slot_multi_name():
    slot = parse_slot(RAW)
    tag = slot['server1']['tags']['ns=2;i=2']
    assert len(tag['names']) == 2
    assert tag['active_read']['cycle_multiple'] == 3

• Step 2–4: implement parse_slot(raw) (validate shape, default active_read={'enabled': False}, ensure names is a list); run to green.
• Step 5: Commit — git commit -am "feat(gateway): slot v2 loader (contract C1)"

---

Task 9: DataSink.publish_reading (Mongo/Kafka fan-out)

Files:

• Create: ingestor/io/data_sink.py

• Test: tests/unit/io/test_data_sink_publish.py

• Step 1: Failing test — one reading with N names publishes N documents (one per collection), each carrying tag_name/model_id.

# tests/unit/io/test_data_sink_publish.py
from unittest.mock import AsyncMock
import pytest
from ingestor.io.data_sink import DataSink

@pytest.mark.asyncio
async def test_publish_reading_fans_out_per_name():
    dm = AsyncMock()
    sink = DataSink(data_manager=dm, postgres=AsyncMock(), logger=AsyncMock())
    names = [{'collection': 'raw_x', 'tag_name': 'Counter', 'model_id': '1', 'model_name': 'm1'},
             {'collection': 'raw_y', 'tag_name': 'Contador', 'model_id': '2', 'model_name': 'm2'}]
    await sink.publish_reading(tag='ns=2;i=2', value=42, timestamp='2026-07-13T00:00:00Z', names=names)
    assert dm.publish.await_count == 2

• Step 2–4: implement DataSink.publish_reading(tag, value, timestamp, names) — loop names, build {tag, name: tag_name, model_id, timestamp, value}, call self.data_manager.publish(collection, doc) (reuses existing Kafka/Mongo egress). Emit OPC_TAG_VALUE. Run to green.
• Step 5: Commit — git commit -am "feat(gateway): DataSink.publish_reading multi-name fan-out"

---

Task 10: SubscriptionEngine (fan-out on datachange)

Files:

• Create: ingestor/engines/subscription.py, ingestor/engines/__init__.py

• Test: tests/unit/engines/test_subscription.py

• Step 1: Failing test — a datachange for an address with 2 names calls DataSink.publish_reading once with both names; resets cycle/non-receive counters; records OPC_READ_LAG.

# tests/unit/engines/test_subscription.py
from unittest.mock import AsyncMock, MagicMock
import pytest
from ingestor.engines.subscription import SubscriptionEngine

@pytest.mark.asyncio
async def test_datachange_fans_out():
    sink = AsyncMock()
    eng = SubscriptionEngine(pool=AsyncMock(), sink=sink, logger=AsyncMock(), metadata={})
    eng.nodes['ns=2;i=2'] = {'names': [{'tag_name': 'a'}, {'tag_name': 'b'}],
                             'cycle_rule': {'cycle_increment': 1, 'cycle_count': 3}}
    node = MagicMock(); node.__str__ = lambda self: 'ns=2;i=2'
    data = MagicMock()
    data.monitored_item.Value.Value.Value = 42
    await eng.datachange_notification(node, 42, data)
    assert sink.publish_reading.await_count == 1
    _, kwargs = sink.publish_reading.call_args
    assert len(kwargs['names']) == 2

• Step 2: Run — expect FAIL.

• Step 3: Implement SubscriptionEngine by adapting ingestor/managers/opc_manager.py:

  • Subscriptions created via pool.get(...) + client.create_subscription(period, self) (engine is the callback handler).
  • datachange_notification extracts value + SourceTimestamp, computes lag (OPC_READ_LAG = wall_now − source_ts) and calls await self.sink.publish_reading(tag, value, ts, names=self.nodes[tag]['names']) (fan-out — replaces the old per-topics loop).
  • Keep check_cycles/check_opc_listenning behavior.
  • Slot lease/heartbeat: reuse ResourceManager (unchanged); the engine consumes parse_slot(...) output.

• Step 4: Run — expect PASS.

• Step 5: Commit — git commit -am "feat(gateway): SubscriptionEngine multi-name fan-out (C1)"

---

Phase F2 — Active write via NATS (contracts C2, C3)

Task 11: WriteEngine

Files:

• Create: ingestor/engines/write.py

• Test: tests/unit/engines/test_write.py

• Step 1: Failing tests — (a) writes a DataValue with StatusCode Good + SourceTimestamp when flags on; (b) a converter failure returns an error result (not an exception) and increments OPC_ERROR_BY_TYPE; (c) rate-limit enforces a min pause between writes.

# tests/unit/engines/test_write.py
from unittest.mock import AsyncMock, MagicMock
import pytest
from ingestor.engines.write import WriteEngine, WriteResult

@pytest.mark.asyncio
async def test_write_ok_sets_status_and_timestamp():
    node = AsyncMock()
    client = MagicMock(); client.get_node.return_value = node
    pool = AsyncMock(); pool.get.return_value = client
    eng = WriteEngine(pool=pool, sink=AsyncMock(), logger=AsyncMock(), metadata={},
                      write_timestamp=True, write_quality=True, min_pause_s=0)
    res = await eng.write({'server_id': 's', 'addr': 'ns=2;i=5', 'value': '42.5',
        'data_type': 'float', 'model_id': '1', 'timestamp': '2026-07-13T00:00:00Z',
        'schema': 'sientia_data', 'table_name': 't'})
    assert isinstance(res, WriteResult) and res.ok
    node.write_value.assert_awaited()  # DataValue passed

@pytest.mark.asyncio
async def test_write_bad_type_returns_error_not_raise():
    client = MagicMock(); client.get_node.return_value = AsyncMock()
    pool = AsyncMock(); pool.get.return_value = client
    eng = WriteEngine(pool=pool, sink=AsyncMock(), logger=AsyncMock(), metadata={}, min_pause_s=0)
    res = await eng.write({'server_id': 's', 'addr': 'ns=2;i=5', 'value': 'abc',
        'data_type': 'int', 'model_id': '1', 'timestamp': 't', 'schema': 'x', 'table_name': 't'})
    assert not res.ok and res.status == 'error'

• Step 2: Run — expect FAIL.

• Step 3: Implement ingestor/engines/write.py:

  • WriteResult dataclass: ok: bool, status: str (ok|error|session_bad), opc_status_code: str, message: str.
  • Per-message flags (C2): write_timestamp/write_quality arrive on each NATS message; resolve as cmd.get('write_timestamp', self.write_timestamp) (per-message wins; config is the fallback default). Do not hard-code engine-global-only behavior.
  • async write(cmd): rate-limit (min pause via MonotonicClock); value = to_typed_value(cmd['value'], cmd['data_type']) inside try/except → on ConversionError return error WriteResult + OPC_ERROR_BY_TYPE; build ua.DataValue(ua.Variant(value, ua_type), StatusCode_=ua.StatusCode(ua.StatusCodes.Good) if write_quality else None, SourceTimestamp=parse(ts) if write_timestamp else None); node = client.get_node(addr); await node.write_value(dv); map OPC exceptions → error_type for OPC_ERROR_BY_TYPE; emit OPC_WRITE_TOTAL/OPC_WRITE_LATENCY/OPC_WRITE_STATUS_BY_MODEL; then await self.sink.update_write_status(cmd, res).
  • Quality downgrade Good_Local_Override (0xD8) when a clock problem is detected — leave a # ponytail: marker; wire only when NTP/clock detection is added.

• Step 4: Run — expect PASS.

• Step 5: Commit — git commit -am "feat(gateway): WriteEngine (status_code/timestamp/flags, safe converter, rate-limit)"

---

Task 12: DataSink.update_write_status (Postgres, contract C3)

Files:

• Modify: ingestor/io/data_sink.py

• Test: tests/unit/io/test_data_sink_status.py

• Step 1: Failing test — issues an UPDATE <schema>.<table> SET write_status=... WHERE timestamp=... AND model_id=....

# tests/unit/io/test_data_sink_status.py
from unittest.mock import AsyncMock
import pytest
from ingestor.io.data_sink import DataSink
from ingestor.engines.write import WriteResult

@pytest.mark.asyncio
async def test_update_write_status_targets_model_row():
    pg = AsyncMock()
    sink = DataSink(data_manager=AsyncMock(), postgres=pg, logger=AsyncMock())
    cmd = {'schema': 'sientia_data', 'table_name': 'pred_m1',
           'model_id': '1', 'timestamp': '2026-07-13T00:00:00Z'}
    await sink.update_write_status(cmd, WriteResult(ok=True, status='ok', opc_status_code='Good', message=''))
    assert pg.execute.await_count == 1
    sql = pg.execute.call_args[0][0]
    assert 'update' in sql.lower() and 'write_status' in sql.lower()

• Step 2–4: implement update_write_status(cmd, result) — parametrized UPDATE on cmd['schema'].cmd['table_name'] filtered by timestamp+model_id; use the sientia_do Postgres client (self.postgres). Never f-string the values (SQL injection) — bind params; the identifiers (schema/table) are validated against [A-Za-z0-9_]+. Run to green.
• Step 5: Commit — git commit -am "feat(gateway): DataSink.update_write_status Postgres (C3)"

Cross-repo note (1975): the column only appears on new per-model tables once Laborious's predictions DataFrame includes write_status; existing tables need the Flyway backfill ALTER TABLE ... ADD COLUMN IF NOT EXISTS write_status. Both are 1975/migrations work — this task only writes the value.

---

Task 13: NatsWriteConsumer (contract C2)

Files:

• Create: ingestor/adapters/nats_consumer.py, ingestor/adapters/__init__.py

• Test: tests/unit/adapters/test_nats_consumer.py

• Step 1: Failing test — an incoming NATS message is JSON-decoded and dispatched to WriteEngine.write; malformed JSON is logged and dropped (no raise).

# tests/unit/adapters/test_nats_consumer.py
import json
from unittest.mock import AsyncMock, MagicMock
import pytest
from ingestor.adapters.nats_consumer import NatsWriteConsumer

@pytest.mark.asyncio
async def test_message_dispatched_to_write_engine():
    we = AsyncMock()
    consumer = NatsWriteConsumer(nats_url='nats://x:4222', write_engine=we, logger=AsyncMock())
    msg = MagicMock(); msg.data = json.dumps({'server_id': 's', 'addr': 'ns=2;i=5'}).encode()
    await consumer._handle(msg)
    we.write.assert_awaited_once()

@pytest.mark.asyncio
async def test_bad_json_does_not_raise():
    consumer = NatsWriteConsumer(nats_url='x', write_engine=AsyncMock(), logger=AsyncMock())
    msg = MagicMock(); msg.data = b'not-json'
    await consumer._handle(msg)  # must not raise

• Step 2: Run — expect FAIL.

• Step 3: Implement NatsWriteConsumer:

  • async start(): self.nc = await nats.connect(self.nats_url); await self.nc.subscribe('opc.write', queue='gateway-writers', cb=self._handle) (queue group → exactly one instance handles each message).
  • async _handle(msg): json.loads in try/except (drop + log on error); await self.write_engine.write(payload) (fire-and-forget — no reply).
  • async stop(): drain/close.

• Step 4: Run — expect PASS.

• Step 5: Commit — git commit -am "feat(gateway): NatsWriteConsumer (Core NATS, queue group) (C2)"

---

Phase F3 — Active reads: scheduled + on-demand gRPC (contract C4)

Task 14: ActiveReadEngine

Files:

• Create: ingestor/engines/active_read.py

• Test: tests/unit/engines/test_active_read.py

• Step 1: Failing test — read_now(server_id, addr) reads via pool, persists via DataSink.publish_reading, and returns value+status+timestamps.

# tests/unit/engines/test_active_read.py
from unittest.mock import AsyncMock, MagicMock
import pytest
from ingestor.engines.active_read import ActiveReadEngine

@pytest.mark.asyncio
async def test_read_now_persists_and_returns():
    dv = MagicMock(); dv.Value.Value = 42
    node = AsyncMock(); node.read_data_value.return_value = dv
    client = MagicMock(); client.get_node.return_value = node
    pool = AsyncMock(); pool.get.return_value = client
    sink = AsyncMock()
    eng = ActiveReadEngine(pool=pool, sink=sink, logger=AsyncMock(), metadata={})
    res = await eng.read_now('s', 'ns=2;i=2', names=[{'collection': 'c', 'tag_name': 't', 'model_id': '1'}])
    assert res['value'] == 42 and res['persisted'] is True
    sink.publish_reading.assert_awaited_once()

• Step 2: Run — expect FAIL.
• Step 3: Implement — read_now(...) (read via node.read_data_value(), emit OPC_ACTIVE_READ_*, persist, return dict) and scheduled_tick(slot) (iterate tags with active_read.enabled, read every cycle_multiple ticks, persist). Run to green.
• Step 5: Commit — git commit -am "feat(gateway): ActiveReadEngine (scheduled + read_now)"

---

Task 15: gRPC proto + codegen

Files:

• Create: ingestor/proto/opc_gateway.proto

• Generate: ingestor/proto/opc_gateway_pb2.py, _pb2_grpc.py

• Step 1: Write proto (contract C4):

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;
}

• Step 2: Generate stubs

Run: python -m grpc_tools.protoc -I ingestor/proto --python_out=ingestor/proto --grpc_python_out=ingestor/proto ingestor/proto/opc_gateway.proto Expected: opc_gateway_pb2.py and opc_gateway_pb2_grpc.py created. Fix the generated import to be package-relative if needed.

• Step 3: Commit — git commit -am "feat(gateway): gRPC proto + generated stubs (C4)"

---

Task 16: GrpcServer

Files:

• Create: ingestor/adapters/grpc_server.py

• Test: tests/unit/adapters/test_grpc_server.py

• Step 1: Failing test — ReadTag delegates to ActiveReadEngine.read_now and maps the dict to the response message.

# tests/unit/adapters/test_grpc_server.py
from unittest.mock import AsyncMock
import pytest
from ingestor.adapters.grpc_server import ReadTagServicer
from ingestor.proto import opc_gateway_pb2 as pb

@pytest.mark.asyncio
async def test_readtag_maps_response():
    eng = AsyncMock()
    eng.read_now.return_value = {'value': '42', 'opc_status_code': 'Good',
        'source_timestamp': 'ts', 'server_timestamp': 'ts2', 'persisted': True}
    servicer = ReadTagServicer(active_read=eng, slot_lookup=AsyncMock())
    resp = await servicer.ReadTag(pb.ReadTagRequest(server_id='s', addr='ns=2;i=2'), None)
    assert resp.value == '42' and resp.persisted is True

• Step 2–4: implement ReadTagServicer (async grpc.aio) + GrpcServer.start()/stop() binding grpc.aio.server() on the configured port; look up names for the address from the slot layer (or accept none). Run to green.
• Step 5: Commit — git commit -am "feat(gateway): gRPC server ReadTag (C4)"

---

Phase F4 — Wiring, health, deploy, hardening

Task 17: Config additions

Files:

• Modify: ingestor/config.py (or wherever env is read), .env.example

• Test: tests/unit/test_config.py

• Step 1–4 (TDD): add env: NATS_URL, NATS_WRITE_SUBJECT (default opc.write), GRPC_PORT (default 50051), GRPC_ENABLED, CERT_BUCKET (default sientia-dataops), CERT_CACHE_DIR (default /tmp/opc-certs), WRITE_TIMESTAMP/WRITE_QUALITY/WRITE_MIN_PAUSE_S, NTP_ENABLED. Test defaults. Run to green.

• Step 5: Commit — git commit -am "feat(gateway): config for NATS/gRPC/certs/write flags"

---

Task 18: Wire engines + adapters into app

Files:

• Modify: ingestor/app.py, ingestor/ingestor.py

• Test: tests/unit/test_app.py (extend)

• Step 1: Failing test — app startup constructs pool + engines + adapters and sets APP_UP=1; shutdown stops adapters, closes pool, sets APP_UP=0.

• Step 2: Run — expect FAIL.

• Step 3: Implement — in startup: build OpcConnectionPool, DataSink, SubscriptionEngine, ActiveReadEngine, WriteEngine, start NatsWriteConsumer and (if GRPC_ENABLED) GrpcServer; register them in the main loop and signal handlers. Reuse the existing start_http_server(HTTP_METRICS_PORT) for /metrics.

• Step 4: Run — expect PASS.

• Step 5: Commit — git commit -am "feat(gateway): wire pool+engines+adapters into app"

---

Task 19: Health probe alignment (spec §9/§10)

Files:

• Modify: ingestor/app.py (ensure APP_UP is set on the :9090/metrics endpoint), deploy values (values/opc-gateway.yaml if present in this repo; otherwise note for the Helm repo).

• Step 1: confirm app_up{pod_id=...} 1 is exposed on :9090/metrics (already emitted; verify the probe string curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1' matches).

• Step 2: if a values/opc-gateway.yaml lives here, set the exec probe + waitForDependencies (Gitea, MongoDB, Redis, Postgres, MinIO, NATS) + ServiceMonitor label release: kube-prometheus-stack. Otherwise add a docs/deploy-notes.md capturing these for the Helm repo.

• Step 3: Commit — git commit -am "chore(gateway): app_up health probe + deploy notes"

---

Task 20: Dockerfile / entrypoint

Files:

• Modify: Dockerfile (or note for sientia-container-images)

• Step 1: ensure the image runs ingestor.app, non-root, readOnlyRootFilesystem-compatible (writable /tmp for CERT_CACHE_DIR); pin base image; no latest. Hadolint clean.

• Step 2: Run hadolint Dockerfile (or note).

• Step 3: Commit — git commit -am "chore(gateway): dockerfile entrypoint + read-only-fs cert cache"

---

Task 21: Full test + lint sweep and 1879 checklist verification

Files: all

• Step 1: python -m pytest tests -v → all green.
• Step 2: ruff check ingestor tests → clean.
• Step 3: Walk spec §13 (1879) and check each box in code: Sha256, monotonic, status_code+timestamp, timestamp/quality flags, rate-limit, BUG-1/BUG-2, channel>session, OPC error typing, file/function size. Add any missing micro-test.
• Step 4: Commit — git commit -am "test(gateway): full sweep + 1879 checklist verification"

---

Deferred / out of this plan (tracked in spec §17)

• NATS provisioning in air-gapped Helm (platform).
• Moving 1569 cert write path to sientia-dataops (platform + 1569 app).
• gRPC caller identity / auth model (Vitor) → may add auth to GrpcServer later.
• Flyway write_status column migration + Laborious DataFrame change (1975 plan).
• JetStream durability (revisit if Core NATS loss becomes real).

Open validation notes for the implementer

• Verify exact sientia_do signatures (MongoDBRepository.find_one, MinioRepository.download_file, Postgres execute) against the pinned version; adapt calls.
• asyncua ua.DataValue/ua.Variant/StatusCode construction: confirm the constructor kwargs on the pinned asyncua (1.1.5) before Task 11.
• The functional test against sientia-dataops-opc_simulator (sibling repo) is a good end-to-end smoke once F0–F3 land.