SIENTIAPDE-2202

SIENTIAPDE-2202 Add gateway end-to-end test suite (e2e/)

End-to-end suite at the repo root, following the Laborious/Orchestrator
pattern: numbered scenarios per topic in e2e/scenarios.md (1:1 with
test_scenario_* functions), scenario inputs as JSON, Postgres DDL mirrored
from production, `integration` marker, excluded from the default pytest run.

Every dependency runs for real: MongoDB, Redis, PostgreSQL, MinIO and NATS
via testcontainers, OPC UA servers in process (asyncua). Pods are built from
the environment like ingestor/app.py and driven one loop iteration at a time.

69 scenarios across 7 topics: slot leasing and failover, subscription reads,
NATS writes and write_status (incl. the SIENTIAPDE-2202 timestamp/quality
matrix, checked on the server and on the wire), gRPC ReadTag, active reads,
client certificates/security, pod lifecycle.

Replaces tests/functional (legacy slot v1 tests, fully commented out) and
tests/functional/test_write_e2e.py (now scenarios 3.1.x).

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

e2e/README.md

@@ -0,0 +1,48 @@
+# OPC Gateway end-to-end tests
+
+End-to-end tests run every external dependency for real (MongoDB, Redis, PostgreSQL, MinIO and
+NATS via testcontainers; OPC UA servers in process with asyncua) and build each gateway pod from
+the environment exactly like `ingestor/app.py` does.
+
+## Requirements
+
+- Docker (for testcontainers), with the reaper disabled once per machine:
+  `echo "ryuk.disabled=true" >> ~/.testcontainers.properties` (see [scenarios.md](scenarios.md)).
+- Python dev dependencies: `pip install -r requirements.txt -r requirements-dev.txt`
+
+## Run locally
+
+```bash
+source ./venv/bin/activate
+pytest e2e/ -o testpaths=e2e -m integration
+```
+
+One topic, or stop on first failure:
+
+```bash
+pytest e2e/test_writes.py -o testpaths=e2e -m integration
+pytest e2e/ -o testpaths=e2e -m integration -x
+```
+
+The default `pytest` run only collects `tests/` (`testpaths`), so the E2E suite never runs there.
+
+## Layout
+
+| Path | Content |
+|---|---|
+| `conftest.py` | Containers (session), environment + state reset (per test), factories: `opc_servers`, `pods`, `put_slot`, `publish_command`, `client_certificate`, `opc_wire` |
+| `helpers.py` | `Pod` (one gateway pod, `tick()` = one main loop iteration), `load_scenario_input`, `wait_until`/`stays_false`, Mongo/Postgres/Redis queries, DataValue mask |
+| `opc_server.py` | In-process OPC UA server (`ns=2;s=<name>` variables, restart, secured mode, received writes) |
+| `db_schema.sql` | `sientia_data.predictions` DDL, mirrored from production |
+| `scenario_inputs/` | Slots (v2), connector and write command used by the scenarios |
+| `test_*.py` | One file per topic, `test_scenario_<topic>_<group>_<n>_*` |
+
+## Coverage (separate from unit tests)
+
+```bash
+COVERAGE_FILE=.coverage.e2e pytest e2e/ -o testpaths=e2e -m integration --cov=ingestor --cov-branch
+```
+
+## Scenario catalog
+
+See [scenarios.md](scenarios.md) for the numbered scenarios and which test implements each one.

e2e/conftest.py

@@ -0,0 +1,355 @@
+"""Pytest configuration and fixtures for the gateway E2E tests.
+
+Every external dependency runs for real: MongoDB, Redis, PostgreSQL, MinIO and NATS
+via testcontainers (session-scoped), OPC UA servers in process (``e2e/opc_server.py``,
+function-scoped). Gateway code under ``ingestor/**`` is not mocked: pods are built from
+the environment exactly like ``ingestor/app.py`` builds them (``e2e/helpers.Pod``).
+State is reset before every test.
+"""
+
+import io
+import json
+import os
+from pathlib import Path
+
+import nats
+import pytest
+import pytest_asyncio
+from asyncua import ua
+from asyncua.client import ua_client
+from minio import Minio
+from pymongo import MongoClient
+from redis import Redis
+from sqlalchemy import create_engine, text
+from testcontainers.community.minio import MinioContainer
+from testcontainers.community.mongodb import MongoDbContainer
+from testcontainers.community.postgres import PostgresContainer
+from testcontainers.community.redis import RedisContainer
+from testcontainers.core.container import DockerContainer
+from testcontainers.core.wait_strategies import LogMessageWaitStrategy
+
+from e2e.helpers import CERT_BUCKET, Pod
+from e2e.opc_server import OpcTestServer, free_port, make_certificate
+
+DB_SCHEMA_SQL_PATH = Path(__file__).parent / 'db_schema.sql'
+MONGO_USER = MONGO_PASSWORD = 'sientia'
+MONGO_DATABASE = 'sientia_e2e'
+WRITE_SUBJECT = 'opc.write'
+
+
+# --------------------------------------------------------------------------------------
+# Session infrastructure
+# --------------------------------------------------------------------------------------
+
+
+@pytest.fixture(scope='session')
+def mongo_container():
+    """MongoDB with root credentials: the gateway always connects as user:password@url."""
+    mongo = MongoDbContainer('mongo:7', username=MONGO_USER, password=MONGO_PASSWORD)
+    mongo.start()
+    yield mongo
+    mongo.stop()
+
+
+@pytest.fixture(scope='session')
+def redis_container():
+    redis = RedisContainer('redis:7-alpine')
+    redis.start()
+    yield redis
+    redis.stop()
+
+
+@pytest.fixture(scope='session')
+def postgres_container():
+    postgres = PostgresContainer(
+        'postgres:15', username='sientia', password='sientia', dbname='sientia'
+    )
+    postgres.start()
+    yield postgres
+    postgres.stop()
+
+
+@pytest.fixture(scope='session')
+def minio_container():
+    """
+    MinIO holding the OPC client certificates. Same Bitnami "legacy" image as the
+    Laborious E2E suite and production: Docker Hub no longer serves ``minio/minio``.
+    """
+    minio = MinioContainer(image='docker.io/bitnamilegacy/minio:2025.7.23-debian-12-r3')
+    minio.with_env('MINIO_ROOT_USER', minio.access_key)
+    minio.with_env('MINIO_ROOT_PASSWORD', minio.secret_key)
+    minio.with_kwargs(user='root')
+    minio.start()
+    yield minio
+    minio.stop()
+
+
+NATS_IMAGE = 'nats:2.10-alpine'
+
+
+def nats_server_container(host_port: int | None = None) -> DockerContainer:
+    """NATS server; ``host_port`` pins the published port (to start it after the gateway)."""
+    container = DockerContainer(NATS_IMAGE).waiting_for(LogMessageWaitStrategy('Server is ready'))
+    if host_port is None:
+        return container.with_exposed_ports(4222)
+    return container.with_bind_ports(4222, host_port)
+
+
+@pytest.fixture(scope='session')
+def nats_container():
+    nats_server = nats_server_container()
+    nats_server.start()
+    yield nats_server
+    nats_server.stop()
+
+
+@pytest.fixture(scope='session')
+def mongo_client(mongo_container):
+    port = mongo_container.get_exposed_port(27017)
+    client = MongoClient(f'mongodb://{MONGO_USER}:{MONGO_PASSWORD}@localhost:{port}')
+    yield client
+    client.close()
+
+
+@pytest.fixture(scope='session')
+def postgres_engine(postgres_container):
+    engine = create_engine(postgres_container.get_connection_url())
+    yield engine
+    engine.dispose()
+
+
+@pytest.fixture(scope='session')
+def redis_client(redis_container):
+    client = Redis(
+        host=redis_container.get_container_host_ip(),
+        port=int(redis_container.get_exposed_port(6379)),
+        decode_responses=True,
+    )
+    yield client
+    client.close()
+
+
+@pytest.fixture(scope='session')
+def minio_client(minio_container):
+    return minio_container.get_client()
+
+
+@pytest.fixture(scope='session')
+def nats_url(nats_container) -> str:
+    host = nats_container.get_container_host_ip()
+    return f'nats://{host}:{nats_container.get_exposed_port(4222)}'
+
+
+# --------------------------------------------------------------------------------------
+# Per-test environment and state reset
+# --------------------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def gateway_env(
+    monkeypatch,
+    tmp_path,
+    mongo_container,
+    redis_container,
+    postgres_container,
+    minio_container,
+    nats_url,
+):
+    """Point every gateway setting at the containers, with the app's defaults otherwise."""
+    minio_cfg = minio_container.get_config()
+    env = {
+        'HOSTNAME': 'pod-a',
+        'REDIS_HOST': redis_container.get_container_host_ip(),
+        'REDIS_PORT': str(redis_container.get_exposed_port(6379)),
+        'LEASE_TTL': '10',
+        'HEARTBEAT_TTL': '20',
+        'POLL_INTERVAL': '1',
+        'MONGODB_URL': f'localhost:{mongo_container.get_exposed_port(27017)}',
+        'MONGODB_USERNAME': MONGO_USER,
+        'MONGODB_PASSWORD': MONGO_PASSWORD,
+        'MONGODB_DATABASE': MONGO_DATABASE,
+        'POSTGRES_HOST': postgres_container.get_container_host_ip(),
+        'POSTGRES_PORT': str(postgres_container.get_exposed_port(5432)),
+        'POSTGRES_USER': 'sientia',
+        'POSTGRES_PASSWORD': 'sientia',
+        'POSTGRES_DBNAME': 'sientia',
+        'POSTGRES_MIN_CONNECTIONS': '1',
+        'POSTGRES_MAX_CONNECTIONS': '2',
+        'MINIO_ENDPOINT': minio_cfg['endpoint'],
+        'MINIO_ACCESS_KEY': minio_cfg['access_key'],
+        'MINIO_SECRET_KEY': minio_cfg['secret_key'],
+        'MINIO_SECURE': 'false',
+        'NATS_URL': nats_url,
+        'NATS_WRITE_SUBJECT': WRITE_SUBJECT,
+        'NATS_QUEUE': 'gateway-writers',
+        'GRPC_ENABLED': 'false',
+        'GRPC_PORT': str(free_port()),
+        'CERT_CACHE_DIR': str(tmp_path / 'cert-cache'),
+        'WRITE_MIN_PAUSE_S': '0',
+        'IDLE_CONN_MAX_S': '300',
+    }
+    for name, value in env.items():
+        monkeypatch.setenv(name, value)
+    for name in ('REDIS_USERNAME', 'REDIS_PASSWORD', 'WRITE_TIMESTAMP', 'WRITE_QUALITY'):
+        monkeypatch.delenv(name, raising=False)
+    return env
+
+
+@pytest.fixture(autouse=True)
+def reset_state(redis_client, mongo_client, postgres_engine, minio_client):
+    """Known-empty baseline: Redis flushed, Mongo database dropped, Postgres schema
+    recreated from ``e2e/db_schema.sql``, certificate bucket emptied."""
+    redis_client.flushdb()
+    mongo_client.drop_database(MONGO_DATABASE)
+    with postgres_engine.begin() as conn:
+        conn.execute(text('DROP SCHEMA IF EXISTS sientia_data CASCADE'))
+        conn.exec_driver_sql(DB_SCHEMA_SQL_PATH.read_text(encoding='utf-8'))
+    if minio_client.bucket_exists(CERT_BUCKET):
+        for obj in minio_client.list_objects(CERT_BUCKET, recursive=True):
+            minio_client.remove_object(CERT_BUCKET, obj.object_name)
+    else:
+        minio_client.make_bucket(CERT_BUCKET)
+    yield
+
+
+@pytest.fixture
+def mongo_db(mongo_client):
+    return mongo_client[MONGO_DATABASE]
+
+
+# --------------------------------------------------------------------------------------
+# Factories: OPC servers, pods, slots, certificates, NATS commands
+# --------------------------------------------------------------------------------------
+
+
+@pytest_asyncio.fixture
+async def opc_servers(tmp_path):
+    """
+    Factory of started in-process OPC servers.
+
+    ``await opc_servers('plant-a', {'Temperature': 20.0}, secured=False)`` declares the
+    variables (``{name: value}`` or ``{name: (value, data_type, writable)}``) and starts it.
+    """
+    started: dict[str, OpcTestServer] = {}
+
+    async def start(name: str, variables: dict, secured: bool = False) -> OpcTestServer:
+        server = OpcTestServer(name, cert_dir=tmp_path / 'server-certs', secured=secured)
+        for var, spec in variables.items():
+            value, data_type, writable = spec if isinstance(spec, tuple) else (spec, 'float', True)
+            server.add_variable(var, value, data_type, writable)
+        await server.start()
+        started[name] = server
+        return server
+
+    yield start
+    for server in started.values():
+        await server.stop()
+
+
+@pytest_asyncio.fixture
+async def pods(monkeypatch):
+    """
+    Factory of started gateway pods: ``await pods('pod-a', runtime=True, **env)``.
+
+    Extra keyword arguments are environment overrides applied before the pod is built
+    (e.g. ``WRITE_TIMESTAMP='false'``). Every pod is stopped at teardown.
+    """
+    started: list[Pod] = []
+
+    async def start(name: str = 'pod-a', runtime: bool = True, **env: str) -> Pod:
+        for key, value in env.items():
+            monkeypatch.setenv(key, value)
+        pod = Pod(name, runtime=runtime)
+        started.append(pod)
+        return await pod.start()
+
+    yield start
+    for pod in reversed(started):
+        await pod.stop()
+
+
+@pytest.fixture
+def put_slot(redis_client):
+    """Write a slot v2 to Redis (``slot:opc_tags:<n>``), as the Orchestrator does."""
+
+    def put(number: int, slot: dict) -> None:
+        redis_client.set(f'slot:opc_tags:{number}', json.dumps(slot))
+
+    return put
+
+
+@pytest_asyncio.fixture
+async def publish_command(nats_url):
+    """Publish C2 write commands (dicts) or raw payloads (bytes) on the write subject."""
+    client = await nats.connect(nats_url)
+
+    async def publish(payload) -> None:
+        data = payload if isinstance(payload, bytes) else json.dumps(payload).encode()
+        await client.publish(WRITE_SUBJECT, data)
+        await client.flush()
+
+    yield publish
+    await client.drain()
+
+
+@pytest_asyncio.fixture
+async def client_certificate(tmp_path, minio_client: Minio, mongo_db):
+    """
+    Create an OPC client certificate the way the platform stores it: cert (DER) and
+    private key (PEM) uploaded to MinIO, and an ``opc_certificates`` document (active,
+    kind client). Returns the certificate in the slot format (contract C1).
+    """
+
+    async def create(
+        certificate_id: int,
+        upload: bool = True,
+        status: str = 'active',
+        app_uri: str | None = None,
+    ) -> dict:
+        app_uri = app_uri or f'urn:sientia:e2e:client:{certificate_id}'
+        cert, key = await make_certificate(
+            tmp_path / 'client-certs', f'client-{certificate_id}', app_uri, server=False
+        )
+        objects = {
+            'cert': {'bucket': CERT_BUCKET, 'object_name': f'{certificate_id}/client.der'},
+            'private_key': {'bucket': CERT_BUCKET, 'object_name': f'{certificate_id}/client.pem'},
+        }
+        if upload:
+            for path, field in ((cert, 'cert'), (key, 'private_key')):
+                data = path.read_bytes()
+                minio_client.put_object(
+                    CERT_BUCKET, objects[field]['object_name'], io.BytesIO(data), len(data)
+                )
+        mongo_db['opc_certificates'].insert_one(
+            {
+                'certificate_id': certificate_id,
+                'kind': 'client',
+                'status': status,
+                'application_uri': app_uri,
+                **objects,
+            }
+        )
+        return {'certificate_id': certificate_id, 'application_uri': app_uri, **objects}
+
+    return create
+
+
+@pytest.fixture
+def cert_cache_dir() -> Path:
+    return Path(os.environ['CERT_CACHE_DIR'])
+
+
+@pytest.fixture
+def opc_wire(monkeypatch):
+    """Serialized WriteRequests, exactly as the gateway's OPC client sends them."""
+    sent: list[bytes] = []
+    original = ua_client.struct_to_binary
+
+    def capture(obj):
+        data = original(obj)
+        if isinstance(obj, ua.WriteRequest):
+            sent.append(data)
+        return data
+
+    monkeypatch.setattr(ua_client, 'struct_to_binary', capture)
+    return sent

e2e/db_schema.sql

@@ -0,0 +1,33 @@
+-- =============================================================================
+-- E2E test database schema for the ``sientia_data`` namespace.
+--
+-- Mirrors the production DDL one-to-one so any change in production can be
+-- pasted directly into this file. The conftest fixture drops the schema and
+-- loads this SQL into the testcontainers Postgres before every test.
+--
+-- The gateway only touches ``sientia_data.predictions``: it updates
+-- ``write_status`` (contract C3, SIENTIAPDE-1975) on the row the Laborious
+-- inserted. Same DDL as ``sientia-dataops-laborious/e2e/db_schema.sql``.
+-- =============================================================================
+
+CREATE SCHEMA IF NOT EXISTS sientia_data;
+
+-- -----------------------------------------------------------------------------
+-- sientia_data.predictions
+-- -----------------------------------------------------------------------------
+CREATE TABLE IF NOT EXISTS sientia_data.predictions (
+    model_id              int4 NOT NULL,
+    prediction            numeric NULL,
+    prediction_confidence numeric NOT NULL,
+    response_time         numeric NOT NULL,
+    prediction_status     text NOT NULL,
+    "timestamp"           timestamptz NOT NULL,
+    created_at            timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL,
+    "comments"            text NULL,
+    metadata_id           varchar(24) NULL,
+    write_status          text NULL,
+    CONSTRAINT unique_model_id_timestamp
+        UNIQUE (model_id, "timestamp"),
+    CONSTRAINT chk_predictions_write_status
+        CHECK (write_status IN ('pending', 'ok', 'error', 'session_bad'))
+);

e2e/helpers.py

@@ -0,0 +1,243 @@
+"""Shared helpers for the gateway E2E tests (OPC UA + Redis + MongoDB + Postgres + MinIO + NATS)."""
+
+import asyncio
+import copy
+import inspect
+import json
+import os
+import time
+from collections.abc import Awaitable, Callable
+from pathlib import Path
+from typing import Any
+
+from asyncua import ua
+from asyncua.ua.ua_binary import Primitives, nodeid_to_binary
+from pymongo.database import Database
+from sqlalchemy import text
+from sqlalchemy.engine import Engine
+
+from ingestor.config import GatewayConfig
+from ingestor.gateway_runtime import GatewayRuntime
+from ingestor.ingestor import Ingestor
+
+SCENARIO_INPUTS_DIR = Path(__file__).parent / 'scenario_inputs'
+
+# Collections the gateway reads (connectors, opc_certificates) or writes
+# (readings, notification_queue). Dropped before every test with the whole database.
+NOTIFICATIONS = 'notification_queue'
+CERT_BUCKET = 'opc-certificates'
+PREDICTIONS_TABLE = 'predictions'
+
+
+def _resolve_markers(value: Any, servers: dict) -> Any:
+    """Replace ``@server:<name>`` with the URL of that in-process OPC server."""
+    if isinstance(value, dict):
+        return {k: _resolve_markers(v, servers) for k, v in value.items()}
+    if isinstance(value, list):
+        return [_resolve_markers(v, servers) for v in value]
+    if isinstance(value, str) and value.startswith('@server:'):
+        return servers[value.removeprefix('@server:')].url
+    return value
+
+
+def load_scenario_input(name: str, servers: dict | None = None, **overrides: Any) -> Any:
+    """
+    Load a JSON scenario input from ``e2e/scenario_inputs``.
+
+    Args:
+        name: File name (e.g. ``slot_single_server.json``).
+        servers: ``{name: OpcTestServer}`` used to resolve ``@server:<name>`` markers.
+        **overrides: Top-level keys replaced in the loaded dict.
+
+    Return:
+        Any: The resolved JSON content.
+    """
+    data = json.loads((SCENARIO_INPUTS_DIR / name).read_text(encoding='utf-8'))
+    data = _resolve_markers(data, servers or {})
+    if overrides:
+        data = {**data, **copy.deepcopy(overrides)}
+    return data
+
+
+async def wait_until(
+    predicate: Callable[[], Any | Awaitable[Any]], timeout: float = 15.0, interval: float = 0.1
+) -> Any:
+    """
+    Poll ``predicate`` (sync or async) until it returns a truthy value.
+
+    Sleeps on the event loop between attempts, so OPC subscription callbacks and NATS
+    handlers keep running while the test waits.
+
+    Raises:
+        AssertionError: When the predicate is still falsy after ``timeout`` seconds.
+    """
+    deadline = time.monotonic() + timeout
+    while True:
+        result = predicate()
+        if inspect.isawaitable(result):
+            result = await result
+        if result:
+            return result
+        if time.monotonic() > deadline:
+            raise AssertionError(f'condition not met within {timeout}s')
+        await asyncio.sleep(interval)
+
+
+async def stays_false(
+    predicate: Callable[[], Any | Awaitable[Any]], duration: float = 1.5, interval: float = 0.1
+) -> bool:
+    """True when ``predicate`` never becomes truthy during ``duration`` seconds."""
+    deadline = time.monotonic() + duration
+    while time.monotonic() < deadline:
+        result = predicate()
+        if inspect.isawaitable(result):
+            result = await result
+        if result:
+            return False
+        await asyncio.sleep(interval)
+    return True
+
+
+class Pod:
+    """
+    One gateway pod, wired exactly as ``ingestor/app.py`` does (minus the Prometheus
+    HTTP server and the sleep between iterations): ``Ingestor.prepare_ingestor``, then
+    ``GatewayRuntime`` on its data manager and connection pool. ``tick`` runs one main
+    loop iteration (``Ingestor.loop`` + ``GatewayRuntime.idle_sweep``).
+
+    Every component reads its settings from the environment at construction time, so
+    the caller sets the env (``HOSTNAME`` included) before ``start``.
+    """
+
+    def __init__(self, name: str, runtime: bool = True):
+        self.name = name
+        self.with_runtime = runtime
+        self.ingestor: Ingestor | None = None
+        self.runtime: GatewayRuntime | None = None
+
+    async def start(self) -> 'Pod':
+        os.environ['HOSTNAME'] = self.name
+        self.ingestor = Ingestor()
+        await self.ingestor.prepare_ingestor()
+        if self.with_runtime:
+            manager = self.ingestor.ingestor_manager
+            assert manager is not None
+            self.runtime = GatewayRuntime(
+                config=GatewayConfig.from_env(),
+                data_manager=manager.data_manager,
+                pool=manager.pool,
+                logger=self.ingestor.logger,
+                notification_handler=self.ingestor.notification_handler,
+                metrics_controller=self.ingestor.metrics_controller,
+                metadata=self.ingestor.metadata,
+            )
+            await self.runtime.start()
+            consumer = self.runtime.nats_consumer
+            if consumer.nc is not None:
+                # PING/PONG: the server has processed the SUB, so a command published
+                # right after start is not lost for lack of a subscriber.
+                await consumer.nc.flush()
+        return self
+
+    @property
+    def manager(self):
+        assert self.ingestor is not None and self.ingestor.ingestor_manager is not None
+        return self.ingestor.ingestor_manager
+
+    @property
+    def slots(self) -> set[str]:
+        """Slots this pod currently manages."""
+        return set(self.manager.managed_tags)
+
+    async def tick(self, times: int = 1) -> None:
+        assert self.ingestor is not None
+        for _ in range(times):
+            await self.ingestor.loop()
+            if self.runtime is not None:
+                await self.runtime.idle_sweep()
+
+    async def stop(self) -> None:
+        if self.runtime is not None:
+            await self.runtime.stop()
+            self.runtime = None
+        if self.ingestor is not None:
+            await self.ingestor.shutdown()
+            self.ingestor.notification_handler.shutdown()
+            self.ingestor = None
+
+
+def redis_json(redis_client, key: str) -> Any:
+    """Value stored by the gateway's RedisRepository (JSON-encoded), or None."""
+    raw = redis_client.get(key)
+    return None if raw is None else json.loads(raw)
+
+
+def readings(db: Database, collection: str, **query: Any) -> list[dict]:
+    """Documents the read path persisted into ``collection`` (oldest first)."""
+    return list(db[collection].find(query, {'_id': 0}).sort('inserted_at', 1))
+
+
+def notifications(db: Database, prefix: str = '') -> list[dict]:
+    """Notifications sent by the gateway whose ``notification_id`` starts with ``prefix``."""
+    return [
+        n
+        for n in db[NOTIFICATIONS].find({}, {'_id': 0})
+        if str(n.get('notification_id', '')).startswith(prefix)
+    ]
+
+
+def insert_prediction(engine: Engine, model_id: int, timestamp: str, write_status: str | None):
+    """One ``sientia_data.predictions`` row, as the Laborious inserts it before publishing C2."""
+    with engine.begin() as conn:
+        conn.execute(
+            text(
+                'INSERT INTO sientia_data.predictions (model_id, prediction, '
+                'prediction_confidence, response_time, prediction_status, "timestamp", '
+                'write_status) VALUES (:model_id, 1.5, 0.9, 0.1, :status, :ts, :write_status)'
+            ),
+            {
+                'model_id': model_id,
+                'status': 'success',
+                'ts': timestamp,
+                'write_status': write_status,
+            },
+        )
+
+
+def write_status(engine: Engine, model_id: int, timestamp: str) -> str | None:
+    """``write_status`` of a prediction row (``'<missing>'`` when the row does not exist)."""
+    with engine.begin() as conn:
+        row = conn.execute(
+            text(
+                'SELECT write_status FROM sientia_data.predictions '
+                'WHERE model_id = :model_id AND "timestamp" = :ts'
+            ),
+            {'model_id': model_id, 'ts': timestamp},
+        ).first()
+    return '<missing>' if row is None else row[0]
+
+
+def prediction_count(engine: Engine) -> int:
+    with engine.begin() as conn:
+        return conn.execute(text('SELECT count(*) FROM sientia_data.predictions')).scalar_one()
+
+
+# DataValue encoding mask bits (OPC UA Part 6, 5.2.2.17).
+STATUS_CODE_BIT = 0x02
+SOURCE_TIMESTAMP_BIT = 0x04
+
+
+def datavalue_mask(request: bytes, addr: str) -> int:
+    """
+    Encoding mask of the DataValue written to ``addr`` in a serialized WriteRequest.
+
+    An omitted StatusCode decodes as Good on the server (OPC UA default), so whether
+    the gateway sent the quality is only visible on the wire. WriteValue = NodeId,
+    AttributeId, IndexRange, DataValue: the byte right after the first three is the mask.
+    """
+    prefix = (
+        nodeid_to_binary(ua.NodeId.from_string(addr))
+        + Primitives.UInt32.pack(ua.AttributeIds.Value)
+        + Primitives.String.pack(None)
+    )
+    return request[request.index(prefix) + len(prefix)]

e2e/opc_server.py

@@ -0,0 +1,134 @@
+"""In-process OPC UA servers for the E2E suite (asyncua `Server`).
+
+Each `OpcTestServer` listens on its own localhost port and exposes string NodeIds
+under namespace 2 (`ns=2;s=<name>`), so a slot can address them the same way the
+Orchestrator addresses a real server. Writes received from the gateway are recorded
+(decoded `WriteValue`s) through the server's PreWrite callback.
+"""
+
+import socket
+import time
+from pathlib import Path
+
+from asyncua import Server, ua
+from asyncua.common.callback import CallbackType
+from asyncua.crypto.cert_gen import setup_self_signed_certificate
+from cryptography.x509.oid import ExtendedKeyUsageOID
+
+NAMESPACE_URI = 'urn:sientia:e2e'
+
+_VARIANT_TYPES = {
+    'float': ua.VariantType.Float,
+    'double': ua.VariantType.Double,
+    'int': ua.VariantType.Int32,
+    'bool': ua.VariantType.Boolean,
+    'string': ua.VariantType.String,
+}
+
+
+def free_port() -> int:
+    with socket.socket() as s:
+        s.bind(('127.0.0.1', 0))
+        return s.getsockname()[1]
+
+
+def address(name: str) -> str:
+    """NodeId string of a variable created by `OpcTestServer.add_variable`."""
+    return f'ns=2;s={name}'
+
+
+async def make_certificate(
+    folder: Path, name: str, app_uri: str, server: bool
+) -> tuple[Path, Path]:
+    """Self-signed application certificate (DER) + private key (PEM) with `app_uri` as
+    the SAN URI, as asyncua expects from both ends of a secure channel."""
+    folder.mkdir(parents=True, exist_ok=True)
+    cert, key = folder / f'{name}.der', folder / f'{name}.pem'
+    usage = ExtendedKeyUsageOID.SERVER_AUTH if server else ExtendedKeyUsageOID.CLIENT_AUTH
+    await setup_self_signed_certificate(
+        key, cert, app_uri, 'localhost', [usage], {'commonName': name}
+    )
+    return cert, key
+
+
+class OpcTestServer:
+    """One asyncua server. `secured=True` exposes only Basic256Sha256 SignAndEncrypt."""
+
+    def __init__(self, name: str, cert_dir: Path | None = None, secured: bool = False):
+        self.name = name
+        self.port = free_port()
+        self.secured = secured
+        self.cert_dir = cert_dir
+        self.server: Server | None = None
+        self.variables: dict[str, tuple[object, str, bool]] = {}
+        self.writes: list[ua.WriteValue] = []
+        self.write_times: list[float] = []  # time.monotonic() of each received write
+
+    @property
+    def url(self) -> str:
+        return f'opc.tcp://127.0.0.1:{self.port}'
+
+    def add_variable(
+        self, name: str, value: object, data_type: str = 'float', writable: bool = True
+    ) -> str:
+        """Declare a variable (created on `start`, and again on every restart)."""
+        self.variables[name] = (value, data_type, writable)
+        return address(name)
+
+    async def start(self) -> None:
+        server = Server()
+        await server.init()
+        server.set_endpoint(self.url)
+        server.set_server_name(self.name)
+        if self.secured:
+            assert self.cert_dir is not None, 'a secured server needs a cert_dir'
+            cert, key = await make_certificate(
+                self.cert_dir, f'server-{self.name}', f'urn:sientia:e2e:{self.name}', True
+            )
+            await server.load_certificate(str(cert))
+            await server.load_private_key(str(key))
+            server.set_security_policy([ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt])
+        idx = await server.register_namespace(NAMESPACE_URI)
+        assert idx == 2, f'unexpected namespace index {idx}'
+        for name, (value, data_type, writable) in self.variables.items():
+            node = await server.nodes.objects.add_variable(
+                address(name), name, value, _VARIANT_TYPES[data_type]
+            )
+            if writable:
+                await node.set_writable()
+        server.subscribe_server_callback(CallbackType.PreWrite, self._record_write)
+        await server.start()
+        self.server = server
+
+    async def stop(self) -> None:
+        if self.server is not None:
+            await self.server.stop()
+            self.server = None
+
+    async def restart(self) -> None:
+        """Stop and start again on the same port (values reset to their initial ones)."""
+        await self.stop()
+        await self.start()
+
+    async def set(self, name: str, value: object) -> None:
+        """Change a value server-side (raises a data change for subscribed clients)."""
+        assert self.server is not None
+        _, data_type, _ = self.variables[name]
+        node = self.server.get_node(address(name))
+        await node.write_value(ua.Variant(value, _VARIANT_TYPES[data_type]))
+
+    async def read(self, name: str) -> ua.DataValue:
+        assert self.server is not None
+        return await self.server.get_node(address(name)).read_data_value()
+
+    def session_count(self) -> int:
+        """Client sessions currently open on this server."""
+        assert self.server is not None
+        return len(self.server.iserver._external_sessions)
+
+    async def _record_write(self, event, _dispatcher) -> None:
+        if not event.is_external:  # the server updates its own nodes (ServerStatus, ...)
+            return
+        for value in event.request_params.NodesToWrite:
+            self.writes.append(value)
+            self.write_times.append(time.monotonic())

e2e/scenario_inputs/connector_plant_a.json

@@ -0,0 +1,10 @@
+{
+  "connector_id": "1",
+  "type": "opc_ua",
+  "name": "plant-a",
+  "url": "@server:plant-a",
+  "status": "Active",
+  "subscription_period_ms": 100,
+  "security": {"mode": "None", "policy": "None", "client_certificate_id": null},
+  "legacy_opc_server_id": "1"
+}

e2e/scenario_inputs/slot_active_read.json

@@ -0,0 +1,25 @@
+{
+  "plant-a": {
+    "name": "plant-a",
+    "server_id": "1",
+    "url": "@server:plant-a",
+    "subscription_period_ms": 100,
+    "security": {"mode": "None", "policy": "None"},
+    "certificate": null,
+    "tags": {
+      "ns=2;s=Level": {
+        "frequency": 1000,
+        "active_read": {"enabled": true, "cycle_multiple": 1},
+        "names": [
+          {"collection": "active_tank", "tag_name": "tank_level", "model_id": "30"},
+          {"collection": "active_mixer", "tag_name": "mixer_inlet_level", "model_id": "40"}
+        ]
+      },
+      "ns=2;s=Temperature": {
+        "frequency": 1000,
+        "active_read": {"enabled": false},
+        "names": [{"collection": "active_tank", "tag_name": "tank_temperature", "model_id": "30"}]
+      }
+    }
+  }
+}

e2e/scenario_inputs/slot_data_types.json

@@ -0,0 +1,28 @@
+{
+  "plant-a": {
+    "name": "plant-a",
+    "server_id": "1",
+    "url": "@server:plant-a",
+    "subscription_period_ms": 100,
+    "security": {"mode": "None", "policy": "None"},
+    "certificate": null,
+    "tags": {
+      "ns=2;s=Temperature": {
+        "frequency": 1000,
+        "names": [{"collection": "raw_types", "tag_name": "temperature", "model_id": "10"}]
+      },
+      "ns=2;s=Counter": {
+        "frequency": 1000,
+        "names": [{"collection": "raw_types", "tag_name": "counter", "model_id": "10"}]
+      },
+      "ns=2;s=Running": {
+        "frequency": 1000,
+        "names": [{"collection": "raw_types", "tag_name": "running", "model_id": "10"}]
+      },
+      "ns=2;s=Batch": {
+        "frequency": 1000,
+        "names": [{"collection": "raw_types", "tag_name": "batch", "model_id": "10"}]
+      }
+    }
+  }
+}

e2e/scenario_inputs/slot_multi_name.json

@@ -0,0 +1,20 @@
+{
+  "plant-a": {
+    "name": "plant-a",
+    "server_id": "1",
+    "url": "@server:plant-a",
+    "subscription_period_ms": 100,
+    "security": {"mode": "None", "policy": "None"},
+    "certificate": null,
+    "tags": {
+      "ns=2;s=Pressure": {
+        "frequency": 1000,
+        "active_read": {"enabled": false},
+        "names": [
+          {"collection": "raw_furnace", "tag_name": "furnace_pressure", "model_id": "10", "model_name": "furnace"},
+          {"collection": "raw_boiler", "tag_name": "boiler_inlet_pressure", "model_id": "20", "model_name": "boiler"}
+        ]
+      }
+    }
+  }
+}

e2e/scenario_inputs/slot_secured.json

@@ -0,0 +1,19 @@
+{
+  "plant-s": {
+    "name": "plant-s",
+    "server_id": "7",
+    "url": "@server:plant-s",
+    "subscription_period_ms": 100,
+    "security": {"mode": "SignAndEncrypt", "policy": "Basic256Sha256"},
+    "certificate": null,
+    "tags": {
+      "ns=2;s=Temperature": {
+        "frequency": 1000,
+        "active_read": {"enabled": false},
+        "names": [
+          {"collection": "raw_secured", "tag_name": "secured_temperature", "model_id": "70"}
+        ]
+      }
+    }
+  }
+}

e2e/scenario_inputs/slot_single_server.json

@@ -0,0 +1,19 @@
+{
+  "plant-a": {
+    "name": "plant-a",
+    "server_id": "1",
+    "url": "@server:plant-a",
+    "subscription_period_ms": 100,
+    "security": {"mode": "None", "policy": "None"},
+    "certificate": null,
+    "tags": {
+      "ns=2;s=Temperature": {
+        "frequency": 1000,
+        "active_read": {"enabled": false},
+        "names": [
+          {"collection": "raw_plant_a", "tag_name": "temperature", "model_id": "10", "model_name": "furnace"}
+        ]
+      }
+    }
+  }
+}

e2e/scenario_inputs/slot_two_servers.json

@@ -0,0 +1,36 @@
+{
+  "plant-a": {
+    "name": "plant-a",
+    "server_id": "1",
+    "url": "@server:plant-a",
+    "subscription_period_ms": 100,
+    "security": {"mode": "None", "policy": "None"},
+    "certificate": null,
+    "tags": {
+      "ns=2;s=Temperature": {
+        "frequency": 1000,
+        "active_read": {"enabled": false},
+        "names": [
+          {"collection": "raw_plant_a", "tag_name": "temperature", "model_id": "10", "model_name": "furnace"}
+        ]
+      }
+    }
+  },
+  "plant-b": {
+    "name": "plant-b",
+    "server_id": "2",
+    "url": "@server:plant-b",
+    "subscription_period_ms": 100,
+    "security": {"mode": "None", "policy": "None"},
+    "certificate": null,
+    "tags": {
+      "ns=2;s=Flow": {
+        "frequency": 1000,
+        "active_read": {"enabled": false},
+        "names": [
+          {"collection": "raw_plant_b", "tag_name": "flow", "model_id": "20", "model_name": "boiler"}
+        ]
+      }
+    }
+  }
+}

e2e/scenario_inputs/slot_v1_invalid.json

@@ -0,0 +1,10 @@
+{
+  "plant-a": {
+    "name": "plant-a",
+    "url": "@server:plant-a",
+    "server_uri": "http://opcua-server.simulator",
+    "tags": {
+      "ns=2;s=Temperature": {"tag_name": "Temperature", "frequency": 1000, "topics": ["raw_plant_a"]}
+    }
+  }
+}

e2e/scenario_inputs/write_command.json

@@ -0,0 +1,12 @@
+{
+  "server_id": "1",
+  "addr": "ns=2;s=Setpoint",
+  "value": 1.25,
+  "data_type": "float",
+  "tag_type": "prediction",
+  "model_id": "2",
+  "model_name": "sin-approx",
+  "schema": "sientia_data",
+  "table_name": "predictions",
+  "timestamp": "2026-10-13T12:00:00-03:00"
+}

e2e/scenarios.md

@@ -0,0 +1,584 @@
+# E2E Scenario Documentation — OPC Gateway
+
+Functional reference for the gateway end-to-end scenarios: slot leasing, subscription reads,
+NATS writes, gRPC reads, active reads, certificates and the pod lifecycle. Tests live under
+`e2e/`, are marked with `@pytest.mark.integration` and run with:
+
+```bash
+pytest e2e/ -o testpaths=e2e -m integration
+```
+
+**Every section below maps 1:1 to an actual test function** (or to one `pytest.param` id of a
+parametrized test) — this file describes what the suite verifies today. Adding a scenario here
+without a matching test (or vice versa) is a doc bug; fix the mismatch instead of leaving it
+for later.
+
+## Execution Context
+
+### One-time setup: disable the testcontainers reaper
+
+Same as the Laborious and Orchestrator suites. Once per machine:
+
+```bash
+echo "ryuk.disabled=true" >> ~/.testcontainers.properties
+```
+
+Under Docker Desktop on Linux the reaper ("Ryuk") cannot mount the Docker socket and every
+test errors at setup (`ERROR`, not `FAILED`). Every fixture stops its own containers in
+teardown; after a killed run, clean up by hand:
+
+```bash
+docker ps -a --filter "label=org.testcontainers=true" -q | xargs -r docker rm -f
+```
+
+### What runs for real
+
+| Dependency | How | Scope |
+|---|---|---|
+| MongoDB (`mongo:7`, user/password) | testcontainers | session |
+| Redis (`redis:7-alpine`) | testcontainers | session |
+| PostgreSQL (`postgres:15`, `e2e/db_schema.sql`) | testcontainers | session |
+| MinIO (`bitnamilegacy/minio`, client certificates) | testcontainers | session |
+| NATS (`nats:2.10-alpine`) | testcontainers | session |
+| OPC UA servers | in process (`e2e/opc_server.py`, asyncua `Server`) | per test |
+
+- Gateway code under `ingestor/**` is not mocked. A pod (`e2e/helpers.Pod`) is built from the
+  environment exactly like `ingestor/app.py` builds it — `Ingestor.prepare_ingestor`, then
+  `GatewayRuntime` on the same data manager and connection pool — minus the Prometheus HTTP
+  server. `Pod.tick()` runs one main loop iteration (`Ingestor.loop` + `GatewayRuntime.idle_sweep`)
+  without the `POLL_INTERVAL` sleep, so every scenario controls exactly how many iterations run.
+- The environment (`gateway_env` in `e2e/conftest.py`) points every setting at the containers;
+  scenarios override single variables per pod (`LEASE_TTL`, `WRITE_TIMESTAMP`, ...).
+- State is reset before every test: Redis `FLUSHDB`, Mongo database dropped, Postgres schema
+  recreated from `e2e/db_schema.sql`, certificate bucket emptied.
+- Inputs live in `e2e/scenario_inputs/*.json` and are loaded with `load_scenario_input`;
+  `@server:<name>` markers resolve to the URL of that test's in-process OPC server.
+- Default `POLL_INTERVAL=1` and tag `frequency=1000`: one OPC "cycle" per loop iteration.
+- Not covered by the default `pytest` run (`testpaths = ["tests"]`) nor by the unit coverage.
+  `--strict-markers` is on: a test without `@pytest.mark.integration` is not collected, and a run
+  that collects nothing still exits 0 — check the collected count.
+
+---
+
+## 1. Slot Leasing and Load Balancing
+Source: `e2e/test_slot_leasing.py`
+
+### 1.1.1 Single Pod Leases the Slot and Heartbeats
+**Summary**: A lone pod takes the only slot on its first loop iteration.
+
+**Description**:
+- Slot 1 reads `plant-a`; one pod starts and runs one iteration.
+
+**Expected Outcome**:
+- The pod manages slot `1`; `lease:opc_tags:1 = "pod-a"` with TTL ≤ `LEASE_TTL` (10 s).
+- `heartbeat:ingestor:pod-a = 1` with TTL ≤ `HEARTBEAT_TTL` (20 s).
+- Readings of the slot reach MongoDB.
+
+### 1.1.2 Idle Pod Takes a Slot Created After Startup
+**Summary**: A pod without slots picks up a slot the Orchestrator creates later.
+
+**Description**:
+- The pod starts with Redis empty and runs one iteration (nothing to lease).
+- Slot 1 is written; the pod runs another iteration.
+
+**Expected Outcome**:
+- The pod manages slot `1`, holds its lease and its readings reach MongoDB.
+
+### 1.2.1 Two Pods Split Two Slots
+**Summary**: The rebalancing hands one slot to each pod.
+
+**Description**:
+- Slot 1 reads `plant-a`, slot 2 reads `plant-b`. `pod-a` starts alone, then `pod-b` starts.
+- Iterations: `pod-a`, `pod-b`, `pod-a`, `pod-b`.
+
+**Expected Outcome**:
+- Alone, `pod-a` takes both slots; `pod-b` finds them leased.
+- `pod-a` sees a pod without slots and gives slot `2` back (unsubscribing first); `pod-b` takes it.
+- Leases: slot 1 → `pod-a`, slot 2 → `pod-b`; each pod only has the reader of its own server.
+- New values of both servers reach MongoDB.
+
+### 1.2.2 Extra Pod Stays Idle
+**Summary**: With more pods than slots, no slot is read twice.
+
+**Description**:
+- Two slots balanced between `pod-a` and `pod-b`; `pod-c` joins and every pod iterates twice.
+
+**Expected Outcome**:
+- `pod-c` manages nothing; `pod-a` and `pod-b` hold disjoint slots covering `{1, 2}`.
+- Leases belong to `pod-a` and `pod-b` only.
+
+### 1.2.3 Failover After Lease Expiry
+**Summary**: A slot left by a dead pod is taken over once its lease expires.
+
+**Description**:
+- `pod-a` (`LEASE_TTL=2`) reads slot 1; `pod-b` runs idle.
+- `pod-a` stops without releasing the lease (crash-like).
+
+**Expected Outcome**:
+- The lease still names `pod-a` right after the stop, then expires.
+- On its next iteration `pod-b` leases slot `1` and new values reach MongoDB.
+
+### 1.3.1 Deleted Slot Stops the Reads
+**Summary**: Removing a slot from Redis unsubscribes it.
+
+**Description**:
+- A pod reads slot 1; the slot key is deleted; the pod iterates.
+
+**Expected Outcome**:
+- The pod manages no slot and has no OPC reader.
+- A value changed afterwards is not persisted.
+
+### 1.3.2 Invalid Slot Is Notified and Not Read
+**Summary**: A slot v1 (`topics`, no `security`/`names`) is rejected.
+
+**Description**:
+- Slot 1 holds the legacy v1 shape (`slot_v1_invalid.json`).
+
+**Expected Outcome**:
+- Notification `OPC_SLOT_CONFIG_ERROR_1` in `notification_queue`.
+- The pod manages nothing; no reading is persisted.
+
+### 1.3.3 Fixed Slot Is Read Once Its Lease Expires
+**Summary**: Fixing an invalid slot takes effect after the lease of the failed acquisition expires.
+
+**Description**:
+- `pod-a` (`LEASE_TTL=2`) leases the invalid slot 1 (the lease is kept, the slot is not managed).
+- The slot is fixed; the pod iterates before and after the lease expires.
+
+**Expected Outcome**:
+- Before expiry the pod still manages nothing (its own lease blocks the re-acquisition).
+- After expiry the pod manages slot `1` and its readings reach MongoDB.
+
+---
+
+## 2. Subscription Reads (OPC UA → MongoDB)
+Source: `e2e/test_subscription_reads.py`
+
+### 2.1.1 Value Change Is Persisted
+**Summary**: Each data change becomes one document in the name's collection.
+
+**Description**:
+- Slot 1 subscribes `ns=2;s=Temperature` of `plant-a` into `raw_plant_a`; the value changes.
+
+**Expected Outcome**:
+- The initial value and the new one are persisted.
+- Document fields: `tag` (address), `name` (`tag_name`), `model_id`, `value`, `timestamp`
+  (source timestamp, `YYYY-MM-DD HH:MM:SS-0300`) and `inserted_at` — nothing else.
+
+### 2.1.2 One Address Fans Out to Every Name
+**Summary**: An address with several `names` is written to each collection.
+
+**Description**:
+- `ns=2;s=Pressure` has two names: `raw_furnace`/`furnace_pressure`/model 10 and
+  `raw_boiler`/`boiler_inlet_pressure`/model 20 (`slot_multi_name.json`).
+
+**Expected Outcome**:
+- One document per collection for the change, each with its own `name`/`model_id`, the same
+  `tag` and the same `timestamp`.
+
+### 2.1.3 Two Servers in One Slot
+**Summary**: A slot can hold several OPC servers.
+
+**Description**:
+- Slot 1 holds `plant-a` and `plant-b` (`slot_two_servers.json`); both values change.
+
+**Expected Outcome**:
+- Both changes are persisted; the pod has one reader per server.
+
+### 2.1.4 Values Keep Their OPC Type
+**Summary**: Double, Int32, Boolean and String values are stored with their native type.
+
+**Description**:
+- Four tags of different OPC types change value (`slot_data_types.json`).
+
+**Expected Outcome**:
+- Each document stores the value with the matching Python/BSON type (`float`, `int`, `bool`, `str`).
+
+### 2.2.1 Tag Added to the Slot Is Subscribed
+**Summary**: A new tag in the slot is subscribed on the next iteration.
+
+**Description**:
+- The Orchestrator adds `ns=2;s=Level` to slot 1; the pod iterates.
+
+**Expected Outcome**:
+- Changes of the new tag and of the existing one are persisted.
+
+### 2.2.2 Tag Removed from the Slot Is No Longer Read
+**Summary**: A tag removed from the slot stops being persisted.
+
+**Description**:
+- Slot 1 reads `Temperature` and `Level`; `Level` is removed; the pod iterates.
+
+**Expected Outcome**:
+- `Temperature` keeps being persisted; a new `Level` value is not.
+
+### 2.2.3 Name Added to a Tag Receives Its Readings
+**Summary**: A new name on an existing tag starts receiving the readings.
+
+**Description**:
+- A second name (`raw_dryer`/`dryer_temperature`/model 50) is added to `Temperature`.
+
+**Expected Outcome**:
+- The next change is persisted in both collections.
+
+### 2.2.4 Server URL Change Reconnects to the New Server
+**Summary**: Changing the connection settings of a server re-initializes its reader.
+
+**Description**:
+- Slot 1 points `plant-a` to a second server (same tags) and the pod iterates.
+
+**Expected Outcome**:
+- Values of the new server are persisted; the session on the old server is closed.
+- Changes on the old server are no longer persisted.
+
+### 2.3.1 Unreachable Server Does Not Stop the Others
+**Summary**: A server that cannot be reached is dropped; the other servers of the slot keep reading.
+
+**Description**:
+- Slot 1 holds `plant-a` and `plant-b`, whose URL has nothing listening.
+
+**Expected Outcome**:
+- Notification `OPC_CONNECTION_ERROR_plant-b`; the pod only has the `plant-a` reader.
+- `plant-a` values are persisted.
+
+### 2.3.2 Tag Without Data Raises a Warning
+**Summary**: A tag without data changes for 5 cycles is notified.
+
+**Description**:
+- The value does not change for 5 iterations.
+
+**Expected Outcome**:
+- One WARNING notification `TAG_ns=2;s=Temperature:temperature_LISTENNING_STOPPED`.
+
+### 2.3.3 Silent Server Is Reconnected After 15 Cycles
+**Summary**: A server without any data change for 15 cycles is treated as silently dead.
+
+**Description**:
+- The value does not change for 15 iterations, then the pod iterates once more.
+
+**Expected Outcome**:
+- Notifications `OPC_LISTENNING_STOPPED__1` and `OPC_CONNECTION_RETRY__1`; the reader is dropped
+  (pool connection evicted).
+- The next iteration re-initializes the reader with a new connection; changes are persisted again.
+
+### 2.3.4 Server Restart Resumes the Reads
+**Summary**: A server that goes down and comes back is read again.
+
+**Description**:
+- The server stops; the pod iterates 15 times (reader dropped) and once more while it is down;
+  the server starts again on the same port; the pod iterates twice.
+
+**Expected Outcome**:
+- While down, the reconnection fails with notification `OPC_CONNECTION_ERROR_plant-a`.
+- Once back, the reader is re-initialized and changes are persisted again.
+
+---
+
+## 3. Writes (NATS C2 → OPC UA → Postgres `write_status` C3)
+Source: `e2e/test_writes.py`
+
+Common setup: `plant-a` with a writable `Setpoint`, its connector in `connectors`
+(`connector_plant_a.json`, `connector_id "1"`) and a `pending` row in `sientia_data.predictions`
+(model 2, `2026-10-13T12:00:00-03:00`). Commands come from `write_command.json` and are published
+on `opc.write`; the gateway answers through the row's `write_status`.
+
+### 3.1 DataValue Parts: Timestamp and Quality (SIENTIAPDE-2202)
+**Summary**: The value is always written; the prediction timestamp (`SourceTimestamp`) and a Good
+`StatusCode` are written unless turned off. Each flag comes from the command, else from the env
+(`WRITE_TIMESTAMP` / `WRITE_QUALITY`), else `true`.
+
+**Description**:
+- `test_scenario_3_1_datavalue_parts`, one case per id. The timestamp is checked on the value the
+  server stored; the quality on the DataValue encoding mask of the WriteRequest the gateway sent
+  (an omitted StatusCode decodes as Good on the server, so only the wire tells them apart).
+
+| Id | Env | Command flags | SourceTimestamp | StatusCode |
+|---|---|---|---|---|
+| `3_1_1_defaults_on` | — | — | sent | sent |
+| `3_1_2_timestamp_off` | `WRITE_TIMESTAMP=false` | — | not sent | sent |
+| `3_1_3_quality_off` | `WRITE_QUALITY=false` | — | sent | not sent |
+| `3_1_4_both_off` | both `false` | — | not sent | not sent |
+| `3_1_5_command_overrides_env` | `WRITE_TIMESTAMP=true` | `write_timestamp: false` | not sent | sent |
+| `3_1_6_command_turns_on_over_env` | both `false` | both `true` | sent | sent |
+
+**Expected Outcome**:
+- `write_status = ok`; the server holds `1.25`.
+- When sent, the stored `SourceTimestamp` equals the command `timestamp`; otherwise it is empty.
+
+### 3.2.1 Pending Row Becomes Ok
+**Summary**: A successful write records `ok`.
+
+**Expected Outcome**:
+- `write_status = ok`; the server holds the command value.
+
+### 3.2.2 Null write_status Becomes Ok
+**Summary**: Rows without a status (inserted before the column existed) are updated too.
+
+**Expected Outcome**:
+- `write_status` goes from NULL to `ok`.
+
+### 3.2.3 Worse Outcome Is Kept
+**Summary**: `ok` never replaces a worse outcome (`ok` < `error` < `session_bad`).
+
+**Description**:
+- Parametrized: the row starts as `error` or `session_bad`; a successful command arrives.
+
+**Expected Outcome**:
+- The OPC write happens, the row keeps its worse status.
+
+### 3.2.4 Failure Replaces Ok
+**Summary**: A worse outcome replaces a better one.
+
+**Description**:
+- A successful command records `ok`; a second command for the same row fails (conversion).
+
+**Expected Outcome**:
+- `write_status = error`.
+
+### 3.2.5 Missing Row Is Not Created
+**Summary**: The gateway only updates rows; it never inserts one.
+
+**Description**:
+- The command targets model 3, which has no prediction row.
+
+**Expected Outcome**:
+- The OPC write happens; no row for model 3 exists afterwards.
+
+### 3.3.1 Unknown Server Is an Error
+**Summary**: A `server_id` with no live connection and no connector fails the write.
+
+**Expected Outcome**:
+- `write_status = error`; nothing is written to the server.
+
+### 3.3.2 Unusable Connector Is an Error
+**Summary**: Only active `opc_ua` connectors with a URL and full security are used.
+
+**Description**:
+- Parametrized: connector `inactive`, `not_opc_ua` (`type: modbus`), `no_policy`, `no_url`.
+
+**Expected Outcome**:
+- `write_status = error`; nothing is written to the server.
+
+### 3.3.3 Invalid Command Is an Error Without OPC Write
+**Summary**: Commands that cannot become a DataValue are rejected before touching the server.
+
+**Description**:
+- Parametrized: `conversion` (`"abc"` as `int`), `unknown_data_type` (`decimal128`), `missing_addr`.
+
+**Expected Outcome**:
+- `write_status = error`; nothing is written to the server.
+
+### 3.3.4 Bad Timestamp Skips the Write and Keeps the Row
+**Summary**: A non-ISO `timestamp` (with the timestamp flag on) fails before the OPC write.
+
+**Description**:
+- The command carries `13/10/2026 12:00`; a marker command is used to know it was handled.
+
+**Expected Outcome**:
+- Only the marker reaches the server; the row stays `pending` — the status cannot be recorded
+  either, since the row is found by that same timestamp.
+
+### 3.3.5 OPC Rejection Is an Error
+**Summary**: A write the server refuses is recorded as `error`.
+
+**Description**:
+- Parametrized: `not_writable` (read-only `Locked`), `unknown_node` (`ns=2;s=Missing`).
+
+**Expected Outcome**:
+- `write_status = error`; `Locked` keeps its value.
+
+### 3.3.6 Malformed JSON Is Dropped and the Consumer Keeps Going
+**Summary**: A payload that is not JSON is discarded without stopping the consumer.
+
+**Expected Outcome**:
+- The next valid command is written and recorded as `ok`.
+
+### 3.4.1 Write Reuses the Read Connection
+**Summary**: A server read by the pod is written through the same pooled connection.
+
+**Description**:
+- `plant-a` is only known through slot 1 (no connector in Mongo).
+
+**Expected Outcome**:
+- `write_status = ok`; the server has a single session.
+
+### 3.4.2 Write-Only Server Gets Its Own Connection
+**Summary**: A server only known as a connector is connected on the first write.
+
+**Expected Outcome**:
+- No session before the command; one session after; the pool holds the connector's settings.
+
+### 3.5.1 Minimum Pause Between Writes
+**Summary**: `WRITE_MIN_PAUSE_S` spaces consecutive writes.
+
+**Description**:
+- `WRITE_MIN_PAUSE_S=0.5`; three commands are published back to back.
+
+**Expected Outcome**:
+- The server receives the three writes at least ~0.5 s apart.
+
+### 3.5.2 Queue Group Writes Each Command Once
+**Summary**: Pods share the `gateway-writers` queue: each command is handled by a single pod.
+
+**Description**:
+- Two pods with the runtime; six commands.
+
+**Expected Outcome**:
+- The server receives exactly six writes.
+
+### 3.6.1 NATS Down at Startup Connects Later
+**Summary**: The runtime starts without NATS and connects from the main loop once NATS is up.
+
+**Description**:
+- `NATS_URL` points to a port with nothing listening; the pod starts and iterates once.
+- A NATS server starts on that port; the pod iterates again.
+
+**Expected Outcome**:
+- One `GATEWAY_NATS_CONSUMER_START_FAILED` notification for the whole outage (not one per retry).
+- After NATS starts, the consumer connects and a published command is recorded as `ok`.
+
+---
+
+## 4. On-demand gRPC ReadTag
+Source: `e2e/test_grpc_read_tag.py`
+
+Common setup: `GRPC_ENABLED=true`; slot 1 reads `plant-a`, so the pod's pool holds a connection.
+
+### 4.1.1 Reads a Tag of a Connected Server
+**Expected Outcome**:
+- `value` (as text), `opc_status_code = Good`, `source_timestamp` / `server_timestamp` in
+  `YYYY-MM-DD HH:MM:SS-0300`.
+- `persisted = false` and no collection is created: the slot lookup of `ReadTag` is not wired yet.
+
+### 4.1.2 Reads the Current Value
+**Expected Outcome**:
+- After a server-side change, `ReadTag` returns the new value.
+
+### 4.2.1 Unknown Address Is an Internal Error
+**Expected Outcome**:
+- gRPC `INTERNAL` whose details carry `BadNodeIdUnknown`.
+
+### 4.2.2 Server Not Connected by the Reads Is an Internal Error
+**Summary**: `ReadTag` has no connection settings of its own: it only reads servers the read path
+already connected (documents the current limitation).
+
+**Description**:
+- `plant-a` only exists as a connector; no slot.
+
+**Expected Outcome**:
+- gRPC `INTERNAL`; no session is opened on the server.
+
+### 4.2.3 gRPC Disabled by Default
+**Expected Outcome**:
+- Without `GRPC_ENABLED`, nothing listens on `GRPC_PORT` (`UNAVAILABLE`/`DEADLINE_EXCEEDED`).
+
+---
+
+## 5. Scheduled Active Reads
+Source: `e2e/test_active_read.py`
+
+`GatewayRuntime` builds `ActiveReadEngine` on the pod's pool and sink, but no loop calls
+`scheduled_tick` yet; these scenarios call it on the running pod's engine with the parsed slot
+(`slot_active_read.json`).
+
+### 5.1.1 Enabled Tag Is Read into Every Name
+**Expected Outcome**:
+- `Level` (`active_read.enabled`) is persisted in `active_tank` (model 30) and `active_mixer`
+  (model 40); `Temperature` (disabled) is not read.
+
+### 5.1.2 cycle_multiple Reads Every Nth Tick
+**Description**:
+- `cycle_multiple = 3`; six ticks.
+
+**Expected Outcome**:
+- Persisted documents after each tick: `0, 0, 1, 1, 1, 2`.
+
+### 5.2.1 Unreachable Server Does Not Stop the Tick
+**Expected Outcome**:
+- The tick does not raise; the unreachable server's tag is not persisted, the other server's is.
+
+### 5.2.2 Active Read Uses the Slot Connection Settings
+**Expected Outcome**:
+- The first tick opens the connection with the slot settings; later ticks reuse it (one session);
+  each tick persists the current value.
+
+---
+
+## 6. Client Certificates and OPC Security
+Source: `e2e/test_certificates.py`
+
+Common setup: `plant-s` exposes only Basic256Sha256 SignAndEncrypt. Client certificates are created
+like the platform stores them: cert (DER) and key (PEM) in MinIO bucket `opc-certificates`, plus an
+`opc_certificates` document (`kind: client`).
+
+### 6.1.1 Secured Read with Certificate from MinIO
+**Expected Outcome**:
+- Readings are persisted; the certificate is cached as `<CERT_CACHE_DIR>/7/cert.der` and `key.pem`.
+- No connection error is notified.
+
+### 6.1.2 Secured Write Resolved from Connectors and Certificates
+**Description**:
+- No slot: the connector (`client_certificate_id: 7`) is joined with `opc_certificates`.
+
+**Expected Outcome**:
+- `write_status = ok`; the server holds the value.
+
+### 6.2.1 Certificate Missing in MinIO
+**Expected Outcome**:
+- Notification `OPC_CONNECTION_ERROR_plant-s`; no reader; nothing cached; no readings.
+
+### 6.2.2 Unsupported Security Policy
+**Expected Outcome**:
+- Notification `OPC_CONNECTION_ERROR_plant-s` whose message names the unsupported mode/policy.
+
+### 6.2.3 No Certificate Cannot Reach a Secured-only Server
+**Description**:
+- The slot has `certificate: null` (connect without security, D5).
+
+**Expected Outcome**:
+- Notification `OPC_CONNECTION_ERROR_plant-s`; no readings.
+
+### 6.2.4 Inactive Certificate Write Fails
+**Description**:
+- The connector's certificate is `revoked`: the write connects without security.
+
+**Expected Outcome**:
+- `write_status = error`; nothing is written to the server.
+
+### 6.3.1 Certificate Rotation Reconnects with the New One
+**Description**:
+- Slot 1 switches from certificate 7 to certificate 8; the pod iterates.
+
+**Expected Outcome**:
+- The pooled connection is replaced and opened with certificate 8 (cached under `8/`).
+- Readings continue; the server ends with a single session.
+
+---
+
+## 7. Pod Lifecycle
+Source: `e2e/test_lifecycle.py`
+
+### 7.1.1 Shutdown Closes Sessions and Stops Consuming
+**Expected Outcome**:
+- After `GatewayRuntime.stop` + `Ingestor.shutdown`, the OPC server has no session.
+- A command published afterwards is not handled (row stays `pending`); value changes are not persisted.
+
+### 7.1.2 Idle Write Connection Is Closed and Reopened
+**Description**:
+- `IDLE_CONN_MAX_S=1`; `plant-a` is read (slot), `plant-b` only written (connector `2`).
+
+**Expected Outcome**:
+- After the idle time, `idle_sweep` closes the `plant-b` session; the subscribed `plant-a`
+  connection is pinned and stays open.
+- The next write to `plant-b` reconnects and is recorded as `ok`.
+
+### 7.2.1 Redis Unreachable Fails the Startup
+**Expected Outcome**:
+- `prepare_ingestor` raises a connection error (`app.py` then stops the process).
+
+### 7.2.2 Restarted Pod Takes Its Slot Back
+**Description**:
+- `pod-a` (`LEASE_TTL=2`) stops; a new `pod-a` starts after the lease expired.
+
+**Expected Outcome**:
+- The restarted pod leases slot `1` and its readings reach MongoDB.

e2e/test_active_read.py

@@ -0,0 +1,84 @@
+"""E2E scenarios — 5. Scheduled active reads. See ``e2e/scenarios.md``.
+
+``GatewayRuntime`` builds the ``ActiveReadEngine`` with the pod's pool and sink, but no
+loop calls ``scheduled_tick`` yet (its wiring is pending in ``gateway_runtime.py``), so
+these scenarios call it on the running pod's engine with the parsed slot.
+"""
+
+import pytest
+
+from e2e.helpers import load_scenario_input, readings, stays_false, wait_until
+from e2e.opc_server import free_port
+from ingestor.io.slots import parse_slot
+
+pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
+
+
+async def test_scenario_5_1_1_enabled_tag_is_read_into_every_name(opc_servers, pods, mongo_db):
+    server = await opc_servers('plant-a', {'Level': 2.5, 'Temperature': 20.0})
+    slot = parse_slot(load_scenario_input('slot_active_read.json', {'plant-a': server}))
+    pod = await pods('pod-a')
+
+    await pod.runtime.active_read.scheduled_tick(slot)
+
+    [tank] = readings(mongo_db, 'active_tank', name='tank_level')
+    [mixer] = readings(mongo_db, 'active_mixer', name='mixer_inlet_level')
+    assert tank['value'] == mixer['value'] == 2.5
+    assert (tank['model_id'], mixer['model_id']) == ('30', '40')
+    assert tank['tag'] == 'ns=2;s=Level'
+    assert readings(mongo_db, 'active_tank', name='tank_temperature') == []  # not enabled
+
+
+async def test_scenario_5_1_2_cycle_multiple_reads_every_nth_tick(opc_servers, pods, mongo_db):
+    server = await opc_servers('plant-a', {'Level': 2.5, 'Temperature': 20.0})
+    raw = load_scenario_input('slot_active_read.json', {'plant-a': server})
+    raw['plant-a']['tags']['ns=2;s=Level']['active_read']['cycle_multiple'] = 3
+    slot = parse_slot(raw)
+    pod = await pods('pod-a')
+
+    counts = []
+    for _ in range(6):
+        await pod.runtime.active_read.scheduled_tick(slot)
+        counts.append(len(readings(mongo_db, 'active_tank', name='tank_level')))
+
+    assert counts == [0, 0, 1, 1, 1, 2]
+
+
+async def test_scenario_5_2_1_unreachable_server_does_not_stop_the_tick(
+    opc_servers, pods, mongo_db
+):
+    server = await opc_servers('plant-a', {'Level': 2.5, 'Temperature': 20.0})
+    raw = load_scenario_input('slot_active_read.json', {'plant-a': server})
+    down = {**raw['plant-a'], 'name': 'plant-down', 'server_id': '9'}
+    down['url'] = f'opc.tcp://127.0.0.1:{free_port()}'
+    down['tags'] = {
+        'ns=2;s=Level': {
+            **raw['plant-a']['tags']['ns=2;s=Level'],
+            'names': [{'collection': 'active_down', 'tag_name': 'down_level', 'model_id': '90'}],
+        }
+    }
+    slot = parse_slot({'plant-down': down, 'plant-a': raw['plant-a']})
+    pod = await pods('pod-a')
+
+    await pod.runtime.active_read.scheduled_tick(slot)  # does not raise
+
+    assert readings(mongo_db, 'active_down') == []
+    await wait_until(lambda: readings(mongo_db, 'active_tank', name='tank_level'))
+
+
+async def test_scenario_5_2_2_active_read_uses_the_slot_connection_settings(
+    opc_servers, pods, mongo_db
+):
+    server = await opc_servers('plant-a', {'Level': 2.5, 'Temperature': 20.0})
+    slot = parse_slot(load_scenario_input('slot_active_read.json', {'plant-a': server}))
+    pod = await pods('pod-a')
+    assert pod.manager.pool.config_for('1') is None
+
+    await pod.runtime.active_read.scheduled_tick(slot)
+
+    assert pod.manager.pool.config_for('1')['url'] == server.url
+    assert server.session_count() == 1
+    await server.set('Level', 3.0)
+    await pod.runtime.active_read.scheduled_tick(slot)
+    assert await stays_false(lambda: server.session_count() != 1, duration=0.5)
+    assert [d['value'] for d in readings(mongo_db, 'active_tank', name='tank_level')] == [2.5, 3.0]

e2e/test_certificates.py

@@ -0,0 +1,165 @@
+"""E2E scenarios — 6. Client certificates and OPC security (MinIO + opc_certificates). See ``e2e/scenarios.md``."""
+
+import pytest
+import pytest_asyncio
+
+from e2e.helpers import (
+    insert_prediction,
+    load_scenario_input,
+    notifications,
+    readings,
+    stays_false,
+    wait_until,
+    write_status,
+)
+
+pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
+
+TIMESTAMP = '2026-10-13T12:00:00-03:00'
+SECURED = {'mode': 'SignAndEncrypt', 'policy': 'Basic256Sha256'}
+
+
+@pytest_asyncio.fixture
+async def secured_server(opc_servers):
+    """plant-s: only Basic256Sha256 SignAndEncrypt endpoints, accepts any client certificate."""
+    return await opc_servers('plant-s', {'Temperature': 20.0, 'Setpoint': 0.0}, secured=True)
+
+
+def secured_slot(server, certificate, **security):
+    slot = load_scenario_input('slot_secured.json', {'plant-s': server})
+    slot['plant-s']['certificate'] = certificate
+    slot['plant-s']['security'] = {**SECURED, **security}
+    return slot
+
+
+def secured_connector(server, certificate_id):
+    return {
+        'connector_id': '7',
+        'type': 'opc_ua',
+        'name': 'plant-s',
+        'url': server.url,
+        'status': 'Active',
+        'subscription_period_ms': 100,
+        'security': {**SECURED, 'client_certificate_id': certificate_id},
+    }
+
+
+def write_command():
+    return load_scenario_input(
+        'write_command.json', server_id='7', model_id='2', timestamp=TIMESTAMP
+    )
+
+
+# 6.1 Secured connections ---------------------------------------------------------------
+
+
+async def test_scenario_6_1_1_secured_read_with_certificate_from_minio(
+    secured_server, client_certificate, pods, put_slot, mongo_db, cert_cache_dir
+):
+    certificate = await client_certificate(7)
+    put_slot(1, secured_slot(secured_server, certificate))
+    pod = await pods('pod-a')
+
+    await pod.tick()
+
+    await secured_server.set('Temperature', 21.0)
+    await wait_until(lambda: readings(mongo_db, 'raw_secured', value=21.0))
+    assert sorted(p.name for p in (cert_cache_dir / '7').iterdir()) == ['cert.der', 'key.pem']
+    assert notifications(mongo_db, 'OPC_CONNECTION_ERROR') == []
+
+
+async def test_scenario_6_1_2_secured_write_resolved_from_connectors_and_certificates(
+    secured_server, client_certificate, pods, mongo_db, postgres_engine, publish_command
+):
+    await client_certificate(7)
+    mongo_db['connectors'].insert_one(secured_connector(secured_server, 7))
+    insert_prediction(postgres_engine, 2, TIMESTAMP, 'pending')
+    await pods('pod-a')
+
+    await publish_command(write_command())
+
+    await wait_until(lambda: write_status(postgres_engine, 2, TIMESTAMP) == 'ok')
+    assert (await secured_server.read('Setpoint')).Value.Value == 1.25
+
+
+# 6.2 Failures --------------------------------------------------------------------------
+
+
+async def test_scenario_6_2_1_certificate_missing_in_minio(
+    secured_server, client_certificate, pods, put_slot, mongo_db, cert_cache_dir
+):
+    certificate = await client_certificate(7, upload=False)
+    put_slot(1, secured_slot(secured_server, certificate))
+    pod = await pods('pod-a')
+
+    await pod.tick()
+
+    assert notifications(mongo_db, 'OPC_CONNECTION_ERROR_plant-s')
+    assert pod.manager.opc_managers == {}
+    assert not list((cert_cache_dir / '7').glob('*.der'))
+    assert await stays_false(lambda: readings(mongo_db, 'raw_secured'))
+
+
+async def test_scenario_6_2_2_unsupported_security_policy(
+    secured_server, client_certificate, pods, put_slot, mongo_db
+):
+    certificate = await client_certificate(7)
+    put_slot(1, secured_slot(secured_server, certificate, policy='Basic999'))
+    pod = await pods('pod-a')
+
+    await pod.tick()
+
+    [error] = notifications(mongo_db, 'OPC_CONNECTION_ERROR_plant-s')
+    assert 'Unsupported OPC security' in error['message']
+    assert pod.manager.opc_managers == {}
+
+
+async def test_scenario_6_2_3_no_certificate_cannot_reach_a_secured_only_server(
+    secured_server, pods, put_slot, mongo_db
+):
+    put_slot(1, secured_slot(secured_server, None))
+    pod = await pods('pod-a')
+
+    await pod.tick()
+
+    assert notifications(mongo_db, 'OPC_CONNECTION_ERROR_plant-s')
+    assert await stays_false(lambda: readings(mongo_db, 'raw_secured'))
+
+
+async def test_scenario_6_2_4_inactive_certificate_write_fails(
+    secured_server, client_certificate, pods, mongo_db, postgres_engine, publish_command
+):
+    await client_certificate(7, status='revoked')
+    mongo_db['connectors'].insert_one(secured_connector(secured_server, 7))
+    insert_prediction(postgres_engine, 2, TIMESTAMP, 'pending')
+    await pods('pod-a')
+
+    await publish_command(write_command())
+
+    await wait_until(lambda: write_status(postgres_engine, 2, TIMESTAMP) == 'error')
+    assert secured_server.writes == []
+
+
+# 6.3 Rotation --------------------------------------------------------------------------
+
+
+async def test_scenario_6_3_1_certificate_rotation_reconnects_with_the_new_one(
+    secured_server, client_certificate, pods, put_slot, mongo_db, cert_cache_dir
+):
+    first = await client_certificate(7)
+    put_slot(1, secured_slot(secured_server, first))
+    pod = await pods('pod-a')
+    await pod.tick()
+    await wait_until(lambda: readings(mongo_db, 'raw_secured'))
+    first_client = pod.manager.pool._connections['7'][0]
+
+    second = await client_certificate(8)
+    put_slot(1, secured_slot(secured_server, second))
+    await pod.tick()
+
+    assert pod.manager.pool._connections['7'][0] is not first_client
+    assert pod.manager.pool.config_for('7')['certificate']['certificate_id'] == 8
+    assert (cert_cache_dir / '8' / 'cert.der').is_file()
+    await secured_server.set('Temperature', 22.0)
+    await wait_until(lambda: readings(mongo_db, 'raw_secured', value=22.0))
+    await wait_until(lambda: secured_server.session_count() == 1)

e2e/test_grpc_read_tag.py

@@ -0,0 +1,113 @@
+"""E2E scenarios — 4. On-demand gRPC ReadTag. See ``e2e/scenarios.md``."""
+
+import os
+import re
+
+import grpc
+import pytest
+import pytest_asyncio
+
+from e2e.helpers import load_scenario_input, readings, wait_until
+from ingestor.proto import opc_gateway_pb2, opc_gateway_pb2_grpc
+
+pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
+
+FORMATTED_TIMESTAMP = re.compile(r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}-0300$')
+
+
+@pytest_asyncio.fixture
+async def grpc_stub():
+    """Client of the gateway's gRPC server (GRPC_PORT of the test environment)."""
+    channel = grpc.aio.insecure_channel(f'127.0.0.1:{os.environ["GRPC_PORT"]}')
+    yield opc_gateway_pb2_grpc.OpcGatewayStub(channel)
+    await channel.close()
+
+
+async def _reading_pod(opc_servers, pods, put_slot, mongo_db):
+    server = await opc_servers('plant-a', {'Temperature': 20.0, 'Level': 3.25})
+    put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server}))
+    pod = await pods('pod-a', GRPC_ENABLED='true')
+    await pod.tick()
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a'))
+    return server, pod
+
+
+async def test_scenario_4_1_1_reads_a_tag_of_a_connected_server(
+    opc_servers, pods, put_slot, mongo_db, grpc_stub
+):
+    await _reading_pod(opc_servers, pods, put_slot, mongo_db)
+    persisted_before = mongo_db.list_collection_names()
+
+    response = await grpc_stub.ReadTag(
+        opc_gateway_pb2.ReadTagRequest(server_id='1', addr='ns=2;s=Level', data_type='float')
+    )
+
+    assert response.value == '3.25'
+    assert response.opc_status_code == 'Good'
+    assert FORMATTED_TIMESTAMP.match(response.source_timestamp), response.source_timestamp
+    assert FORMATTED_TIMESTAMP.match(response.server_timestamp), response.server_timestamp
+    assert response.persisted is False  # no slot lookup wired yet: nothing is written
+    assert mongo_db.list_collection_names() == persisted_before
+
+
+async def test_scenario_4_1_2_reads_the_current_value(
+    opc_servers, pods, put_slot, mongo_db, grpc_stub
+):
+    server, _ = await _reading_pod(opc_servers, pods, put_slot, mongo_db)
+    await server.set('Level', 4.5)
+
+    response = await grpc_stub.ReadTag(
+        opc_gateway_pb2.ReadTagRequest(server_id='1', addr='ns=2;s=Level', data_type='float')
+    )
+
+    assert response.value == '4.5'
+
+
+async def test_scenario_4_2_1_unknown_address_is_internal_error(
+    opc_servers, pods, put_slot, mongo_db, grpc_stub
+):
+    await _reading_pod(opc_servers, pods, put_slot, mongo_db)
+
+    with pytest.raises(grpc.aio.AioRpcError) as error:
+        await grpc_stub.ReadTag(
+            opc_gateway_pb2.ReadTagRequest(server_id='1', addr='ns=2;s=Missing', data_type='float')
+        )
+
+    assert error.value.code() == grpc.StatusCode.INTERNAL
+    assert 'BadNodeIdUnknown' in error.value.details()
+
+
+async def test_scenario_4_2_2_server_not_connected_by_the_reads_is_internal_error(
+    opc_servers, pods, mongo_db, grpc_stub
+):
+    server = await opc_servers('plant-a', {'Level': 3.25})
+    mongo_db['connectors'].insert_one(
+        load_scenario_input('connector_plant_a.json', {'plant-a': server})
+    )
+    await pods('pod-a', GRPC_ENABLED='true')  # no slot: the pool has no connection to plant-a
+
+    with pytest.raises(grpc.aio.AioRpcError) as error:
+        await grpc_stub.ReadTag(
+            opc_gateway_pb2.ReadTagRequest(server_id='1', addr='ns=2;s=Level', data_type='float')
+        )
+
+    assert error.value.code() == grpc.StatusCode.INTERNAL
+    assert server.session_count() == 0
+
+
+async def test_scenario_4_2_3_grpc_disabled_by_default(
+    opc_servers, pods, put_slot, mongo_db, grpc_stub
+):
+    server = await opc_servers('plant-a', {'Temperature': 20.0})
+    put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server}))
+    pod = await pods('pod-a')
+    await pod.tick()
+
+    with pytest.raises(grpc.aio.AioRpcError) as error:
+        await grpc_stub.ReadTag(
+            opc_gateway_pb2.ReadTagRequest(server_id='1', addr='ns=2;s=Temperature'),
+            timeout=2,
+        )
+
+    assert error.value.code() in (grpc.StatusCode.UNAVAILABLE, grpc.StatusCode.DEADLINE_EXCEEDED)
+    assert pod.runtime.grpc_server is None

e2e/test_lifecycle.py

@@ -0,0 +1,93 @@
+"""E2E scenarios — 7. Pod lifecycle: shutdown, idle connections, startup failures. See ``e2e/scenarios.md``."""
+
+import asyncio
+
+import pytest
+
+from e2e.helpers import (
+    Pod,
+    insert_prediction,
+    load_scenario_input,
+    readings,
+    stays_false,
+    wait_until,
+    write_status,
+)
+from e2e.opc_server import free_port
+
+pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
+
+TIMESTAMP = '2026-10-13T12:00:00-03:00'
+
+
+async def test_scenario_7_1_1_shutdown_closes_sessions_and_stops_consuming(
+    opc_servers, pods, put_slot, mongo_db, postgres_engine, publish_command
+):
+    server = await opc_servers('plant-a', {'Temperature': 20.0, 'Setpoint': 0.0})
+    put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server}))
+    insert_prediction(postgres_engine, 2, TIMESTAMP, 'pending')
+    pod = await pods('pod-a')
+    await pod.tick()
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a'))
+    assert server.session_count() == 1
+
+    await pod.stop()
+
+    await wait_until(lambda: server.session_count() == 0)
+    await publish_command(load_scenario_input('write_command.json'))
+    assert await stays_false(lambda: write_status(postgres_engine, 2, TIMESTAMP) != 'pending')
+    await server.set('Temperature', 50.0)
+    assert await stays_false(lambda: readings(mongo_db, 'raw_plant_a', value=50.0))
+
+
+async def test_scenario_7_1_2_idle_write_connection_is_closed_and_reopened(
+    opc_servers, pods, put_slot, mongo_db, postgres_engine, publish_command
+):
+    reader = await opc_servers('plant-a', {'Temperature': 20.0})
+    writer = await opc_servers('plant-b', {'Setpoint': 0.0})
+    put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': reader}))
+    connector = load_scenario_input('connector_plant_a.json', {'plant-a': writer})
+    mongo_db['connectors'].insert_one({**connector, 'connector_id': '2', 'name': 'plant-b'})
+    insert_prediction(postgres_engine, 2, TIMESTAMP, 'pending')
+    pod = await pods('pod-a', IDLE_CONN_MAX_S='1')
+    await pod.tick()
+
+    await publish_command(load_scenario_input('write_command.json', server_id='2'))
+    await wait_until(lambda: write_status(postgres_engine, 2, TIMESTAMP) == 'ok')
+    assert writer.session_count() == 1
+
+    await asyncio.sleep(1.5)
+    await pod.tick()  # idle_sweep
+
+    await wait_until(lambda: writer.session_count() == 0)
+    assert reader.session_count() == 1  # subscribed: pinned, never idle
+
+    insert_prediction(postgres_engine, 3, TIMESTAMP, 'pending')
+    await publish_command(load_scenario_input('write_command.json', server_id='2', model_id='3'))
+    await wait_until(lambda: write_status(postgres_engine, 3, TIMESTAMP) == 'ok')
+    assert writer.session_count() == 1
+
+
+async def test_scenario_7_2_1_redis_unreachable_fails_the_startup(monkeypatch):
+    monkeypatch.setenv('REDIS_PORT', str(free_port()))
+
+    with pytest.raises(Exception, match='(?i)connect'):
+        await Pod('pod-a').start()
+
+
+async def test_scenario_7_2_2_restarted_pod_takes_its_slot_back(
+    opc_servers, pods, put_slot, mongo_db, redis_client
+):
+    server = await opc_servers('plant-a', {'Temperature': 20.0})
+    put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server}))
+    pod = await pods('pod-a', LEASE_TTL='2', HEARTBEAT_TTL='2')
+    await pod.tick()
+    await pod.stop()
+
+    await wait_until(lambda: redis_client.get('lease:opc_tags:1') is None, timeout=5)
+    restarted = await pods('pod-a', LEASE_TTL='2', HEARTBEAT_TTL='2')
+    await restarted.tick()
+
+    assert restarted.slots == {'1'}
+    await server.set('Temperature', 31.0)
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=31.0))

e2e/test_slot_leasing.py

@@ -0,0 +1,193 @@
+"""E2E scenarios — 1. Slot leasing and load balancing (Redis). See ``e2e/scenarios.md``."""
+
+import asyncio
+
+import pytest
+
+from e2e.helpers import (
+    load_scenario_input,
+    notifications,
+    readings,
+    redis_json,
+    stays_false,
+    wait_until,
+)
+
+pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
+
+
+async def _two_slots(opc_servers, put_slot):
+    """Slot 1 reads plant-a, slot 2 reads plant-b."""
+    plant_a = await opc_servers('plant-a', {'Temperature': 20.0})
+    plant_b = await opc_servers('plant-b', {'Flow': 3.0})
+    both = load_scenario_input('slot_two_servers.json', {'plant-a': plant_a, 'plant-b': plant_b})
+    put_slot(1, {'plant-a': both['plant-a']})
+    put_slot(2, {'plant-b': both['plant-b']})
+    return plant_a, plant_b
+
+
+# 1.1 Acquisition ----------------------------------------------------------------------
+
+
+async def test_scenario_1_1_1_single_pod_leases_the_slot_and_heartbeats(
+    opc_servers, pods, put_slot, redis_client, mongo_db
+):
+    server = await opc_servers('plant-a', {'Temperature': 20.0})
+    put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server}))
+
+    pod = await pods('pod-a')
+    await pod.tick()
+
+    assert pod.slots == {'1'}
+    assert redis_json(redis_client, 'lease:opc_tags:1') == 'pod-a'
+    assert 0 < redis_client.ttl('lease:opc_tags:1') <= 10
+    assert redis_json(redis_client, 'heartbeat:ingestor:pod-a') == 1
+    assert 0 < redis_client.ttl('heartbeat:ingestor:pod-a') <= 20
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a'))
+
+
+async def test_scenario_1_1_2_idle_pod_takes_a_slot_created_after_startup(
+    opc_servers, pods, put_slot, redis_client, mongo_db
+):
+    server = await opc_servers('plant-a', {'Temperature': 20.0})
+    pod = await pods('pod-a')
+    await pod.tick()
+    assert pod.slots == set()
+
+    put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server}))
+    await pod.tick()
+
+    assert pod.slots == {'1'}
+    assert redis_json(redis_client, 'lease:opc_tags:1') == 'pod-a'
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a'))
+
+
+# 1.2 Balancing and failover ------------------------------------------------------------
+
+
+async def test_scenario_1_2_1_two_pods_split_two_slots(
+    opc_servers, pods, put_slot, redis_client, mongo_db
+):
+    plant_a, plant_b = await _two_slots(opc_servers, put_slot)
+
+    pod_a = await pods('pod-a')
+    await pod_a.tick()
+    assert pod_a.slots == {'1', '2'}  # alone: takes every slot
+
+    pod_b = await pods('pod-b')
+    await pod_b.tick()
+    assert pod_b.slots == set()  # every slot is leased
+    await pod_a.tick()  # pod-a now sees a pod without slot and gives one back
+    assert pod_a.slots == {'1'}
+    await pod_b.tick()
+    assert pod_b.slots == {'2'}
+
+    assert redis_json(redis_client, 'lease:opc_tags:1') == 'pod-a'
+    assert redis_json(redis_client, 'lease:opc_tags:2') == 'pod-b'
+    assert set(pod_a.manager.opc_managers) == {'plant-a'}
+    assert set(pod_b.manager.opc_managers) == {'plant-b'}
+    await plant_a.set('Temperature', 25.0)
+    await plant_b.set('Flow', 4.5)
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=25.0))
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_b', value=4.5))
+
+
+async def test_scenario_1_2_2_extra_pod_stays_idle(opc_servers, pods, put_slot, redis_client):
+    await _two_slots(opc_servers, put_slot)
+    pod_a, pod_b = await pods('pod-a'), await pods('pod-b')
+    await pod_a.tick()
+    await pod_b.tick()
+    await pod_a.tick()
+    await pod_b.tick()
+
+    pod_c = await pods('pod-c')
+    for _ in range(2):
+        for pod in (pod_a, pod_b, pod_c):
+            await pod.tick()
+
+    assert pod_c.slots == set()
+    assert pod_a.slots | pod_b.slots == {'1', '2'}
+    assert pod_a.slots.isdisjoint(pod_b.slots)
+    assert {
+        redis_json(redis_client, 'lease:opc_tags:1'),
+        redis_json(redis_client, 'lease:opc_tags:2'),
+    } == {
+        'pod-a',
+        'pod-b',
+    }
+
+
+async def test_scenario_1_2_3_failover_after_lease_expiry(
+    opc_servers, pods, put_slot, redis_client, mongo_db
+):
+    server = await opc_servers('plant-a', {'Temperature': 20.0})
+    put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server}))
+    pod_a = await pods('pod-a', LEASE_TTL='2', HEARTBEAT_TTL='2')
+    await pod_a.tick()
+    pod_b = await pods('pod-b')
+    await pod_b.tick()
+    assert (pod_a.slots, pod_b.slots) == ({'1'}, set())
+
+    await pod_a.stop()  # crash-like: the lease is not released, it expires
+    assert redis_json(redis_client, 'lease:opc_tags:1') == 'pod-a'
+    await wait_until(lambda: redis_json(redis_client, 'lease:opc_tags:1') is None, timeout=5)
+    await pod_b.tick()
+
+    assert pod_b.slots == {'1'}
+    assert redis_json(redis_client, 'lease:opc_tags:1') == 'pod-b'
+    await server.set('Temperature', 30.0)
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=30.0))
+
+
+# 1.3 Slot changes and invalid slots ----------------------------------------------------
+
+
+async def test_scenario_1_3_1_deleted_slot_stops_the_reads(
+    opc_servers, pods, put_slot, redis_client, mongo_db
+):
+    server = await opc_servers('plant-a', {'Temperature': 20.0})
+    put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server}))
+    pod = await pods('pod-a')
+    await pod.tick()
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a'))
+
+    redis_client.delete('slot:opc_tags:1')
+    await pod.tick()
+
+    assert pod.slots == set()
+    assert pod.manager.opc_managers == {}
+    await server.set('Temperature', 99.0)
+    assert await stays_false(lambda: readings(mongo_db, 'raw_plant_a', value=99.0))
+
+
+async def test_scenario_1_3_2_invalid_slot_is_notified_and_not_read(
+    opc_servers, pods, put_slot, mongo_db
+):
+    server = await opc_servers('plant-a', {'Temperature': 20.0})
+    put_slot(1, load_scenario_input('slot_v1_invalid.json', {'plant-a': server}))
+
+    pod = await pods('pod-a')
+    await pod.tick()
+
+    assert pod.slots == set()
+    assert notifications(mongo_db, 'OPC_SLOT_CONFIG_ERROR_1')
+    assert await stays_false(lambda: readings(mongo_db, 'raw_plant_a'))
+
+
+async def test_scenario_1_3_3_fixed_slot_is_read_once_its_lease_expires(
+    opc_servers, pods, put_slot, mongo_db
+):
+    server = await opc_servers('plant-a', {'Temperature': 20.0})
+    put_slot(1, load_scenario_input('slot_v1_invalid.json', {'plant-a': server}))
+    pod = await pods('pod-a', LEASE_TTL='2')
+    await pod.tick()
+    assert pod.slots == set()
+
+    put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server}))
+    await pod.tick()
+    assert pod.slots == set()  # still leased by the failed acquisition
+
+    await asyncio.sleep(2.5)
+    await pod.tick()
+    assert pod.slots == {'1'}
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a'))

e2e/test_subscription_reads.py

@@ -0,0 +1,266 @@
+"""E2E scenarios — 2. Subscription reads (OPC UA → MongoDB). See ``e2e/scenarios.md``."""
+
+import re
+
+import pytest
+
+from e2e.helpers import load_scenario_input, notifications, readings, stays_false, wait_until
+from e2e.opc_server import free_port
+
+pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
+
+# DATETIME_FORMAT_WITH_TZ in OPC_TIMEZONE (sientia_do), e.g. 2026-10-29 19:48:51-0300
+READING_TIMESTAMP = re.compile(r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}-0300$')
+
+
+async def _read_slot(opc_servers, pods, put_slot, slot_file, variables):
+    server = await opc_servers('plant-a', variables)
+    slot = load_scenario_input(slot_file, {'plant-a': server})
+    put_slot(1, slot)
+    pod = await pods('pod-a')
+    await pod.tick()
+    return server, slot, pod
+
+
+# 2.1 Persistence -----------------------------------------------------------------------
+
+
+async def test_scenario_2_1_1_value_change_is_persisted(opc_servers, pods, put_slot, mongo_db):
+    server, _, _ = await _read_slot(
+        opc_servers, pods, put_slot, 'slot_single_server.json', {'Temperature': 20.0}
+    )
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=20.0))  # initial value
+
+    await server.set('Temperature', 21.5)
+
+    [doc] = await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=21.5))
+    assert doc['tag'] == 'ns=2;s=Temperature'
+    assert doc['name'] == 'temperature'
+    assert doc['model_id'] == '10'
+    assert READING_TIMESTAMP.match(doc['timestamp']), doc['timestamp']
+    assert doc['inserted_at'] is not None
+    assert set(doc) == {'tag', 'name', 'model_id', 'timestamp', 'value', 'inserted_at'}
+
+
+async def test_scenario_2_1_2_one_address_fans_out_to_every_name(
+    opc_servers, pods, put_slot, mongo_db
+):
+    server, _, _ = await _read_slot(
+        opc_servers, pods, put_slot, 'slot_multi_name.json', {'Pressure': 1.0}
+    )
+
+    await server.set('Pressure', 2.5)
+
+    [furnace] = await wait_until(lambda: readings(mongo_db, 'raw_furnace', value=2.5))
+    [boiler] = await wait_until(lambda: readings(mongo_db, 'raw_boiler', value=2.5))
+    assert (furnace['name'], furnace['model_id']) == ('furnace_pressure', '10')
+    assert (boiler['name'], boiler['model_id']) == ('boiler_inlet_pressure', '20')
+    assert furnace['tag'] == boiler['tag'] == 'ns=2;s=Pressure'
+    assert furnace['timestamp'] == boiler['timestamp']
+
+
+async def test_scenario_2_1_3_two_servers_in_one_slot(opc_servers, pods, put_slot, mongo_db):
+    plant_a = await opc_servers('plant-a', {'Temperature': 20.0})
+    plant_b = await opc_servers('plant-b', {'Flow': 3.0})
+    put_slot(
+        1, load_scenario_input('slot_two_servers.json', {'plant-a': plant_a, 'plant-b': plant_b})
+    )
+    pod = await pods('pod-a')
+    await pod.tick()
+
+    await plant_a.set('Temperature', 22.0)
+    await plant_b.set('Flow', 3.5)
+
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=22.0))
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_b', value=3.5))
+    assert set(pod.manager.opc_managers) == {'plant-a', 'plant-b'}
+
+
+async def test_scenario_2_1_4_values_keep_their_opc_type(opc_servers, pods, put_slot, mongo_db):
+    variables = {
+        'Temperature': (20.0, 'double', True),
+        'Counter': (1, 'int', True),
+        'Running': (False, 'bool', True),
+        'Batch': ('B-001', 'string', True),
+    }
+    server, _, _ = await _read_slot(opc_servers, pods, put_slot, 'slot_data_types.json', variables)
+
+    await server.set('Temperature', 20.25)
+    await server.set('Counter', 7)
+    await server.set('Running', True)
+    await server.set('Batch', 'B-002')
+
+    expected = {'temperature': 20.25, 'counter': 7, 'running': True, 'batch': 'B-002'}
+    for name, value in expected.items():
+        [doc] = await wait_until(
+            lambda n=name, v=value: readings(mongo_db, 'raw_types', name=n, value=v)
+        )
+        assert type(doc['value']) is type(value)
+
+
+# 2.2 Slot changes ----------------------------------------------------------------------
+
+
+async def test_scenario_2_2_1_tag_added_to_the_slot_is_subscribed(
+    opc_servers, pods, put_slot, mongo_db
+):
+    server, slot, pod = await _read_slot(
+        opc_servers, pods, put_slot, 'slot_single_server.json', {'Temperature': 20.0, 'Level': 1.0}
+    )
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', name='temperature'))
+
+    slot['plant-a']['tags']['ns=2;s=Level'] = {
+        'frequency': 1000,
+        'names': [{'collection': 'raw_plant_a', 'tag_name': 'level', 'model_id': '10'}],
+    }
+    put_slot(1, slot)
+    await pod.tick()
+
+    await server.set('Level', 1.75)
+    await server.set('Temperature', 23.0)
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', name='level', value=1.75))
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', name='temperature', value=23.0))
+
+
+async def test_scenario_2_2_2_tag_removed_from_the_slot_is_no_longer_read(
+    opc_servers, pods, put_slot, mongo_db
+):
+    server = await opc_servers('plant-a', {'Temperature': 20.0, 'Level': 1.0})
+    slot = load_scenario_input('slot_single_server.json', {'plant-a': server})
+    slot['plant-a']['tags']['ns=2;s=Level'] = {
+        'frequency': 1000,
+        'names': [{'collection': 'raw_plant_a', 'tag_name': 'level', 'model_id': '10'}],
+    }
+    put_slot(1, slot)
+    pod = await pods('pod-a')
+    await pod.tick()
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', name='level'))
+
+    del slot['plant-a']['tags']['ns=2;s=Level']
+    put_slot(1, slot)
+    await pod.tick()
+
+    await server.set('Level', 9.0)
+    await server.set('Temperature', 24.0)
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', name='temperature', value=24.0))
+    assert await stays_false(lambda: readings(mongo_db, 'raw_plant_a', name='level', value=9.0))
+
+
+async def test_scenario_2_2_3_name_added_to_a_tag_receives_its_readings(
+    opc_servers, pods, put_slot, mongo_db
+):
+    server, slot, pod = await _read_slot(
+        opc_servers, pods, put_slot, 'slot_single_server.json', {'Temperature': 20.0}
+    )
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a'))
+
+    slot['plant-a']['tags']['ns=2;s=Temperature']['names'].append(
+        {'collection': 'raw_dryer', 'tag_name': 'dryer_temperature', 'model_id': '50'}
+    )
+    put_slot(1, slot)
+    await pod.tick()
+
+    await server.set('Temperature', 26.0)
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=26.0))
+    [doc] = await wait_until(lambda: readings(mongo_db, 'raw_dryer', value=26.0))
+    assert (doc['name'], doc['model_id']) == ('dryer_temperature', '50')
+
+
+async def test_scenario_2_2_4_server_url_change_reconnects_to_the_new_server(
+    opc_servers, pods, put_slot, mongo_db
+):
+    old = await opc_servers('plant-a', {'Temperature': 20.0})
+    new = await opc_servers('plant-a-new', {'Temperature': 40.0})
+    slot = load_scenario_input('slot_single_server.json', {'plant-a': old})
+    put_slot(1, slot)
+    pod = await pods('pod-a')
+    await pod.tick()
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=20.0))
+
+    slot['plant-a']['url'] = new.url
+    put_slot(1, slot)
+    await pod.tick()
+
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=40.0))
+    await new.set('Temperature', 41.0)
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=41.0))
+    await wait_until(lambda: old.session_count() == 0)
+    await old.set('Temperature', 21.0)
+    assert await stays_false(lambda: readings(mongo_db, 'raw_plant_a', value=21.0))
+
+
+# 2.3 Failures and resilience -----------------------------------------------------------
+
+
+async def test_scenario_2_3_1_unreachable_server_does_not_stop_the_others(
+    opc_servers, pods, put_slot, mongo_db
+):
+    plant_a = await opc_servers('plant-a', {'Temperature': 20.0})
+    plant_b = await opc_servers('plant-b', {'Flow': 3.0})
+    slot = load_scenario_input('slot_two_servers.json', {'plant-a': plant_a, 'plant-b': plant_b})
+    slot['plant-b']['url'] = f'opc.tcp://127.0.0.1:{free_port()}'  # nothing listens there
+    put_slot(1, slot)
+
+    pod = await pods('pod-a')
+    await pod.tick()
+
+    assert notifications(mongo_db, 'OPC_CONNECTION_ERROR_plant-b')
+    assert set(pod.manager.opc_managers) == {'plant-a'}
+    await plant_a.set('Temperature', 27.0)
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=27.0))
+
+
+async def test_scenario_2_3_2_tag_without_data_raises_a_warning(
+    opc_servers, pods, put_slot, mongo_db
+):
+    _, _, pod = await _read_slot(
+        opc_servers, pods, put_slot, 'slot_single_server.json', {'Temperature': 20.0}
+    )
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a'))
+
+    await pod.tick(5)  # POLL_INTERVAL 1 s / frequency 1000 ms: one cycle per iteration
+
+    [warning] = notifications(mongo_db, 'TAG_ns=2;s=Temperature:temperature_LISTENNING_STOPPED')
+    assert warning['level'] in ('WARNING', 'warning')
+
+
+async def test_scenario_2_3_3_silent_server_is_reconnected_after_15_cycles(
+    opc_servers, pods, put_slot, mongo_db
+):
+    server, _, pod = await _read_slot(
+        opc_servers, pods, put_slot, 'slot_single_server.json', {'Temperature': 20.0}
+    )
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a'))
+    first_client = pod.manager.pool._connections['1'][0]
+
+    await pod.tick(15)
+    assert notifications(mongo_db, 'OPC_LISTENNING_STOPPED__1')
+    assert notifications(mongo_db, 'OPC_CONNECTION_RETRY__1')
+    assert 'plant-a' not in pod.manager.opc_managers
+    await pod.tick()
+
+    assert 'plant-a' in pod.manager.opc_managers
+    assert pod.manager.pool._connections['1'][0] is not first_client
+    await server.set('Temperature', 28.0)
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=28.0))
+
+
+async def test_scenario_2_3_4_server_restart_resumes_the_reads(
+    opc_servers, pods, put_slot, mongo_db
+):
+    server, _, pod = await _read_slot(
+        opc_servers, pods, put_slot, 'slot_single_server.json', {'Temperature': 20.0}
+    )
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a'))
+
+    await server.stop()
+    await pod.tick(15)  # no data: the server is dropped
+    assert 'plant-a' not in pod.manager.opc_managers
+    await pod.tick()  # still down: reconnection fails and is notified
+    assert notifications(mongo_db, 'OPC_CONNECTION_ERROR_plant-a')
+
+    await server.start()
+    await pod.tick(2)
+    assert 'plant-a' in pod.manager.opc_managers
+    await server.set('Temperature', 29.0)
+    await wait_until(lambda: readings(mongo_db, 'raw_plant_a', value=29.0))

e2e/test_writes.py

@@ -0,0 +1,386 @@
+"""E2E scenarios — 3. Writes (NATS C2 → OPC UA → Postgres write_status C3). See ``e2e/scenarios.md``."""
+
+import json
+from datetime import datetime
+
+import nats
+import pytest
+import pytest_asyncio
+
+from e2e.conftest import WRITE_SUBJECT, nats_server_container
+from e2e.helpers import (
+    SOURCE_TIMESTAMP_BIT,
+    STATUS_CODE_BIT,
+    datavalue_mask,
+    insert_prediction,
+    load_scenario_input,
+    notifications,
+    prediction_count,
+    stays_false,
+    wait_until,
+    write_status,
+)
+from e2e.opc_server import free_port
+
+pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
+
+TIMESTAMP = '2026-10-13T12:00:00-03:00'
+MODEL_ID = 2
+MARKER_MODEL_ID = 99
+
+
+@pytest_asyncio.fixture
+async def target(opc_servers, mongo_db, postgres_engine):
+    """plant-a with a writable Setpoint and a read-only Locked, its connector in Mongo
+    and a pending prediction row (model 2)."""
+    server = await opc_servers(
+        'plant-a',
+        {
+            'Setpoint': 0.0,
+            'Counter': (0, 'int', True),
+            'Locked': (5.0, 'float', False),
+            'Marker': 0.0,
+        },
+    )
+    mongo_db['connectors'].insert_one(
+        load_scenario_input('connector_plant_a.json', {'plant-a': server})
+    )
+    insert_prediction(postgres_engine, MODEL_ID, TIMESTAMP, 'pending')
+    return server
+
+
+def command(**overrides) -> dict:
+    return load_scenario_input('write_command.json', **overrides)
+
+
+async def processed(publish_command, postgres_engine) -> None:
+    """Wait until every command published so far was handled: NATS delivers the
+    messages of a subscription in order, so once a marker command (another row, the
+    ``Marker`` variable) is recorded, the previous ones are done."""
+    insert_prediction(postgres_engine, MARKER_MODEL_ID, TIMESTAMP, 'pending')
+    await publish_command(command(model_id=str(MARKER_MODEL_ID), addr='ns=2;s=Marker'))
+    await wait_until(lambda: write_status(postgres_engine, MARKER_MODEL_ID, TIMESTAMP) == 'ok')
+
+
+async def wait_status(postgres_engine, expected: str, model_id: int = MODEL_ID) -> None:
+    await wait_until(lambda: write_status(postgres_engine, model_id, TIMESTAMP) == expected)
+
+
+# 3.1 DataValue parts: timestamp and quality (SIENTIAPDE-2202) ------------------------
+
+DATAVALUE_CASES = [
+    pytest.param({}, {}, True, True, id='3_1_1_defaults_on'),
+    pytest.param({'WRITE_TIMESTAMP': 'false'}, {}, False, True, id='3_1_2_timestamp_off'),
+    pytest.param({'WRITE_QUALITY': 'false'}, {}, True, False, id='3_1_3_quality_off'),
+    pytest.param(
+        {'WRITE_TIMESTAMP': 'false', 'WRITE_QUALITY': 'false'},
+        {},
+        False,
+        False,
+        id='3_1_4_both_off',
+    ),
+    pytest.param(
+        {'WRITE_TIMESTAMP': 'true'},
+        {'write_timestamp': False},
+        False,
+        True,
+        id='3_1_5_command_overrides_env',
+    ),
+    pytest.param(
+        {'WRITE_TIMESTAMP': 'false', 'WRITE_QUALITY': 'false'},
+        {'write_timestamp': True, 'write_quality': True},
+        True,
+        True,
+        id='3_1_6_command_turns_on_over_env',
+    ),
+]
+
+
+@pytest.mark.parametrize('env,flags,expect_timestamp,expect_quality', DATAVALUE_CASES)
+async def test_scenario_3_1_datavalue_parts(
+    target,
+    pods,
+    publish_command,
+    postgres_engine,
+    opc_wire,
+    env,
+    flags,
+    expect_timestamp,
+    expect_quality,
+):
+    await pods('pod-a', **env)
+
+    await publish_command(command(**flags))
+
+    await wait_status(postgres_engine, 'ok')
+    stored = await target.read('Setpoint')
+    assert stored.Value.Value == 1.25
+    if expect_timestamp:
+        assert stored.SourceTimestamp == datetime.fromisoformat(TIMESTAMP)
+    else:
+        assert stored.SourceTimestamp is None
+    [request] = opc_wire
+    mask = datavalue_mask(request, 'ns=2;s=Setpoint')
+    assert bool(mask & SOURCE_TIMESTAMP_BIT) is expect_timestamp
+    assert bool(mask & STATUS_CODE_BIT) is expect_quality
+
+
+# 3.2 write_status (C3) -----------------------------------------------------------------
+
+
+async def test_scenario_3_2_1_pending_row_becomes_ok(
+    target, pods, publish_command, postgres_engine
+):
+    await pods('pod-a')
+
+    await publish_command(command(value=3.5))
+
+    await wait_status(postgres_engine, 'ok')
+    assert (await target.read('Setpoint')).Value.Value == 3.5
+
+
+async def test_scenario_3_2_2_null_write_status_becomes_ok(
+    opc_servers, mongo_db, pods, publish_command, postgres_engine
+):
+    server = await opc_servers('plant-a', {'Setpoint': 0.0})
+    mongo_db['connectors'].insert_one(
+        load_scenario_input('connector_plant_a.json', {'plant-a': server})
+    )
+    insert_prediction(postgres_engine, MODEL_ID, TIMESTAMP, None)  # row from before the column
+    await pods('pod-a')
+
+    await publish_command(command())
+
+    await wait_status(postgres_engine, 'ok')
+
+
+@pytest.mark.parametrize('worse', ['error', 'session_bad'])
+async def test_scenario_3_2_3_worse_outcome_is_kept(
+    opc_servers, mongo_db, pods, publish_command, postgres_engine, worse
+):
+    server = await opc_servers('plant-a', {'Setpoint': 0.0, 'Marker': 0.0})
+    mongo_db['connectors'].insert_one(
+        load_scenario_input('connector_plant_a.json', {'plant-a': server})
+    )
+    insert_prediction(postgres_engine, MODEL_ID, TIMESTAMP, worse)
+    await pods('pod-a')
+
+    await publish_command(command(value=4.0))
+    await processed(publish_command, postgres_engine)
+
+    assert (await server.read('Setpoint')).Value.Value == 4.0  # the OPC write happened
+    assert write_status(postgres_engine, MODEL_ID, TIMESTAMP) == worse
+
+
+async def test_scenario_3_2_4_failure_replaces_ok(target, pods, publish_command, postgres_engine):
+    await pods('pod-a')
+    await publish_command(command())
+    await wait_status(postgres_engine, 'ok')
+
+    await publish_command(command(value='not-a-number'))
+
+    await wait_status(postgres_engine, 'error')
+
+
+async def test_scenario_3_2_5_missing_row_is_not_created(
+    target, pods, publish_command, postgres_engine
+):
+    await pods('pod-a')
+
+    await publish_command(command(model_id='3', value=6.0))
+    await processed(publish_command, postgres_engine)
+
+    assert (await target.read('Setpoint')).Value.Value == 6.0
+    assert write_status(postgres_engine, 3, TIMESTAMP) == '<missing>'
+    assert prediction_count(postgres_engine) == 2  # model 2 + marker
+
+
+# 3.3 Failed commands -------------------------------------------------------------------
+
+
+async def test_scenario_3_3_1_unknown_server_is_an_error(
+    target, pods, publish_command, postgres_engine
+):
+    await pods('pod-a')
+
+    await publish_command(command(server_id='404'))
+
+    await wait_status(postgres_engine, 'error')
+    assert target.writes == []
+
+
+@pytest.mark.parametrize(
+    'connector_change',
+    [
+        pytest.param({'status': 'Inactive'}, id='inactive'),
+        pytest.param({'type': 'modbus'}, id='not_opc_ua'),
+        pytest.param({'security': {'mode': 'None'}}, id='no_policy'),
+        pytest.param({'url': ''}, id='no_url'),
+    ],
+)
+async def test_scenario_3_3_2_unusable_connector_is_an_error(
+    target, mongo_db, pods, publish_command, postgres_engine, connector_change
+):
+    mongo_db['connectors'].update_one({'connector_id': '1'}, {'$set': connector_change})
+    await pods('pod-a')
+
+    await publish_command(command())
+
+    await wait_status(postgres_engine, 'error')
+    assert target.writes == []
+
+
+@pytest.mark.parametrize(
+    'overrides',
+    [
+        pytest.param(
+            {'value': 'abc', 'data_type': 'int', 'addr': 'ns=2;s=Counter'}, id='conversion'
+        ),
+        pytest.param({'data_type': 'decimal128'}, id='unknown_data_type'),
+        pytest.param({'addr': None}, id='missing_addr'),
+    ],
+)
+async def test_scenario_3_3_3_invalid_command_is_an_error_without_opc_write(
+    target, pods, publish_command, postgres_engine, overrides
+):
+    await pods('pod-a')
+    cmd = command(**overrides)
+    if overrides.get('addr', '') is None:
+        del cmd['addr']
+
+    await publish_command(cmd)
+
+    await wait_status(postgres_engine, 'error')
+    assert target.writes == []
+
+
+async def test_scenario_3_3_4_bad_timestamp_skips_the_write_and_keeps_the_row(
+    target, pods, publish_command, postgres_engine
+):
+    await pods('pod-a')
+
+    await publish_command(command(timestamp='13/10/2026 12:00'))
+    await processed(publish_command, postgres_engine)
+
+    assert [w.NodeId.Identifier for w in target.writes] == ['Marker']
+    assert write_status(postgres_engine, MODEL_ID, TIMESTAMP) == 'pending'
+
+
+@pytest.mark.parametrize(
+    'addr',
+    [
+        pytest.param('ns=2;s=Locked', id='not_writable'),
+        pytest.param('ns=2;s=Missing', id='unknown_node'),
+    ],
+)
+async def test_scenario_3_3_5_opc_rejection_is_an_error(
+    target, pods, publish_command, postgres_engine, addr
+):
+    await pods('pod-a')
+
+    await publish_command(command(addr=addr))
+
+    await wait_status(postgres_engine, 'error')
+    assert (await target.read('Locked')).Value.Value == 5.0
+
+
+async def test_scenario_3_3_6_malformed_json_is_dropped_and_the_consumer_keeps_going(
+    target, pods, publish_command, postgres_engine
+):
+    await pods('pod-a')
+
+    await publish_command(b'{"server_id": "1", "value": ')
+    await publish_command(command(value=7.0))
+
+    await wait_status(postgres_engine, 'ok')
+    assert (await target.read('Setpoint')).Value.Value == 7.0
+
+
+# 3.4 Connections -----------------------------------------------------------------------
+
+
+async def test_scenario_3_4_1_write_reuses_the_read_connection(
+    opc_servers, pods, put_slot, publish_command, postgres_engine, mongo_db
+):
+    server = await opc_servers('plant-a', {'Temperature': 20.0, 'Setpoint': 0.0})
+    put_slot(1, load_scenario_input('slot_single_server.json', {'plant-a': server}))
+    insert_prediction(postgres_engine, MODEL_ID, TIMESTAMP, 'pending')
+    pod = await pods('pod-a')
+    await pod.tick()
+    assert mongo_db['connectors'].count_documents({}) == 0  # only the slot knows plant-a
+
+    await publish_command(command())
+
+    await wait_status(postgres_engine, 'ok')
+    assert server.session_count() == 1
+
+
+async def test_scenario_3_4_2_write_only_server_gets_its_own_connection(
+    target, pods, publish_command, postgres_engine
+):
+    pod = await pods('pod-a')
+    assert target.session_count() == 0
+
+    await publish_command(command())
+
+    await wait_status(postgres_engine, 'ok')
+    assert target.session_count() == 1
+    assert pod.manager.pool.config_for('1')['url'] == target.url
+
+
+# 3.5 Throughput and scaling ------------------------------------------------------------
+
+
+async def test_scenario_3_5_1_min_pause_between_writes(
+    target, pods, publish_command, postgres_engine
+):
+    await pods('pod-a', WRITE_MIN_PAUSE_S='0.5')
+
+    for value in (1.0, 2.0, 3.0):
+        await publish_command(command(value=value))
+
+    await wait_until(lambda: len(target.writes) == 3)
+    gaps = [b - a for a, b in zip(target.write_times, target.write_times[1:], strict=False)]
+    assert all(gap >= 0.45 for gap in gaps), gaps
+
+
+async def test_scenario_3_5_2_queue_group_writes_each_command_once(
+    target, pods, publish_command, postgres_engine
+):
+    await pods('pod-a')
+    await pods('pod-b')
+
+    for value in range(6):
+        await publish_command(command(value=float(value)))
+
+    await wait_until(lambda: len(target.writes) == 6)
+    assert await stays_false(lambda: len(target.writes) > 6)
+
+
+# 3.6 NATS availability -----------------------------------------------------------------
+
+
+async def test_scenario_3_6_1_nats_down_at_startup_connects_later(
+    target, pods, postgres_engine, mongo_db
+):
+    port = free_port()
+    url = f'nats://127.0.0.1:{port}'
+    pod = await pods('pod-a', NATS_URL=url)  # nothing listens yet
+    await pod.tick()  # retried by idle_sweep, notified once per outage
+
+    assert len(notifications(mongo_db, 'GATEWAY_NATS_CONSUMER_START_FAILED')) == 1
+    nats_server = nats_server_container(host_port=port)
+    nats_server.start()
+    try:
+        await pod.tick()
+        assert pod.runtime._nats_started
+
+        client = await nats.connect(url)
+        cmd = load_scenario_input('write_command.json')
+        await client.publish(WRITE_SUBJECT, json.dumps(cmd).encode())
+        await client.flush()
+        await wait_status(postgres_engine, 'ok')
+        await client.drain()
+    finally:
+        await pod.stop()
+        nats_server.stop()