chore: bump version to 2.1.0

SIENTIAPDE-2202: timestamp e qualidade na escrita OPC por padrão (envs do gateway) + suíte e2e

.gitignore

@@ -188,3 +188,6 @@ docs/superpowers/
 docs/e2e-test-guide.md
 docs/deploy-notes.md
 docs/HANDOFF.md
+
+openspec/
+.claude/

CLAUDE.md

@@ -17,9 +17,9 @@ pip install -r requirements.txt -r requirements-dev.txt
 ./run_local.sh                          # loads .env, runs python -m ingestor.app
 ./run_coverage.sh                       # pytest --cov=ingestor --cov-report=html, opens report
 
-pytest                                  # all tests
+pytest                                  # unit tests (testpaths = tests)
 pytest tests/unit                       # unit only
-pytest tests/functional                 # functional only (spins up real-ish clients, see conftest.py)
+pytest e2e/ -o testpaths=e2e -m integration   # E2E (Docker: testcontainers + in-process OPC servers)
 pytest tests/unit/test_ingestor.py::TestClass::test_name   # single test
 pytest --cov=ingestor --cov-report=html # coverage report -> htmlcov/
 
@@ -72,6 +72,6 @@ Env vars documented in README ("Configuration" section) and `.env.example`. Nota
 ## Testing notes
 
 - `tests/unit/managers/` mirrors `ingestor/managers/` 1:1.
-- `tests/functional/` exercises more end-to-end paths against `conftest.py` fixtures.
+- `e2e/` is the end-to-end suite (Laborious/Orchestrator pattern): real MongoDB/Redis/Postgres/MinIO/NATS via testcontainers, in-process asyncua servers, pods built like `app.py` (`e2e/helpers.Pod`). Scenarios are numbered by topic in `e2e/scenarios.md`, 1:1 with `test_scenario_<n>_<n>_<n>_*` — keep both in sync. Marked `integration`, not collected by the default `pytest`.
 - Coverage config (`pyproject.toml` / `.coveragerc`) excludes `ingestor/app.py` (thin entrypoint, signal wiring) — don't chase coverage there.
 - `ruff` complexity cap is `max-complexity = 15` (mccabe) — factor out branches in manager methods before that.

README.md

@@ -67,9 +67,8 @@ Gateway for on-demand and command-driven interactions with OPC UA servers:
   (`opc_certificates`), named in the slot and downloaded from MinIO
   (`ingestor/connection/certs.py`) rather than shipped in the image.
 
-See `docs/deploy-notes.md` for the deployment-facing details (env vars,
+See [Gateway environment variables](#gateway-environment-variables) and the
-health probe, NetworkPolicy, cert cache) and the `# OPC Gateway` block in
+`# OPC Gateway` blocks in `.env.example` for the new environment variables.
-`.env.example` for the full list of new environment variables.
 
 ### Write commands (C2) and `write_status` (C3)
 
@@ -78,10 +77,16 @@ The Laborious publishes one command per tag (SIENTIAPDE-1975):
 ```json
 {"server_id": "1", "addr": "ns=2;s=Pred", "value": 1.25, "data_type": "float",
  "model_id": "2", "model_name": "sin-approx", "schema": "sientia_data",
- "table_name": "predictions", "timestamp": "2026-10-13T12:00:00-03:00",
+ "table_name": "predictions", "timestamp": "2026-10-13T12:00:00-03:00"}
- "write_timestamp": false, "write_quality": false}
 ```
 
+- **Written `DataValue`** (SIENTIAPDE-2202): the value, plus the prediction
+  `timestamp` as `SourceTimestamp` and a Good `StatusCode`. Timestamp and quality can be
+  turned off separately; the value is always written. Each flag is taken from the
+  command (`write_timestamp` / `write_quality`) when present, else from the env
+  (`WRITE_TIMESTAMP` / `WRITE_QUALITY`), else `true`. The envs apply to every server of
+  the gateway (per-server config is a future SIENTIAPDE-2276 card).
+
 - **Server connection**: the command carries only `server_id`. The connection is
   resolved by `ServerConfigResolver` (`ingestor/connection/servers.py`): the live
   pooled connection opened by the reads, else the `connectors` document (`type: opc_ua`,
@@ -282,11 +287,14 @@ pytest --cov=ingestor
 pytest --cov=ingestor --cov-report=html
 ```
 
-### Functional Tests
+### End-to-end Tests
+Every dependency runs for real (MongoDB, Redis, PostgreSQL, MinIO and NATS via testcontainers;
+OPC UA servers in process). Requires Docker; not part of the default `pytest` run.
 ```bash
-# Run functional tests
+pytest e2e/ -o testpaths=e2e -m integration
-pytest tests/functional/
 ```
+Numbered scenarios (slot leasing, reads, writes, gRPC, active reads, certificates, lifecycle):
+[`e2e/scenarios.md`](e2e/scenarios.md). Setup and layout: [`e2e/README.md`](e2e/README.md).
 
 ## 📊 Monitoring and Metrics
 
@@ -342,9 +350,28 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring:
 | `MONGODB_PASSWORD` | MongoDB password | `sientia` | No |
 | `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No |
 
-See the `# OPC Gateway` block in `.env.example` (`NATS_URL`, `GRPC_ENABLED`,
+#### Gateway environment variables
-`CERT_CACHE_DIR`, `WRITE_TIMESTAMP`, ...) and `docs/deploy-notes.md` for the
+
-gateway-specific environment variables.
+Read by `GatewayConfig` (`ingestor/config.py`). Booleans accept `true`/`1`/`yes`
+(case-insensitive); anything else is `false`.
+
+| Variable | Description | Default |
+|----------|-------------|---------|
+| `NATS_URL` | NATS server for write commands (C2) | `nats://localhost:4222` |
+| `NATS_WRITE_SUBJECT` | Subject of the write commands | `opc.write` |
+| `NATS_QUEUE` | Queue group shared by the gateway pods | `gateway-writers` |
+| `GRPC_ENABLED` | Starts the on-demand `ReadTag` gRPC server | `false` |
+| `GRPC_PORT` | gRPC server port | `50051` |
+| `CERT_CACHE_DIR` | Local cache of the client certificates downloaded from MinIO | `/tmp/opc-certs` |
+| `WRITE_TIMESTAMP` | Writes the prediction `timestamp` as `SourceTimestamp` when the command has no `write_timestamp` | `true` |
+| `WRITE_QUALITY` | Writes a Good `StatusCode` when the command has no `write_quality` | `true` |
+| `WRITE_MIN_PAUSE_S` | Minimum pause between two OPC writes (seconds) | `0.0` |
+| `IDLE_CONN_MAX_S` | Idle time before an unpinned pool connection is closed (seconds) | `300.0` |
+
+`WRITE_TIMESTAMP` / `WRITE_QUALITY` are independent and apply to every OPC server of
+the gateway; the value is always written. See [Write commands](#write-commands-c2-and-write_status-c3)
+for the precedence (command → env → `true`). MinIO (`MINIO_*`) and Postgres
+(`POSTGRES_*`) settings are listed in `.env.example`.
 
 ### OPC Server Configuration
 

docs/planning/SIENTIAPDE-1972-banco-de-tags-e-certificados.md

@@ -1,354 +0,0 @@
-# SIENTIAPDE-1972 — Estado atual, pendências e plano: banco de tags + servidores OPC ↔ certificados (MinIO)
-
-> Levantamento em 28/10/2026, a partir do Jira (1972 e subtasks, 1568/1569/159x), do código do
-> `sientia-dataops-opc-ingestor` (`release/SIENTIAPDE-1973` @ `d68e842`, 2.0.0-rc4), do
-> `sientia-dataops-orchestrator_temporal` (`release/SIENTIAPDE-1974`), do Streamlit
-> (`sientia-projects-templates` `rc/SIENTIAPDE-1588`), do `sientia-dataops-database-migrations`
-> e do design `aig-docs/07 - Projects & R&D/sientia/opc-gateway/design.md`.
-
----
-
-## 1. Resumo executivo
-
-- **O gateway funciona no dev** (leitura v2, escrita via NATS, `write_status`) e as subtasks 1973, 1974,
-  1975 e 1985 estão **Concluídas**. Mas **nada disso está na `main`/produção**: o PR #48
-  (`release/SIENTIAPDE-1973` → `main`) está aberto, e a `release/SIENTIAPDE-1974` do orchestrator
-  nem PR para a `main` tem (5 commits à frente, 30 atrás).
-- **Risco imediato**: a `main` do Laborious (1.6.0) já escreve OPC **só via NATS**. Se ela for para
-  um ambiente com o ingestor 1.4.x, toda escrita OPC fica em `pending`.
-- **O banco de tags não existe em nenhum ambiente.** `connectors`/`connector_tags` só existiram no dev
-  durante o teste A2 da 1974 e foram apagadas. O orchestrator e o gateway já sabem ler o banco de tags
-  (por `tag_id`), mas sem dado, tudo roda pelo caminho legado (`opc-servers` + `server_id/tag_address`).
-- **Certificado hoje é um só para a plataforma** (`platform_certificates`, `find_one({})`). Não existe
-  correlação servidor OPC ↔ certificado no Mongo. O `server_cert_path` (cert do servidor) é ignorado
-  com aviso. Bucket em uso: `streamlit-connectors` (o design pede `sientia-dataops`).
-- **O CRUD (Streamlit 1568/1569) está On Hold** desde 04/2026 para migrar ao Angular; não há card
-  Angular aberto para isso (só a idea `SIENTIAPDI-84`, em parking lot).
-
-O que falta para fechar o 1972 é pouco (merges + 2202 + 2270). O que vocês tinham pensado — **banco de
-tags + servidores OPC estruturados com certificados no MinIO** — é praticamente uma **fase 2 inteira**
-que não tem card hoje. Este documento propõe o modelo e quebra em cards.
-
----
-
-## 2. Estado das subtasks do 1972
-
-| Card | Status | O que entregou / o que falta |
-|---|---|---|
-| 1973 Gateway | ✅ Concluído (02/10) | Escrita NATS, gRPC ReadTag, pool, CertProvider. PRs empilhados #31–#33 fechados, código na `release/SIENTIAPDE-1973` |
-| 1974 Orchestrator + leitura v2 | ✅ Concluído (09/10) | Gateway #41 (leitura v2, cert do MinIO), #42 (KeyError) mergeados na release. Orchestrator #45 (slot v2, `names[]`, banco de tags opcional por `tag_id`) mergeado na **`release/SIENTIAPDE-1974`**, não na `main`. G6/G7 (A6/A7) não serão feitos: não existe leitura ativa programada; leitura ativa é só on-demand via gRPC, sem persistir (G-AR) |
-| 1975 Laborious NATS + `write_status` | ✅ Concluído (23/10) | migrations #37 (V16), gateway #43/#47, laborious #5 (main 1.6.0). Validado até o ProSYS (M0–M4) |
-| 1985 NATS no cluster | ✅ Concluído (30/09) | NATS 2.14.6 via helm. `#108` (release/1.0.0) e `#126` (development) — conferir se ambos já foram mergeados |
-| **2202** timestamp/qualidade na escrita | 🔄 Em andamento (Eduardo) | Decidir onde fica a config (sugestão: por servidor). **Encaixa direto no modelo de `connectors` deste plano** |
-| **2270** mlops-bff `write_status` no Model Sustain | 🔄 Em andamento (Carla) | Depende de decisão do PO (falha de escrita deixa o modelo CRITICAL ou não) e da V16 aplicada antes do deploy do BFF |
-
----
-
-## 3. Pendências (consolidado dos comentários dos cards)
-
-### 3.1 Promoção para produção (bloqueia tudo o mais)
-
-| # | Pendência | Onde | Situação |
-|---|---|---|---|
-| P1 | PR #48 `release/SIENTIAPDE-1973` → `main` (gateway 2.x) | opc-ingestor | Aberto. Combinar número de versão (bump pode não gerar 2.0.0) |
-| P2 | PR `release/SIENTIAPDE-1974` → `main` | orchestrator | **Não existe.** Release está 30 commits atrás da `main` (0.8.2), com conflito em 4 arquivos (`orchestrator_functions.py`, teste dele, `README.md`, `pyproject.toml`). **Desde 19/10 o dev roda o orchestrator 0.8.1 (`main`, sem slot v2)**: o deploy do PR #49 fez o bump de `GITHUB_BRANCH` 0.7.0-rc1 → 0.8.1 no chart (`development`, commit `07ffc99`) |
-| P3 | Card do helm (H1–H4 do 1975): migrations com V16, Grafana sem 12/13 e com `write_status`, `docs/nats-broker.md` §11. NATS + envs já estão em todo profile com `opc-ingestor` ligado (desligado só em galvani, quartz-homologacao, gcp-hom, que não têm `opc-ingestor` nem `laborious-worker`) | sientia-helm-chart | **Card não criado** |
-| P4 | Ordem de deploy por ambiente: migrations V16 → gateway 2.x → laborious 1.6.0 | todos | Documentar no runbook de release |
-| P5 | Correção da `main` do gateway (SQLAlchemy 2.1 + senha do Mongo no log), branch `fix/sqlalchemy-21-mongo-uri-log` | opc-ingestor | Fica obsoleta se o #48 entrar; senão, abrir PR de hotfix |
-| P6 | Servidores OPC de produção aceitarem `Basic256Sha256` e confiarem no certificado da plataforma (SAN URI = `server_uri`) | infra/cliente | Pré-requisito do deploy da leitura v2 em cada cliente |
-| P7 | Certificado da plataforma cadastrado (`platform_certificates` + MinIO) em cada ambiente | por ambiente | Hoje só no dev, cadastrado à mão |
-
-### 3.2 Gateway (código)
-
-| # | Pendência | Referência |
-|---|---|---|
-| G-AR | **Leitura ativa = unitária e atômica: lê e devolve o valor, nada mais** (decisão 28/10). Sem leitura programada e sem persistência no gRPC. Hoje já se comporta assim (`slot_lookup=None` → `persisted=false`), falta remover o código morto: `scheduled_tick`/`_tick_tag`/`_should_read`/`_ticks` e o parâmetro `names` + `sink.publish_reading` de `read_now`/`_build_result` (`engines/active_read.py`); `slot_lookup` de `ReadTagServicer`/`GrpcServer` (`adapters/grpc_server.py`); `ponytail` de `gateway_runtime.py:10,93`; campo `persisted` do `ReadTagResponse` (marcar `reserved 5` no proto); `active_read` do slot v2 no orchestrator e no `parse_slot`; testes correspondentes | decisão 28/10 |
-| G-DOC | Atualizar design §6 C4 e §14 ("Reads, persists to the pipeline, and returns the value") e os requisitos G3/G4/A6/A7 do 1974 para refletir a decisão | aig-docs `opc-gateway/design.md` |
-| G-PB | Remover plano B `READ_ENGINE=legacy` (`OpcManager`, `scripts/generate-and-push-opc-cert.sh`) | critério do 1974; pode sair no #48 ou logo depois |
-| G-RC | Reconectar ao receber `status_change_notification` (KEPServer derruba sessão a cada 2 h → 2,5–3 min sem dado) | achado do 1974 |
-| G-SC | Suporte a `server_cert_path` (validar cert do servidor) — hoje só aviso | `ingestor/managers/ingestor_manager.py:708` — resolvido pela Fase 2 abaixo |
-
-### 3.3 Fora do gateway
-
-| # | Pendência | Situação |
-|---|---|---|
-| C1 | Consumidores dos códigos 12/13/14: `sientia-dataops-report`, `aignosi-scripts` (`sientia_status_report.sh`) | **Sem card** (só o BFF tem, 2270) |
-| C2 | Laborious: remover `write_opc_data` antigo quando não houver execução anterior ao patch | sem card |
-| C3 | Limpeza no dev: tabela `sientia_data.e2e_predictions`, pipeline `laborious-test-1975-write`, branches `test-SIENTIAPDE-197x-*` no Gitea | pendente de ok |
-| C4 | Melhoria de baixa prioridade do scouter (timeout Redis, retries, catch-up) | sugerido, sem card |
-| C5 | Usuário `aignosi` da VM ProSYS + senha em texto no README do `sientia-iac` | sem card |
-
-### 3.4 Banco de tags e certificados (bloqueios B1–B6 do 1974 — **nenhum resolvido**)
-
-| # | Bloqueio | Situação hoje |
-|---|---|---|
-| B1 | `connectors`/`connector_tags` não existem em nenhum ambiente | **escopo do orchestrator** (decisão 28/10) |
-| B2 | Pipelines não gravam `tag_id` em `read_tags`/`write_tags` | **escopo do orchestrator** |
-| B3 | ~~Onde fica `active_read`~~ | não se aplica: leitura ativa é só on-demand, sem persistir |
-| B4 | Certificado por ambiente + bucket (`sientia-dataops` do design × `streamlit-connectors` do chart/D7) e correlação servidor ↔ certificado | **escopo do orchestrator**; bucket indefinido |
-| B5 | Tela de certificados: 1599 entregou no **Streamlit** (rc, não em produção) | não existe no Angular |
-| B6 | ~~gRPC persist / timestamp nulo~~ | não se aplica: gRPC não persiste; `source_timestamp` vazio só volta vazio na resposta |
-| — | Renomear tag em `connector_tags` não reimplanta o scouter (só `pipelines.updated_at`) | limitação conhecida |
-| — | Restos do 1569: 1598 (`tag_id` sequencial sem corrida), 1600 (`metadata`), 1602 (bucket por servidor), 1603 (Server URI), 1604, 1605 (CSV) | On Hold, "passíveis de cancelamento" |
-
----
-
-## 4. Diagnóstico do modelo atual (Mongo + MinIO)
-
-Hoje coexistem **dois registros de servidor OPC** e **um certificado global**:
-
-```
-opc-servers            { id, name, url, uri, cert_path, private_key_path, server_cert_path, ... }   ← legado, usado em todos os ambientes
-connectors (opc_ua)    { connector_id, type, name, url, server_uri, use_security, subscription_period_ms, status, description }
-connector_tags         { tag_id, connector_id, address, tag_name, category, data_type, unit, status, description, <custom flat keys> }
-platform_certificates  { client_cert: {minio_object_name, sha256_hash, expiration_date, ...}, private_key: {...} }   ← 1 documento só
-MinIO                  <bucket>/platform/certificates/client_cert_<arquivo>, private_key_<arquivo>
-```
-
-Problemas concretos:
-
-1. **Sem correlação servidor ↔ certificado.** `CertProvider._download_all` faz `find_one('platform_certificates', {})`
-   (`ingestor/connection/certs.py:47`). Um cliente com dois servidores que exigem certificados diferentes
-   (ou um servidor que exija validação do cert do servidor) não é atendível.
-2. **`server_uri` é dado do certificado, não do servidor.** O `application_uri` do cliente precisa ser
-   igual à SAN URI do certificado da plataforma (comentário 06/10 do 1974). Hoje ele é digitado por
-   conector (`connectors.server_uri` / `opc-servers.uri`) — e o 1603 queria "chumbar" no backend.
-   Erro de digitação = conexão segura recusada sem pista clara.
-3. **`server_cert_path` ignorado.** O cert do servidor não tem lugar no MinIO/Mongo.
-4. **Dois espaços de id.** `ServerConfigResolver` procura `connectors.connector_id` e depois `opc-servers.id`
-   com o mesmo `server_id` (`ingestor/connection/servers.py:105-115`); o orchestrator trata "mesmos campos"
-   como mesmo servidor. Se os ids divergirem, um `server_id` aponta para servidores diferentes conforme a fonte.
-5. **Ids por `$max + 1`** (Streamlit `tag_monitor.py:141`) — corrida em criação simultânea/import em lote.
-   O repo de migrations já tem o padrão certo (`model_id_counters` com `$inc` atômico).
-6. **Sem schema/índices.** Nada impede `(connector_id, tag_name)` duplicado nem conector sem `url`.
-7. **Sem visibilidade de expiração** do certificado (o Streamlit calcula `expiration_date`, ninguém alerta).
-
----
-
-## 5. Modelo proposto
-
-Princípio: **`connectors` vira o registro canônico de servidores OPC**, `opc-servers` vira legado
-(somente leitura, depois removido); **certificados viram entidade própria** referenciada pelo conector.
-
-### 5.1 `opc_certificates` (nova; substitui `platform_certificates`)
-
-```js
-{
-  certificate_id: 1,                         // int, counter atômico
-  kind: "client",                            // client | server | ca
-  name: "sientia-platform-2026",
-  is_default: true,                          // no máximo 1 client default (índice parcial único)
-  status: "active",                          // active | retired
-  application_uri: "urn:sientia:opc-ingestor",  // SAN URI extraída do cert no upload (kind=client)
-  subject: "CN=sientia-opc-ingestor,...",
-  thumbprint_sha1: "…",                      // o que aparece no console do KEPServer/servidor
-  not_before: ISODate(), not_after: ISODate(),
-  cert:        { bucket, object_name, sha256, size, content_type, original_filename },
-  private_key: { bucket, object_name, sha256, size },   // só kind=client
-  created_at, created_by, updated_at
-}
-```
-
-- Índices: `certificate_id` único; `{is_default:1}` único parcial (`kind:"client", is_default:true`);
-  `not_after` (para alerta).
-- `application_uri`, `thumbprint`, `not_after` **extraídos do arquivo no upload**, nunca digitados.
-
-### 5.2 `connectors` (opc_ua) — campos novos
-
-```js
-{
-  connector_id: 1,                           // = opc-servers.id no backfill (mantém server_id estável)
-  type: "opc_ua", name: "kepserver", url: "opc.tcp://…", status: "Active",
-  subscription_period_ms: 1000,
-  security: {
-    mode: "SignAndEncrypt",                  // None | Sign | SignAndEncrypt
-    policy: "Basic256Sha256",
-    client_certificate_id: null,             // null → certificado default da plataforma
-    server_certificate_id: 7                 // opcional: valida o cert do servidor (resolve G-SC)
-  },
-  write_options: { write_timestamp: true, write_quality: true },   // SIENTIAPDE-2202 (Kepware: false/false)
-  legacy_opc_server_id: "1",                 // rastreio do backfill
-  created_at, updated_at
-}
-```
-
-- `use_security` vira derivado (`security.mode != "None"`); manter o campo booleano durante a transição
-  para não quebrar orchestrator/gateway atuais.
-- `server_uri` **sai do conector**: o `application_uri` vem do certificado cliente resolvido.
-  (Manter leitura do campo antigo como fallback na transição.)
-- Índices: `connector_id` único; `{type, name}` único.
-
-### 5.3 `connector_tags` — ajustes
-
-```js
-{
-  tag_id: 9001, connector_id: 1, address: "ns=2;s=…", tag_name: "TargetSin",
-  category: "Read",                          // Read | Write | Prediction Confidence
-  data_type: "Double", unit: "", status: "Active", description: "",
-  metadata: {},                              // objeto aninhado (decisão do design §14; fecha a dúvida do 1600)
-  created_at, updated_at
-}
-```
-
-- Índices: `tag_id` único; `{connector_id, tag_name}` único; `{connector_id, address}` (não único —
-  multi-nome é permitido); `updated_at` (orchestrator detectar mudança).
-
-### 5.4 `connector_id_counters` (padrão `model_id_counters`)
-
-Documentos `{_id: "connectors"|"connector_tags"|"opc_certificates", seq}`, semeados com o `max` atual,
-incremento por `find_one_and_update($inc)`. Resolve o 1598 sem `_get_next_tag_id_batch` com corrida
-(batch = `$inc: N` e usa a faixa).
-
-### 5.5 MinIO
-
-```
-<CERT_BUCKET>/opc-ua/certificates/<certificate_id>/cert.der
-<CERT_BUCKET>/opc-ua/certificates/<certificate_id>/private_key.pem
-```
-
-- Chave por `certificate_id` (não por nome de conector: renomear conector não move objeto; um cert serve vários conectores).
-- Bucket: **decidir** (ver D1). Policies: escrita só para o usuário do front/BFF; leitura para o gateway.
-- O `sha256` no Mongo é conferido pelo gateway no download (detecta objeto trocado/corrompido).
-
-### 5.6 Como o gateway resolve uma conexão (alvo)
-
-```
-server_id ──► connectors (connector_id) ──► security.client_certificate_id ?? default
-                     │                              │
-                     │                              └─► opc_certificates ─► MinIO (cert+key) ─► cache por certificate_id
-                     ├─► security.server_certificate_id ─► opc_certificates (kind=server) ─► MinIO
-                     └─► application_uri := opc_certificates.application_uri
-fallback (transição): opc-servers.id + certificado default
-```
-
----
-
-## 6. Plano em fases
-
-### Fase 0 — Orchestrator primeiro, depois levar a 2.x para produção (curto prazo)
-
-Ordem: **orchestrator antes do gateway**. Sem o slot v2 na `main` do orchestrator, a 2.x em produção só
-roda o caminho legado (slot v1 convertido): sem multi-nome, sem banco de tags. O orchestrator também é dono
-de popular o banco de tags, gravar `tag_id` nas pipelines e correlacionar servidores ↔ certificados (Fase 1).
-
-1. Atualizar `release/SIENTIAPDE-1974` com a `main` (0.8.2), resolver os 4 conflitos, rodar testes + e2e,
-   abrir PR → `main`, deploy no dev e **refazer A1, A2 e A8** (o dev perdeu o slot v2 em 19/10; os testes
-   da 1975 de 19–23/10 rodaram com slot v1).
-2. Só então revisar/mergear **#48** (gateway → `main`); definir versão `2.0.0`. Decidir se o plano B sai aqui (G-PB).
-3. **Criar card do helm (P3)** com H1–H4.
-4. Runbook de deploy por ambiente (P4, P6, P7): checklist V16 → gateway → laborious; cert cadastrado; servidor
-   confia no cert; Kepware com `write_timestamp/quality=false`.
-5. Concluir **2202** já usando `connectors.write_options` (fallback env `WRITE_TIMESTAMP/WRITE_QUALITY`) — é
-   o primeiro campo do modelo novo e não depende do resto.
-6. **2270**: obter decisão do PO; criar cards de C1 (report, aignosi-scripts).
-7. Limpeza do dev (C3).
-
-**Saída:** 1972 pode ser fechado. Tudo abaixo vira **nova história/épico** (sugestão: "Banco de tags e
-certificados OPC estruturados", ligada à `SIENTIAPDI-84`).
-
-### Fase 1 — Modelo de dados + backfill — escopo do orchestrator (collections/índices via migrate-mongo no `sientia-dataops-database-migrations`)
-
-1. Migration `…-opc-tag-database.js`: cria `connectors`, `connector_tags`, `opc_certificates`,
-   `connector_id_counters` com `$jsonSchema` (validationLevel `moderate` na transição) e os índices de §5.
-2. Migration `…-backfill-tag-database.js` (idempotente, só insere o que não existe):
-   - `opc-servers` → `connectors` com **`connector_id = int(opc-servers.id)`**, `security.mode` derivado de
-     `cert_path`, `legacy_opc_server_id`.
-   - `platform_certificates` → `opc_certificates` (id 1, `is_default: true`), copiando os metadados MinIO
-     existentes (sem mover objeto ainda).
-   - `pipelines.read_tags[]`/`write_tags[]` (`server_id`, `tag_address`, `tag_name`) → `connector_tags`
-     (dedupe por `(connector_id, tag_name)`), e **grava `tag_id` de volta na pipeline** (resolve B1 + B2 de uma vez).
-   - semeia os counters com os máximos.
-3. Dry-run com dump do dev e de um cliente (relatório de conflitos: mesmo `tag_name` com endereços diferentes,
-   `server_id` sem servidor).
-
-### Fase 2 — Gateway (`sientia-dataops-opc-ingestor`)
-
-1. `CertProvider` → `CertificateStore`: `fetch(certificate_id | None)`, cache e lock **por `certificate_id`**,
-   verifica `sha256`; `None` = default (`is_default`), fallback `platform_certificates` durante a transição.
-2. `ServerConfigResolver` + slot: conexão passa a carregar `client_certificate_id`, `server_certificate_id`,
-   `application_uri`, `write_options`. Pool usa esses campos em `apply_security` (passa `server_certificate`).
-3. Validar no carregamento: SAN do cert == `application_uri`; cert não expirado → erro claro + notificação
-   (`OPC_CERTIFICATE_*`), em vez de falha opaca do asyncua.
-4. Métrica `opc_certificate_expiry_seconds{certificate_id}` + alerta (ex.: < 30 dias).
-5. `WriteEngine`: `write_timestamp/quality` = comando → `connectors.write_options` → env (fecha 2202 de vez).
-6. G-AR: limpar leitura ativa para leitura unitária sem persistência (pode ir junto com o #48 ou logo depois) + G-DOC.
-
-### Fase 3 — Orchestrator (`sientia-dataops-orchestrator_temporal`)
-
-1. Fonte de servidor: `connectors` primeiro, `opc-servers` fallback (hoje é o contrário: só usa `connectors`
-   quando a pipeline tem `tag_id`).
-2. Slot v2 por servidor: `connector_id`, `client_certificate_id`, `server_certificate_id`, `application_uri`
-   (no lugar de `use_security`/`server_uri` soltos — manter os antigos até o gateway novo estar em todo lugar).
-3. Remover `active_read` do slot v2 (sem leitura ativa programada).
-4. Reagir a mudança em `connector_tags.updated_at`/`connectors.updated_at` (reimplantar scouter/slot) — fecha a
-   limitação de renomear tag.
-
-### Fase 4 — CRUD no Angular (front `sientia` + BFF)
-
-1. Novo épico/história substituindo 1568/1569: telas de **Conectores**, **Tags** (com import CSV — escopo do 1605)
-   e **Certificados** (upload extrai SAN/thumbprint/validade; marca default; lista conectores que usam).
-2. BFF é quem escreve no Mongo/MinIO (front não fala com MinIO direto); usa os counters.
-3. Tela de pipeline passa a escolher tag por `tag_id` (pipelines novas já nascem com `tag_id`).
-4. Cancelar no Jira os restos Streamlit não iniciados (1600, 1604, 1605, 1603) com link para o novo card.
-
-### Fase 5 — Desligar o legado
-
-1. Orchestrator e gateway sem fallback para `opc-servers`/`platform_certificates`; `server_id/tag_address`
-   nas pipelines viram só leitura histórica.
-2. Remover `opc-servers`, `platform_certificates` (migration com backup) e o caminho Streamlit.
-3. `$jsonSchema` em `validationLevel: strict`.
-
----
-
-## 7. Decisões necessárias
-
-| # | Decisão | Recomendação | Dono |
-|---|---|---|---|
-| D1 | Bucket dos certificados: `sientia-dataops` (design §7/§14, Streamlit rc) × `streamlit-connectors` (chart, D7 da 1974) | `sientia-dataops` (bucket por serviço, estratégia MinIO); migrar objetos na Fase 1 e trocar `CERT_BUCKET` no chart | Vitor / platform |
-| D2 | Cert por servidor ou único da plataforma | Modelo suporta os dois; **default único** + override por conector só quando necessário | Vitor |
-| D3 | `connector_id` = `opc-servers.id` no backfill | Sim — mantém `server_id` de slots, `write_tags` e métricas estáveis | Eduardo |
-| D6 | `metadata` aninhado × chaves soltas (1600) | Aninhado (`metadata: {}`), como no design §14 | PO |
-| D7 | `write_options` por servidor (2202) | Por servidor, padrão `true`; Kepware configurado `false` antes do deploy | Eduardo |
-| D8 | Quem escreve banco de tags em produção até o Angular existir | Backfill por migration + edição via migration/script revisado (sem Streamlit em produção) | Bruno / PO |
-| D9 | Plano B `READ_ENGINE=legacy` sai no #48? | Sim, se A1–A8 seguem ok no dev com a 2.0.0-rc4 | Eduardo |
-
----
-
-## 8. Cards a criar / ajustar
-
-| Card | Tipo | Repo | Fase |
-|---|---|---|---|
-| Helm: V16, NATS+gateway 2.x nos profiles, `OPC_WRITE_MODE`, Grafana `write_status`, docs (H1–H4) | Subtask do 1972 | sientia-helm-chart | 0 |
-| PR `release/SIENTIAPDE-1974` → `main` (orchestrator) | Subtask do 1972 | orchestrator | 0 |
-| Consumidores 12/13/14: report + aignosi-scripts | Subtask | report, aignosi-scripts | 0 |
-| **[Nova história] Banco de tags e certificados OPC estruturados** | História (épico 1274 ou novo) | — | 1–5 |
-| ↳ Orchestrator: popular banco de tags, gravar `tag_id` nas pipelines, correlação servidor ↔ certificado (collections/índices via migrate-mongo) | Subtask | orchestrator, database-migrations | 1 |
-| ↳ Gateway: `CertificateStore` por `certificate_id`, validação SAN/expiração, métrica de expiração | Subtask | opc-ingestor | 2 |
-| ↳ Gateway: leitura ativa unitária sem persistência (G-AR) + design (G-DOC) | Subtask | opc-ingestor, orchestrator, aig-docs | 2 |
-| ↳ Gateway: reconexão por `status_change_notification` | Subtask | opc-ingestor | 2 |
-| ↳ Orchestrator: `connectors` como fonte, slot com certificados, sem `active_read`, reação a `updated_at` | Subtask | orchestrator | 3 |
-| ↳ Angular + BFF: CRUD conectores/tags/certificados (substitui 1568/1569) | História própria | sientia, mlops-bff | 4 |
-| ↳ Remoção do legado (`opc-servers`, `platform_certificates`, `READ_ENGINE=legacy`) | Subtask | vários | 5 |
-
----
-
-## 9. Riscos
-
-- **Laborious 1.6.0 antes do gateway 2.x** em qualquer ambiente → escritas presas em `pending` (P4).
-- **Backfill com ids divergentes** entre `opc-servers` e um `connectors` já existente em algum cliente (o
-  Streamlit rc pode ter rodado em algum lugar) → checar antes, migration aborta em conflito.
-- **Troca de bucket/objeto do certificado** derruba conexões seguras se o gateway não invalidar o cache —
-  já existe `invalidate()` em falha; manter e testar por `certificate_id`.
-- **Servidores de cliente com Basic256 apenas** (asyncua marca deprecated) → leitura v2 não conecta; levantar por cliente antes.
-
----
-
-## 10. Referências
-
-- Jira: SIENTIAPDE-1972, 1973, 1974, 1975, 1985, 2202, 2270, 1568, 1569, 1598, 1599, 1603, 1605; idea SIENTIAPDI-84.
-- Código: `ingestor/connection/certs.py`, `ingestor/connection/servers.py`, `ingestor/gateway_runtime.py`,
-  `ingestor/managers/ingestor_manager.py:708`, `ingestor/io/slots.py`.
-- Orchestrator: `release/SIENTIAPDE-1974` — `orchestrator/activities/formatters.py`, `orchestrator/utils/orchestrator_functions.py`.
-- Streamlit: `sientia-projects-templates` `rc/SIENTIAPDE-1588` — `app/pages/connectors.py` (`_save_platform_certificate`), `app/pages/tag_monitor.py`.
-- Migrations: `mongodb/migrations/20260908170557-model_id_counters_collection.js` (padrão de counter).
-- Design: `aig-docs/07 - Projects & R&D/sientia/opc-gateway/design.md` §3, §7, §14, §15.

docs/planning/SIENTIAPDE-1974-certificado-por-conector.md

@@ -1,241 +0,0 @@
-# SIENTIAPDE-1974 — Certificado OPC por conector (`opc_certificates`)
-
-> Plano de implementação. Atualizado em 2026-10-29.
-> Repositórios: `sientia-dataops-database-migrations`, `sientia-helm-chart`,
-> `sientia-dataops-orchestrator_temporal`, `sientia-dataops-opc-ingestor`.
-
-## Objetivo
-
-Ligar cada servidor OPC (`connectors`) ao certificado que ele usa (`opc_certificates`) por **um único caminho**,
-sem transição, sem fallback e sem legado:
-
-```
-connectors.security.client_certificate_id ──► opc_certificates.certificate_id
-                                                     │
-orchestrator: faz o join e grava no slot v2, por servidor, security + certificado (paths no MinIO + application_uri)
-                                                     │
-ingestor: lê o slot, baixa cert + chave do MinIO e conecta. Sem certificado no slot → conecta sem segurança
-```
-
-O ingestor não usa certificado "default", `platform_certificates`, `opc-servers`, nem o caminho legado
-(`READ_ENGINE=legacy`, `OpcManager`, slot v1).
-
-**Exceção consciente (D1):** a escrita mantém um fallback no Mongo. Quando o pod que recebe o comando não tem a
-conexão no pool, ele resolve o servidor em `connectors` → `opc_certificates`, com a mesma regra do orchestrator. São
-duas fontes da mesma regra, aceitas por enquanto. A otimização fica num card próprio (ver "Fora do escopo").
-
-**Fora do escopo:** migração e correção de dados. As migrations deste plano tratam só de estrutura.
-
----
-
-## Estado atual (verificado)
-
-| Item | Estado | Evidência |
-|---|---|---|
-| `platform_certificates` | Criada pelo front Streamlit, sem migration | `sientia-projects-templates` `origin/rc/SIENTIAPDE-1588` `app/pages/connectors.py:27` |
-| `opc_certificates` | Criada pela migration `20261028170000-opc_tag_database_collections.js`, com schema parcial. Tem dados no dev | Migration, consulta no dev |
-| Vínculo connector → certificado | O campo existe no schema (`connectors.security.client_certificate_id`). No dev está `null` | Migration, consulta no dev |
-| `security.mode` / `security.policy` | Estão no schema e nos dados do dev, mas o ingestor ignora os dois: usa `use_security` e fixa `Basic256Sha256` no código | `ingestor/connection/security.py`, `pool.py:130` |
-| `application_uri` | Vem de `connectors.server_uri`, mas é dado do certificado (tem que bater com a SAN URI do `.pem`) | `pool.py:126-132`, `security.py:9` |
-| Leitura do certificado | `CertProvider` faz `find_one('platform_certificates', {})` e baixa do `CERT_BUCKET` | `ingestor/connection/certs.py:10,47` |
-| Bucket `opc-certificates` | Citado em `opc_certificates.cert.bucket`, mas não provisionado. O chart aponta `CERT_BUCKET=streamlit-connectors` | `sientia-helm-chart/values/minio.yaml:516-523`, `opc-ingestor/values.yaml:116` |
-| Mudança de config de um servidor no slot | O ingestor compara a config inteira do servidor (sem `tags`) e reconecta se algo mudou | `ingestor/managers/ingestor_manager.py:293-307` |
-| Escrita (NATS) | Core NATS, sem JetStream, entrega *at-most-once*. Queue group `gateway-writers`: cada comando vai para um só pod, que não tem como devolvê-lo. O Laborious publica e esquece; o resultado vai para `write_status` no Postgres | `sientia-helm-chart/docs/nats-broker.md:29-39`, `ingestor/adapters/nats_consumer.py`, `laborious/activities/nats_write.py` |
-
----
-
-## Decisões
-
-| # | Decisão | Status |
-|---|---|---|
-| D0 | `opc_certificates` é a oficial; o ingestor deixa de ler a `platform_certificates` (a collection fica) | **Decidido** |
-| D0.1 | Migration da `opc_certificates` completa, documentada e auditável | **Decidido** (Etapa 1) |
-| D0.2 | Bucket `opc-certificates` provisionado no `sientia-helm-chart`, branch `feature/SIENTIAPDE-1974-connectors-pi-web-api` | **Decidido** (Etapa 2) |
-| D0.3 | Sem transição. Caminho único na leitura; a escrita tem a exceção de D1 | **Decidido** |
-| D1 | Escrita: pool primeiro; sem conexão no pool, resolve no Mongo (`connectors` → `opc_certificates`) com a mesma regra do orchestrator. Roteamento ao pod dono vira card novo | **Decidido** (ver abaixo) |
-| D2 | Zero legado: `READ_ENGINE=legacy`, `OpcManager`, slot v1, `opc-servers` | **Decidido** |
-| D3 | Dados fora do plano; migrations só de estrutura; sem runbook | **Decidido** |
-| D4 | `security.mode` + `security.policy` são a fonte. `use_security` e `server_uri` saem | **Decidido** |
-| D5 | Sem certificado registrado, o servidor é tratado como um que não exige certificado: conecta sem segurança (e falha se o servidor exigir) | **Decidido** |
-| D6 | Branch do ingestor: `release/SIENTIAPDE-1973` | **Decidido** |
-| D7 | `created_at`, `updated_at`, `created_by`, `updated_by` obrigatórios. Rastro do backfill não é validado. Sem `additionalProperties: false` | **Decidido** |
-
-### D1 — Escrita: pool primeiro, Mongo como fallback (decidido)
-
-**Como funciona hoje:** o Laborious publica o comando (C2) em `opc.write` só com o `server_id`. No ingestor, o
-`ServerConfigResolver` (`ingestor/connection/servers.py`) usa a conexão do pool, se existir. Senão, consulta o Mongo
-(`connectors` e depois `opc-servers`), com cache de 60 s (10 s quando não encontra), e o pool abre a conexão.
-
-**O que muda:**
-- Sai `opc-servers` (D2) e sai o override por dados de conexão no comando (o Laborious nunca envia).
-- O lookup no Mongo passa a fazer o join `connectors` → `opc_certificates` e monta a config **no mesmo formato do
-  servidor no slot** (`server_id`, `url`, `security`, `certificate`). O pool e o `CertProvider` só conhecem esse formato.
-- A regra do certificado (D5: certificado `active` e do tipo `client`, senão `certificate: None`) fica isolada numa
-  função do ingestor, com testes que espelham os do orchestrator.
-
-**Por que não só o pool:** o Core NATS entrega cada comando a um único pod, sem devolver nem reentregar. Rotear o
-comando ao pod dono da conexão (subject por servidor, republicação ou JetStream) não cobre um servidor que nenhum pod
-tem no pool (servidor só de escrita, ou pod recém-reiniciado) e ainda exige uma regra de expiração do `pending`.
-O fallback no Mongo cobre todos esses casos hoje.
-
-**Custo aceito:**
-- Duas fontes da mesma regra (orchestrator e ingestor).
-- Sessões duplicadas: o pod que recebe uma escrita sem ter a conexão abre a sua própria sessão (já acontece hoje).
-
-### D5 — Sem certificado registrado (decidido)
-
-- O orchestrator grava `certificate: None` no slot quando `client_certificate_id` é `null`, aponta para um id que não
-  existe, ou aponta para um certificado que não é `active` ou não é `client`. Nos dois últimos casos (id preenchido,
-  mas sem certificado válido), o orchestrator registra um aviso em `issues`, porque provavelmente é erro de cadastro.
-  Mesmo assim o servidor entra no slot.
-- O ingestor aplica segurança **só quando o slot tem `certificate`**. Nesse caso usa `security.mode` e
-  `security.policy`. Sem `certificate`, conecta sem segurança, e um servidor que exige certificado recusa a conexão.
-  O erro aparece no log e nas métricas de conexão do ingestor.
-- Não existe certificado "default": `is_default` e o índice parcial `ux_opc_certificates_default_client` saem do schema.
-
----
-
-## Etapa 1 — Migrations de estrutura (`sientia-dataops-database-migrations`)
-
-**Nova migration** `mongodb/migrations/<timestamp>-opc_certificates_schema.js` (SIENTIAPDE-1974):
-
-1. **Schema da `opc_certificates`**, com o cabeçalho do arquivo descrevendo cada campo:
-
-   | Campo | Tipo | Obrigatório | Descrição |
-   |---|---|---|---|
-   | `certificate_id` | int/long | sim | Id estável, referenciado por `connectors.security.client_certificate_id` |
-   | `kind` | `client` \| `server` \| `ca` | sim | `client`: cert + chave do ingestor. `server`/`ca`: só cert |
-   | `name` | string | sim | Nome legível, único por `kind` |
-   | `status` | `active` \| `retired` | sim | Só `active` entra no slot |
-   | `application_uri` | string | sim se `client` | SAN URI do certificado, usada como `client.application_uri` |
-   | `cert` | `{bucket, object_name, sha256}` | sim (`bucket`, `object_name`) | Objeto no MinIO. `sha256`: string (64 hex) ou null |
-   | `private_key` | `{bucket, object_name, sha256}` | sim se `client`, proibido nos outros | Chave privada no MinIO |
-   | `thumbprint_sha1` | string (40 hex) \| null | não | O que aparece no console do servidor OPC |
-   | `not_before` / `not_after` | date \| null | não | Validade (`not_after` já tem índice) |
-   | `created_at` / `updated_at` | date | sim | Auditoria |
-   | `created_by` / `updated_by` | string | sim | Auditoria: quem cadastrou ou trocou o certificado |
-
-   - "Obrigatório se `client`" fica num `anyOf` do `$jsonSchema`.
-   - Campos fora da tabela (como o rastro do backfill) não são descritos nem validados: sem `additionalProperties: false`.
-   - Índices: novo `ux_opc_certificates_kind_name` em `{kind, name}`, único. Sai `ux_opc_certificates_default_client` (D5). Ficam `ux_opc_certificates_certificate_id` e `ix_opc_certificates_not_after`.
-   - `validationLevel: strict`.
-   - `down`: volta ao validator e aos índices da `20261028170000`.
-
-2. **Schema dos `connectors`** (parte do schema da `20261028190000-connectors_pi_web_api_types.js`):
-   - Saem `use_security` e `server_uri` (D4).
-   - `security` passa a ser obrigatório para `type: 'opc_ua'`, com `mode` e `policy` obrigatórios.
-   - `security.client_certificate_id` continua opcional/nullable (D5).
-   - `down` restaura o schema anterior.
-
-3. **Testes** em `tests/`, no padrão de `test_connectors_types.py` e `test_head.py`:
-   - Documentos válidos e inválidos por `kind` (com e sem `private_key` e `application_uri`).
-   - Campos de auditoria ausentes.
-   - Unicidade de `{kind, name}`.
-   - Connector `opc_ua` com e sem `security`, e com `client_certificate_id` nulo.
-   - Roundtrip `up` → `down`.
-   - Head esperado.
-
-## Etapa 2 — Provisionamento (`sientia-helm-chart`, branch `feature/SIENTIAPDE-1974-connectors-pi-web-api`)
-
-1. `values/minio.yaml`: bucket `opc-certificates` em `provisioning.buckets` (linhas 516-523).
-2. Acesso: o ingestor usa as credenciais root do MinIO (`opc-ingestor/values.yaml:106-115`). Avaliar com quem cuida
-   da plataforma um usuário e uma policy de **somente leitura** em `opc-certificates`, como `model-training-user`.
-3. `src/charts-internal/opc-ingestor/values.yaml:98-117`: remover a env `CERT_BUCKET` e o comentário sobre
-   `platform_certificates`. Remover `READ_ENGINE`, se estiver definida em algum profile (D2).
-4. `docs/nats-broker.md:373`: trocar a menção à `platform_certificates` pelo fluxo novo. O NATS não muda (D1).
-
-## Etapa 3 — Orchestrator (`sientia-dataops-orchestrator_temporal`)
-
-1. **`orchestrator/workflows/orchestrator.py` (`DEFAULT_CONNECTORS_QUERY`)**:
-   - Projetar `security.mode`, `security.policy` e `security.client_certificate_id`. Saem `server_uri` e `use_security`.
-   - `$lookup` de `security.client_certificate_id` para `opc_certificates.certificate_id`, com `status: 'active'` e `kind: 'client'`.
-   - Projetar `certificate_id`, `application_uri`, `cert.{bucket,object_name}` e `private_key.{bucket,object_name}`. As datas não entram (BSON que o payload do Temporal não carrega).
-2. **`orchestrator/utils/opc_slots.py`**: servidor do slot v2 (contrato C1):
-   ```python
-   {
-       'server_id': ..., 'name': ..., 'url': ...,
-       'subscription_period_ms': ...,
-       'security': {'mode': 'SignAndEncrypt', 'policy': 'Basic256Sha256'},
-       'certificate': {                   # None: sem certificado registrado (D5)
-           'certificate_id': 1,
-           'application_uri': 'urn:sientia:opc-ingestor',
-           'cert': {'bucket': 'opc-certificates', 'object_name': 'client/sientia-platform/opc-ingestor-cert.pem'},
-           'private_key': {'bucket': 'opc-certificates', 'object_name': 'client/sientia-platform/opc-ingestor-key.pem'},
-       },
-       'tags': {...},
-   }
-   ```
-3. **`orchestrator/utils/tag_database.py`**:
-   - `CONNECTOR_FIELDS` sem `server_uri`.
-   - `client_certificate_id` preenchido, mas sem certificado ativo do tipo `client` → `certificate: None` mais aviso em `issues` e em `formatters.py` (D5).
-4. **Testes**:
-   - `test_opc_slots.py`, `test_tag_database.py`, `tag_database_fixtures.py`, `test_formatters.py`.
-   - `test_orchestrator.py` e a history de replay, se o input do workflow mudar.
-   - Cenários e2e.
-5. **Docs**: README (seção "OPC slots", contrato C1) e CLAUDE.md.
-
-## Etapa 4 — Ingestor (`sientia-dataops-opc-ingestor`, branch `release/SIENTIAPDE-1973`)
-
-**Certificado e segurança**
-
-| Arquivo | Mudança |
-|---|---|
-| `ingestor/connection/certs.py` | Sai o Mongo (`COLLECTION_NAME`, `find_one`, o `mongo` do construtor). `fetch(certificate)` baixa o `bucket/object_name` vindo do slot. Cache, lock e pasta (`<cache_dir>/<certificate_id>/`) separados por `certificate_id`. `invalidate(certificate_id)` |
-| `ingestor/config.py` | Sai o `cert_bucket`; fica o `cert_cache_dir` |
-| `ingestor/connection/security.py` | `apply_security(client, security, certificate, cert_path, key_path)`, com o `application_uri` do certificado e mode e policy do slot (mapa de string para classe do asyncua; desconhecida → erro) |
-| `ingestor/connection/pool.py` | `_connect`: aplica segurança só quando `certificate` não é `None` (D5). Sai o `server_uri`. `_open` invalida o certificado da conexão. `_check_cfg` passa a comparar `security` e `certificate` (só aviso; a reconexão vem do `ingestor_manager`) |
-| `ingestor/io/slots.py` | `SERVER_FIELDS`: `server_id`, `url`, `subscription_period_ms`, `security`, `certificate`. Sai a conversão do slot v1 |
-
-**Escrita (D1)**
-
-| Arquivo | Mudança |
-|---|---|
-| `ingestor/connection/servers.py` | `ServerConfigResolver`: pool primeiro (`config_for`); senão, Mongo, com o cache de TTL atual. `_lookup` faz uma aggregation em `connectors` (`type: 'opc_ua'`, `status: 'Active'`) com `$lookup` em `opc_certificates`, e devolve a config no formato do servidor no slot (`server_id`, `url`, `security`, `certificate`). Saem `opc-servers` (`OPC_SERVERS`, `_opc_server_cfg`, `_id_forms`, já que `connector_id` é slug string) e o override pelo comando |
-| `ingestor/connection/certificates.py` (novo) ou dentro de `servers.py` | Função que, a partir do documento do connector e do certificado do `$lookup`, devolve `certificate` ou `None` com a regra de D5 (certificado `active` e `client`; id preenchido sem certificado válido → aviso no log). Testes que espelham os do orchestrator |
-| `ingestor/engines/write.py` | `_server_cfg` sem o ramo de conexão vinda do comando. Sem mudança de comportamento: servidor não encontrado continua indo para erro no `write_status` |
-
-**Remoção do legado (D2)**
-
-| Arquivo | Mudança |
-|---|---|
-| `ingestor/managers/opc_manager.py` | Removido |
-| `ingestor/managers/ingestor_manager.py` | Saem `READ_ENGINES`/`READ_ENGINE`, os ramos `legacy`, a checagem de slot v1 (`:698`) e `_warn_server_cert_path` (`:708`) |
-| `ingestor/io/data_sink.py` | Sai o tratamento de slot v1 (`:56`) |
-| `ingestor/gateway_runtime.py` | `CertProvider` sem Mongo. `ServerConfigResolver` continua com Mongo (D1) |
-| `tests/` | Remover os testes do legado; adaptar `certs`, `pool`, `slots`, `servers`, `ingestor_manager` e `write` |
-| `README.md`, `CLAUDE.md` | Sem plano B e sem `platform_certificates` (README:67 e :236) |
-
-## Etapa 5 — Deploy (por ambiente)
-
-1. Migrations (Etapa 1). Os dados precisam estar migrados antes (fora do escopo): o `strict` rejeita documento fora do schema.
-2. Helm (Etapa 2).
-3. Ingestor e orchestrator **juntos**. Sem transição, um slot antigo não tem `security`/`certificate`.
-4. Validação no dev:
-   - O slot no Redis tem `security` e `certificate`.
-   - O ingestor conecta ao `kepserver` com o mode e a policy do banco.
-   - A troca de `client_certificate_id` reconecta com o certificado novo.
-   - Um servidor sem certificado conecta sem segurança.
-   - Um id de certificado inválido gera aviso.
-   - Uma escrita para um servidor que está no pool usa a conexão do pool.
-   - Uma escrita para um servidor fora do pool resolve pelo Mongo, com o mesmo certificado que o slot teria.
-   - Uma escrita para um `server_id` inexistente termina em erro no `write_status`.
-
-## Fora do escopo
-
-- Migração e correção de dados (`opc_certificates`, `connectors`, objetos no MinIO).
-- Remoção da collection `platform_certificates` (o ingestor só deixa de lê-la).
-- Validação do certificado do servidor (`security.server_certificate_id`, `kind: server`).
-- Métrica e alerta de expiração (`not_after`).
-- Conferência de `sha256` no download.
-- Upload e CRUD de certificados no front Angular/BFF (quem preenche `created_by`/`updated_by`).
-- **Card novo — Escrita OPC roteada ao pod dono da conexão:** subject por servidor (`opc.write.<server_id>`), slot
-  cobrindo servidores só de escrita e tratamento de comando sem destino (expiração do `pending` ou `request` com
-  *no responders*). Quando estiver pronto, remove o fallback no Mongo do `ServerConfigResolver` e a segunda fonte da
-  regra de certificado (D1).
-
-## Referências
-
-- Plano geral: `docs/planning/SIENTIAPDE-1972-banco-de-tags-e-certificados.md` (§5.1, §5.2, Fases 1–3).
-- Migrations: `mongodb/migrations/20261028170000-opc_tag_database_collections.js`, `20261028190000-connectors_pi_web_api_types.js`.
-- Escrita: `sientia-dataops-laborious` `laborious/activities/nats_write.py`; ingestor `ingestor/connection/servers.py`, `ingestor/engines/write.py:148-160`; `sientia-helm-chart/docs/nats-broker.md`.
-- Streamlit (origem da `platform_certificates`): `sientia-projects-templates` `origin/rc/SIENTIAPDE-1588` `app/pages/connectors.py`.

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()

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "ingestor"
-version = "2.0.0-rc4"
+version = "2.1.0"
 description = "Sientia DataOps Ingestor - OPC Tag Ingestor"
 readme = "README.md"
 requires-python = ">=3.11"
@@ -59,6 +59,11 @@ ignore = [
     "S105",  # hardcoded passwords ok in tests
     "S106",  # hardcoded passwords ok in tests
 ]
+"e2e/**/*.py" = [
+    "S101",  # assert allowed in tests
+    "S105",  # hardcoded passwords ok in tests
+    "S106",  # hardcoded passwords ok in tests
+]
 
 [tool.ruff.lint.mccabe]
 max-complexity = 15
@@ -129,7 +134,7 @@ addopts = [
 ]
 markers = [
     "asyncio: marks tests as async",
-    "integration: marks tests as integration tests",
+    "integration: marks tests as integration tests (the e2e/ suite: Docker required, run with -o testpaths=e2e -m integration)",
     "unit: marks tests as unit tests",
 ]
 

requirements-dev.txt

@@ -13,6 +13,8 @@ types-requests>=2.31.0   # Type stubs for requests
 pytest>=7.4.0            # Testing framework
 pytest-cov>=4.1.0        # Coverage plugin for pytest
 pytest-asyncio>=0.21.0   # Async test support (already in main requirements)
+testcontainers[postgres,minio,mongodb,redis]>=4.15.0  # E2E suite (e2e/): real MongoDB, Redis, PostgreSQL,
+                         # MinIO and NATS containers; `testcontainers.community.*` namespace
 
 # Development Tools
 ipython>=8.12.0          # Enhanced Python shell

requirements-local.txt

@@ -1,4 +1,4 @@
-asyncua==1.1.5
+asyncua>=2.0
 redis
 git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2
 prometheus_client

requirements.txt

@@ -1,4 +1,6 @@
-asyncua
+# >=2.0: WriteEngine builds DataValue(StatusCode=...); 1.x names the field StatusCode_,
+# so every write would fail with TypeError (reported as bad_timestamp).
+asyncua>=2.0
 redis
 # 1.12.5 builds the Postgres engine with an explicit postgresql+psycopg2:// URL: SQLAlchemy 2.1
 # defaults postgresql:// to psycopg (v3), which is not installed ("No module named 'psycopg'").

tests/functional/conftest.py

@@ -1,31 +0,0 @@
-# import subprocess
-# from time import sleep
-# import pytest
-# from redis import Redis
-
-
-# @pytest.fixture(scope="session", autouse=True)
-# def docker_compose():
-#     """Sobe os containers antes dos testes e derruba depois."""
-#     print("\n🚀 Subindo Docker Compose...")
-#     subprocess.run(["docker", "compose", "up", "-d"], check=True)
-
-#     print("⏳ Aguardando containers ficarem prontos...")
-#     sleep(15)  # ajuste conforme necessário
-
-#     yield  # os testes rodam aqui
-
-#     print("\n🧹 Derrubando Docker Compose...")
-#     subprocess.run(["docker", "compose", "down"], check=True)
-
-
-# @pytest.fixture()
-# def redis_client():
-#     redis = Redis(host="localhost", port=6379, decode_responses=True)
-#     redis.flushdb()
-
-#     yield redis
-
-#     # Limpa o banco de dados após os testes
-#     redis.flushdb()
-#     redis.close()

tests/functional/test_single_node.py

@@ -1,93 +0,0 @@
-# import json
-# import subprocess
-# from time import sleep
-
-
-# new_data = {
-#     "slot:opc_tags:1": {
-#         "server1": {
-#             "name": "server1",
-#             "url": "opc.tcp://simulator:4840",
-#             "server_uri": "http://opcua-server.simulator",
-#             "tags": {
-#                 'ns=2;i=2': {
-#                     'tag_name': 'Counter',
-#                     'frequency': 1000,
-#                     'topics': [],
-#                 },
-#                 'ns=2;i=3': {
-#                     'tag_name': 'Rollout',
-#                     'frequency': 1000,
-#                     "topics": [],
-#                 },
-#                 'ns=2;i=4': {
-#                     'tag_name': 'Square',
-#                     'frequency': 1000,
-#                     "topics": [],
-#                 },
-#             }
-#         }
-#     },
-#     "slot:opc_tags:2": {
-#         "server2": {
-#             "name": "server2",
-#             "url": "opc.tcp://simulator:4840",
-#             "server_uri": "http://opcua-server.simulator",
-#             "tags": {
-#                 'ns=2;i=2': {
-#                     'tag_name': 'Counter',
-#                     'frequency': 1000,
-#                     'topics': [],
-#                 },
-#                 'ns=2;i=3': {
-#                     'tag_name': 'Rollout',
-#                     'frequency': 1000,
-#                     "topics": [],
-#                 },
-#                 'ns=2;i=4': {
-#                     'tag_name': 'Square',
-#                     'frequency': 1000,
-#                     "topics": [],
-#                 },
-#             }
-#         }
-#     }
-# }
-
-
-# def test_simple(redis_client):
-#     new_data['slot:opc_tags:1']['server1']['tags']['ns=2;i=2']['topics'] = [
-#         'test_topic_1']
-#     redis_client.set("slot:opc_tags:1",
-#                      json.dumps(new_data['slot:opc_tags:1']))
-
-#     sleep(20)  # Espera o Ingestor processar os dados
-
-#     # Check if lease is in Redis
-#     assert redis_client.get("lease:opc_tags:1") == 'ingestor'
-#     assert redis_client.get("heartbeat:ingestor:ingestor") == '1'
-
-
-# def test_simple_double_slot(redis_client):
-
-#     new_data['slot:opc_tags:1']['server1']['tags']['ns=2;i=2']['topics'] = [
-#         'test_topic_double_slot1']
-#     redis_client.set("slot:opc_tags:1",
-#                      json.dumps(new_data['slot:opc_tags:1']))
-
-#     sleep(20)  # Espera o Ingestor processar os dados
-
-#     assert redis_client.get("lease:opc_tags:1") == 'ingestor'
-#     assert redis_client.get("heartbeat:ingestor:ingestor") == '1'
-
-#     new_data['slot:opc_tags:2']['server2']['tags']['ns=2;i=2']['topics'] = [
-#         'test_topic_double_slot2']
-#     redis_client.set("slot:opc_tags:2",
-#                      json.dumps(new_data['slot:opc_tags:2']))
-
-#     sleep(20)  # Espera o Ingestor processar os dados
-
-#     # Check if lease is in Redis
-#     assert redis_client.get("lease:opc_tags:2") == 'ingestor'
-#     assert redis_client.get("lease:opc_tags:1") == 'ingestor'
-#     assert redis_client.get("heartbeat:ingestor:ingestor") == '1'

tests/unit/engines/test_write.py

@@ -140,6 +140,36 @@ async def test_write_engine_flags_off_omit_statuscode_and_timestamp():
     assert dv.SourceTimestamp is None
 
 
+@pytest.mark.asyncio
+async def test_write_engine_timestamp_off_keeps_statuscode():
+    eng = make_engine(write_timestamp=False)
+    node = make_node()
+    eng.pool.get.return_value = make_client(node)
+
+    await eng.write(base_cmd())
+
+    dv = node.write_value.await_args.args[0]
+    assert dv.Value.Value == 42.0
+    assert dv.SourceTimestamp is None
+    assert dv.StatusCode is not None
+    assert dv.StatusCode.is_good()
+
+
+@pytest.mark.asyncio
+async def test_write_engine_quality_off_keeps_timestamp():
+    # The command has no write_quality, so the engine (env) default applies.
+    eng = make_engine(write_quality=False)
+    node = make_node()
+    eng.pool.get.return_value = make_client(node)
+
+    await eng.write(base_cmd())
+
+    dv = node.write_value.await_args.args[0]
+    assert dv.Value.Value == 42.0
+    assert dv.StatusCode is None
+    assert dv.SourceTimestamp == datetime.fromisoformat('2026-07-13T12:00:00Z')
+
+
 @pytest.mark.asyncio
 async def test_per_message_flag_override_disables_timestamp():
     # Engine default is True, but the command explicitly turns it off.

tests/unit/test_config.py

@@ -41,6 +41,7 @@ def test_from_env_overrides(monkeypatch):
     monkeypatch.setenv('GRPC_ENABLED', 'true')
     monkeypatch.setenv('GRPC_PORT', '6000')
     monkeypatch.setenv('WRITE_MIN_PAUSE_S', '0.5')
+    monkeypatch.setenv('WRITE_TIMESTAMP', 'false')
     monkeypatch.setenv('WRITE_QUALITY', 'false')
 
     cfg = GatewayConfig.from_env()
@@ -48,6 +49,7 @@ def test_from_env_overrides(monkeypatch):
     assert cfg.grpc_enabled is True
     assert cfg.grpc_port == 6000
     assert cfg.write_min_pause_s == 0.5
+    assert cfg.write_timestamp is False
     assert cfg.write_quality is False