SIENTIAPDE-1646

SIENTIAPDE-1646 Enhance OPC integration and E2E testing framework

- Updated .gitignore to exclude '.cursor/*' for better file management.
- Added new marker in pyproject.toml for E2E tests using OPC.
- Introduced async OPC server fixtures and improved connection handling in conftest.py.
- Enhanced error handling in OPC activities and added metrics for session management in metrics.py.
- Updated E2E tests to cover new OPC scenarios, including session errors and reconnect handling.
- Refactored helper functions to improve prediction assertion logic in helpers.py.
- Documented new OPC scenarios in scenarios.md for clarity on expected outcomes.

.gitignore

@@ -55,3 +55,4 @@ mlruns/
 
 relatorio*
 openspec/*
+.cursor/*

docs/opc-communication.md

@@ -0,0 +1,162 @@
+# OPC UA communication (Laborious)
+
+Laborious exports predictions to OPC UA servers through `OpcRepository` ([`laborious/utils/repository/opc_repository.py`](../laborious/utils/repository/opc_repository.py)) and the synchronous Temporal activity layer in [`laborious/activities/opc.py`](../laborious/activities/opc.py). The repository uses `asyncua.sync.Client` (asyncio on a background thread) so activities remain blocking without `async def`.
+
+Implementation plan for session/channel recovery on Tier-1 `Bad*` errors: [`.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md`](../.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md).
+
+## Architecture
+
+```text
+Worker (long-lived)
+  └── OpcRepository per OPC server id (from OPC_CONFIG / env)
+        ├── connect / disconnect / validate_connection (read-only)
+        ├── _connect_locked / _reconnect_locked (under _connection_lock)
+        ├── write_data (single attempt per call)
+        └── background reconnect on Tier-1 Bad*
+
+Temporal activity write_opc_data
+  └── OPC.manage_output_tags → write_data per tag (sequential per activity)
+```
+
+One worker process holds one `OpcRepository` instance per configured server. Multiple Temporal activities can call `write_data` concurrently on the same repository.
+
+## Connection lifecycle
+
+| Phase | Behavior |
+|-------|----------|
+| Startup | `init_opc()` creates repositories and calls `connect()` → `_connect_locked()` |
+| Steady state | `validate_connection()` is read-only (`protocol.state` only); `_session_ready` is checked in `write_data` |
+| Tier-1 Bad* | `_start_reconnect_on_bad` → `_run_reconnect_on_bad` → `_reconnect_locked()` (respects `reconnection_interval`) |
+| Write | `write_data()` checks `_session_ready`, validates, then one `get_node` + `write_value` |
+| Shutdown | `close()` disconnects all repositories |
+
+### Session and channel timeouts
+
+Requested session and secure-channel lifetime: **10 minutes** (`OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS` in `opc_repository.py`). The server may revise these values; negotiated values are logged after connect and exposed as `opc_session_revised_timeout_milliseconds`.
+
+### Reconnection interval
+
+`OPC_RECONNECTION_INTERVAL` is in **seconds** (default `120`). It gates **background** reconnect after Tier-1 `Bad*` (`last_reconnection_time` is updated only in `_reconnect_locked()`). It limits load on the OPC server when many workflows fail at once.
+
+## Concurrency: connection lock and session readiness
+
+To allow **multiple concurrent writes** when the session is healthy, but **block all writes** while the connection is being torn down or re-established:
+
+| Primitive | Role |
+|-----------|------|
+| `_connection_lock` (`asyncio.Lock`) | Held for the entire `disconnect` → `connect` path. Only one connection-maintenance task at a time. |
+| `_session_ready` (`asyncio.Event`) | Set when a session is ready for writes; cleared before reconnect starts and set again after a successful connect. |
+
+**Connection methods (caller holds `_connection_lock` for `_*_locked` helpers):**
+
+| Method | Role |
+|--------|------|
+| `_create_client()` | Create asyncua `Client` + optional `set_security`; raises if `client` already exists |
+| `_open_session()` | `client.connect()` + metrics; raises if session already open or client missing |
+| `_connect_locked()` | `_create_client()` (when needed) + `_open_session()`; raises if already connected |
+| `_disconnect_locked()` | Teardown session and clear `client` |
+| `_reconnect_locked()` | `_disconnect_locked()` + `_connect_locked()`; sets `last_reconnection_time` |
+
+Public `connect()` / `disconnect()` acquire the lock and call `_connect_locked()` / `_disconnect_locked()`.
+
+**Write path (`write_data`):**
+
+1. If `_session_ready` is cleared → **fail immediately** (`opc_error_kind=reconnect_in_progress`).
+2. `validate_connection()` checks `protocol.state` only (read-only).
+3. Single `get_node` + `write_value` (no retry).
+
+**Reconnect path (`_run_reconnect_on_bad`):**
+
+1. `_start_reconnect_on_bad` clears `_session_ready` and schedules the task when the interval allows.
+2. `async with _connection_lock:` → `_reconnect_locked()`.
+3. `_session_ready` is set on successful `_open_session()`.
+
+A second `_connect_locked()` while a session is already open raises `OpcSessionAlreadyConnectedError` (disconnect first).
+
+**asyncua note:** Concurrent `write_value` on the same session is only safe if the stack tolerates it. If production shows issues, serialize writes with an optional `asyncio.Semaphore(1)` while keeping the connection lock semantics above.
+
+**Future threads:** replace `asyncio.Lock` / `Event` with `threading` primitives or route all OPC I/O through one dedicated loop.
+
+## Tier-1 `Bad*` errors and reconnect
+
+When the server invalidates the session (e.g. `BadSessionIdInvalid`) but the client still sees transport as open, `write_data` fails once, records the OPC status in metrics, and **schedules** reconnect if:
+
+- The exception is a `UaStatusCodeError` whose name is in `RECONNECTABLE_OPC_BAD_NAMES` (see plan), and
+- `reconnection_interval` has elapsed since `last_reconnection_time`, and
+- No reconnect task is already running.
+
+There is **no write retry**: the failed export is not sent again in the same activity.
+
+## Prediction confidence and PostgreSQL comments
+
+| `prediction_confidence` | Meaning |
+|-------------------------|---------|
+| (unchanged) | Successful OPC export |
+| **12** | Generic OPC write failure (`OPC_WRITTING_ERROR_CONFIDENCE`) |
+| **14** | Tier-1 session/channel `Bad*` on export (`OPC_SESSION_BAD_CONFIDENCE`) |
+| **14** | Write while reconnect in progress (`OPC_SESSION_BAD_CONFIDENCE`, comment `OPC UA reconnect in progress`) |
+| **13** | PI Web API write failure (separate path) |
+
+Session/channel errors use a stable comment for counting:
+
+```text
+OPC UA session/channel error: BadSessionIdInvalid
+```
+
+Reconnect-in-progress exports use:
+
+```text
+OPC UA reconnect in progress
+```
+
+Example SQL:
+
+```sql
+SELECT count(*) FROM predictions WHERE prediction_confidence = 14;
+SELECT count(*) FROM predictions WHERE comments LIKE 'OPC UA session/channel error:%';
+```
+
+## Prometheus metrics (`opc_*`)
+
+Defined in [`laborious/metrics.py`](../laborious/metrics.py). Do not rename in production without a dashboard migration.
+
+| Metric | Purpose |
+|--------|---------|
+| `opc_connections_initiated_total` | Connection attempts |
+| `opc_connections_failed_total` | Failed connects |
+| `opc_connection_status` | Gauge 1=connected, 0=disconnected |
+| `opc_session_created_total` | Session established after connect |
+| `opc_session_closed_total` | Disconnect initiated |
+| `opc_session_revised_timeout_milliseconds` | Negotiated session timeout (ms) |
+| `opc_write_attempts_total` | Per write; label `result` = `OK` or exception name |
+| `opc_write_inter_arrival_over_session_timeout_total` | Successful writes spaced longer than revised session timeout |
+
+Legacy activity metrics: `laborious_prediction_opc_writing_count`, `laborious_prediction_opc_writing_response_time_monitor`.
+
+## Environment variables
+
+| Variable | Default | Description |
+|----------|---------|-------------|
+| `OPC_CONFIG` | — | JSON map of server configs (overrides single-server env) |
+| `OPC_ID` | `1` | Server id |
+| `OPC_URL` | `opc.tcp://localhost:4840` | Endpoint |
+| `OPC_SERVER_NAME` | `default_server` | Label for metrics/logs |
+| `OPC_SERVER_URI` | same as URL | Application URI / cert SAN |
+| `OPC_CERT_PATH` | — | Client certificate (secure mode) |
+| `OPC_PRIVATE_KEY_PATH` | — | Client private key |
+| `OPC_SERVER_CERT_PATH` | — | Server certificate |
+| `OPC_RECONNECTION_INTERVAL` | `120` | Minimum seconds between reconnects |
+
+## Operations checklist
+
+- Correlate `BadSessionIdInvalid` in `opc_write_attempts_total` with `opc_session_closed_total` / `opc_session_created_total` (reconnect may finish after the row is stored with confidence 14).
+- Use confidence **14** and comment prefix for session invalidation rates; use **12** for other OPC failures.
+- Respect `OPC_RECONNECTION_INTERVAL` under parallel load; bursts of confidence 14 are expected until the next successful cycle.
+
+## Related tests
+
+- Unit: [`tests/laborious/utils/repository/test_opc_repository.py`](../tests/laborious/utils/repository/test_opc_repository.py)
+- Unit: [`tests/laborious/activities/test_opc.py`](../tests/laborious/activities/test_opc.py)
+- E2E (mock OPC): [`e2e/test_predictions_batch_format_export.py`](../e2e/test_predictions_batch_format_export.py)
+- E2E (in-process asyncua server + real `OpcRepository`): [`e2e/test_opc_real_server.py`](../e2e/test_opc_real_server.py) — scenarios 3.1.2, 3.2.2, 3.2.4, 3.2.5
+- Scenarios: [`e2e/scenarios.md`](../e2e/scenarios.md)

e2e/conftest.py

@@ -1,5 +1,7 @@
 """Pytest configuration and fixtures for E2E tests."""
 
+import asyncio
+import time
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from unittest.mock import MagicMock
@@ -14,6 +16,7 @@ from testcontainers.postgres import PostgresContainer
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
 
+from e2e.opc_test_server import OpcE2ETestServer
 from laborious.activities.activities import Activities
 from laborious.workflows.drift import Drift
 from laborious.workflows.minimal_retrain import MinimalRetrain
@@ -461,3 +464,140 @@ async def temporal_worker_minimal_retrain(temporal_test_env, test_activities):
             yield worker
 
 
+def _connect_activities_to_opc_server(activities: Activities, server_url: str) -> None:
+    """
+    Initialize OPC repositories and block until the E2E server session is ready.
+
+    Runs synchronously (typically via ``asyncio.to_thread``) so the asyncua test
+    server event loop is not blocked during ``Client.connect()``.
+
+    Args:
+        activities (Activities): Worker activities under test.
+        server_url (str): ``opc.tcp://`` URL from ``OpcE2ETestServer``.
+    """
+    activities.init_opc()
+    repo = activities.opc_repository['1']
+    deadline = time.monotonic() + 30.0
+    while time.monotonic() < deadline:
+        if repo._session_ready.is_set():
+            return
+        connected, _ = repo.connect()
+        if connected:
+            return
+        time.sleep(0.5)
+    raise RuntimeError(f'Could not connect OpcRepository to OPC E2E server at {server_url}')
+
+
+def _build_e2e_opc_config(server_url: str) -> dict[str, dict]:
+    """
+    OPC server config for E2E Activities pointing at an in-process asyncua server.
+
+    Args:
+        server_url (str): ``opc.tcp://`` endpoint from ``OpcE2ETestServer``.
+
+    Return:
+        dict: ``opc_config`` payload for ``Activities`` (server id ``1``).
+    """
+    return {
+        '1': {
+            'id': '1',
+            'server_name': 'e2e_opcua',
+            'url': server_url,
+            'server_uri': server_url,
+            'cert_path': None,
+            'private_key_path': None,
+            'server_cert_path': None,
+            'reconnection_interval': 0,
+        }
+    }
+
+
+@pytest_asyncio.fixture
+async def opc_e2e_server():
+    """In-process asyncua server with writable prediction/confidence nodes."""
+    server = OpcE2ETestServer()
+    await server.start()
+    await asyncio.sleep(0.5)
+    try:
+        yield server
+    finally:
+        await server.stop()
+
+
+@pytest_asyncio.fixture(scope='function')
+async def test_activities_real_opc(
+    postgres_container,
+    minio_container,
+    mock_logger,
+    notification_handler,
+    metrics_controller,
+    mlflow_repository_stub,
+    plugin_store_stub,
+    pi_web_api_client_stub,
+    opc_e2e_server: OpcE2ETestServer,
+):
+    """Activities with real OpcRepository connected to the in-process OPC UA server."""
+    minio_client = minio_container.get_client()
+    if not minio_client.bucket_exists('test-bucket'):
+        minio_client.make_bucket('test-bucket')
+    minio_port = minio_container.get_exposed_port(9000)
+
+    activities = Activities(
+        postgres_config={
+            'host': 'localhost',
+            'port': int(postgres_container.get_exposed_port(5432)),
+            'user': postgres_container.username,
+            'password': postgres_container.password,
+            'dbname': postgres_container.dbname,
+            'min_connections': 1,
+            'max_connections': 5,
+        },
+        plugin_store=plugin_store_stub,
+        minio_config={
+            'endpoint_url': f'localhost:{minio_port}',
+            'access_key': 'minioadmin',
+            'secret_key': 'minioadmin',
+            'default_bucket': 'test-bucket',
+            'retention_hours': 24,
+            'secure': False,
+        },
+        opc_config=_build_e2e_opc_config(opc_e2e_server.url),
+        pi_web_api_config={
+            'base_url': 'http://localhost:8080',
+            'auth_type': 'bearer',
+            'auth_token': 'test_token',
+        },
+        logger=mock_logger,
+        notification_handler=notification_handler,
+        metrics_controller=metrics_controller,
+        mlflow_repository=mlflow_repository_stub,
+    )
+    activities.pi_web_api_client = pi_web_api_client_stub
+    await asyncio.to_thread(_connect_activities_to_opc_server, activities, opc_e2e_server.url)
+    try:
+        yield activities
+    finally:
+        await asyncio.to_thread(_teardown_real_opc_activities, activities)
+
+
+def _teardown_real_opc_activities(activities: Activities) -> None:
+    """Disconnect OPC sessions and shut down activities (sync, for asyncio.to_thread)."""
+    for opc_repo in activities.opc_repository.values():
+        opc_repo.disconnect()
+    activities.shutdown()
+
+
+@pytest_asyncio.fixture(scope='function')
+async def temporal_worker_real_opc(temporal_test_env, test_activities_real_opc):
+    """Temporal worker using real OpcRepository against the in-process OPC UA server."""
+    with ThreadPoolExecutor(max_workers=32) as activity_executor:
+        async with Worker(
+            temporal_test_env.client,
+            task_queue='test-queue',
+            workflows=[PredictionsBatch, PredictionProcess, FormatAndExportPrediction],
+            activities=_worker_activity_list(test_activities_real_opc),
+            activity_executor=activity_executor,
+        ) as worker:
+            yield worker
+
+

e2e/helpers.py

@@ -205,6 +205,7 @@ def assert_prediction(
     prediction_confidence: int | Decimal = 0,
     prediction_status: str = 'Good',
     comments: str = '',
+    comments_contains: str | None = None,
 ) -> None:
     """
     Assert exactly one prediction row exists for model_id with expected columns.
@@ -215,10 +216,9 @@ def assert_prediction(
         prediction: Expected prediction value.
         prediction_confidence: Expected confidence (int or Decimal for numeric column).
         prediction_status: Expected status string.
-        comments: Expected comments string.
+        comments: Expected exact comments string (ignored when ``comments_contains`` is set).
+        comments_contains: When set, assert this substring appears in comments.
     """
-    import pytest
-
     with postgres_engine.connect() as conn:
         result_query = conn.execute(
             text(
@@ -238,7 +238,13 @@ def assert_prediction(
             str(prediction_confidence)
         ), f'Expected prediction_confidence={prediction_confidence}, got {row[2]}'
         assert row[3] == prediction_status, f"Expected prediction_status='{prediction_status}', got {row[3]}"
-        assert row[4] == comments, f"Expected comments='{comments}', got {row[4]}"
+        actual_comments = row[4] or ''
+        if comments_contains is not None:
+            assert comments_contains in actual_comments, (
+                f"Expected comments to contain '{comments_contains}', got '{actual_comments}'"
+            )
+        else:
+            assert actual_comments == comments, f"Expected comments='{comments}', got '{actual_comments}'"
 
 
 def assert_continue(

e2e/opc_test_server.py

@@ -0,0 +1,189 @@
+"""
+In-process OPC UA server for E2E tests (asyncua).
+
+Provides writable prediction/confidence nodes and optional write faults
+(Tier-1 BadSessionIdInvalid via PreWrite callback).
+"""
+
+from __future__ import annotations
+
+import socket
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from asyncua import Server, ua
+from asyncua.common.callback import CallbackType
+from asyncua.common.utils import ServiceError
+
+if TYPE_CHECKING:
+    from asyncua.common.node import Node
+
+
+UNKNOWN_NODE_ID = 'ns=99;i=9999'
+
+
+@dataclass(frozen=True)
+class OpcE2ENodeIds:
+    """NodeId strings used in opc_output_config for E2E workflows."""
+
+    prediction: str
+    confidence: str
+    unknown: str = UNKNOWN_NODE_ID
+
+
+class OpcE2ETestServer:
+    """
+    Ephemeral asyncua server with Laborious E2E variables and controllable faults.
+
+    Args:
+        host: Bind address (default 127.0.0.1).
+    """
+
+    def __init__(self, host: str = '127.0.0.1') -> None:
+        self._host = host
+        self._server: Server | None = None
+        self._prediction_node: Node | None = None
+        self._confidence_node: Node | None = None
+        self._session_bad_on_write = False
+        self._url: str | None = None
+        self._node_ids: OpcE2ENodeIds | None = None
+
+    @property
+    def url(self) -> str:
+        if self._url is None:
+            raise RuntimeError('OPC E2E server is not started')
+        return self._url
+
+    @property
+    def node_ids(self) -> OpcE2ENodeIds:
+        if self._node_ids is None:
+            raise RuntimeError('OPC E2E server is not started')
+        return self._node_ids
+
+    def set_session_bad_on_write(self, enabled: bool) -> None:
+        """
+        When enabled, every client Write is rejected with BadSessionIdInvalid.
+
+        Args:
+            enabled (bool): Turn Tier-1 session fault injection on or off.
+        """
+        self._session_bad_on_write = enabled
+
+    async def start(self) -> OpcE2ENodeIds:
+        """
+        Start the OPC UA server on a free TCP port.
+
+        Return:
+            OpcE2ENodeIds: NodeId strings for prediction and confidence tags.
+        """
+        port = _free_port(self._host)
+        self._url = f'opc.tcp://{self._host}:{port}/freeopcua/server/'
+
+        server = Server()
+        server.set_endpoint(self._url)
+        await server.init()
+        server.iserver.callback_service.addListener(
+            CallbackType.PreWrite,
+            self._pre_write_callback,
+        )
+
+        idx = await server.register_namespace('http://sientia.test/laborious-e2e')
+        e2e_object = await server.nodes.objects.add_object(idx, 'LaboriousE2E')
+        prediction = await e2e_object.add_variable(
+            idx,
+            'Prediction',
+            ua.Variant(0.0, ua.VariantType.Float),
+        )
+        confidence = await e2e_object.add_variable(
+            idx,
+            'Confidence',
+            ua.Variant(0.0, ua.VariantType.Float),
+        )
+        await prediction.set_writable()
+        await confidence.set_writable()
+
+        await server.start()
+        self._server = server
+        self._prediction_node = prediction
+        self._confidence_node = confidence
+        self._node_ids = OpcE2ENodeIds(
+            prediction=prediction.nodeid.to_string(),
+            confidence=confidence.nodeid.to_string(),
+        )
+        return self._node_ids
+
+    async def stop(self) -> None:
+        """Stop the OPC UA server and release the listening port."""
+        if self._server is not None:
+            await self._server.stop()
+            self._server = None
+        self._prediction_node = None
+        self._confidence_node = None
+        self._url = None
+        self._node_ids = None
+        self._session_bad_on_write = False
+
+    async def read_prediction(self) -> float:
+        """
+        Read the current prediction variable value from the address space.
+
+        Return:
+            float: Stored prediction value.
+        """
+        if self._prediction_node is None:
+            raise RuntimeError('OPC E2E server is not started')
+        value = await self._prediction_node.read_value()
+        return float(value)
+
+    async def read_confidence(self) -> float:
+        """
+        Read the current confidence variable value from the address space.
+
+        Return:
+            float: Stored confidence value.
+        """
+        if self._confidence_node is None:
+            raise RuntimeError('OPC E2E server is not started')
+        value = await self._confidence_node.read_value()
+        return float(value)
+
+    async def _pre_write_callback(self, _event, _service) -> None:
+        if self._session_bad_on_write:
+            raise ServiceError(ua.StatusCodes.BadSessionIdInvalid)
+
+
+def _free_port(host: str) -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind((host, 0))
+        return int(sock.getsockname()[1])
+
+
+def build_opc_output_config(
+    node_ids: OpcE2ENodeIds,
+    *,
+    prediction_tag: str | None = None,
+    confidence_tag: str | None = None,
+    prediction_only: bool = False,
+    server_key: str = '1',
+) -> dict[str, dict]:
+    """
+    Build opc_output_config for PredictionsBatch using real server NodeIds.
+
+    Args:
+        node_ids (OpcE2ENodeIds): Node ids from OpcE2ETestServer.
+        prediction_tag (str | None): Override prediction NodeId (default: node_ids.prediction).
+        confidence_tag (str | None): Override confidence NodeId (default: node_ids.confidence).
+        prediction_only (bool): When True, omit confidence_tags (single write per activity).
+        server_key (str): OPC server id key in opc_output_config.
+
+    Return:
+        dict: opc_output_config payload for workflow input.
+    """
+    pred = prediction_tag if prediction_tag is not None else node_ids.prediction
+    conf = confidence_tag if confidence_tag is not None else node_ids.confidence
+    server_config: dict = {
+        'prediction_tags': {pred: {'data_type': 'float'}},
+    }
+    if not prediction_only:
+        server_config['confidence_tags'] = {conf: {'data_type': 'float'}}
+    return {server_key: server_config}

e2e/scenarios.md

@@ -10,6 +10,27 @@ It is a functional reference of scenario behavior, inputs, and expected outcomes
 - Tests run under `e2e/` and are marked with `@pytest.mark.integration`.
 - PostgreSQL and MinIO are provisioned with testcontainers.
 - `test_minio_offload.py` uses real MinIO I/O; other scenario suites may use stubs/mocks for optional outputs.
+- Real OPC UA scenarios use `@pytest.mark.opc` and an in-process asyncua server (`e2e/test_opc_real_server.py`).
+
+### Local validation
+
+Use the existing project virtualenv and the shared `validate` script for unit/quality gates; run E2E separately (Docker required).
+
+```bash
+source ./venv/bin/activate
+
+# Auto-fix + static checks (no pytest)
+validate --fix --project-name=laborious
+
+# Full unit + quality gate
+validate --project-name=laborious
+
+# E2E (integration)
+pytest e2e/ --override-ini testpaths=e2e -m integration
+
+# E2E (real OPC server only)
+pytest e2e/test_opc_real_server.py --override-ini testpaths=e2e -m opc
+```
 
 ---
 
@@ -195,6 +216,30 @@ Source: `e2e/test_predictions_batch_format_export.py`
 - Workflow completes.
 - Prediction persisted with PI error confidence and descriptive comment.
 
+#### 3.2.4 OPC session / channel error (confidence 14)
+**Summary**: Tier-1 `BadSessionIdInvalid` (or equivalent session error) degrades the prediction without failing the workflow.
+
+**Sources**:
+- Mock: `e2e/test_predictions_batch_format_export.py::test_scenario_3_2_4_opc_session_bad_mock`
+- Real server: `e2e/test_opc_real_server.py::test_scenario_3_2_4_opc_session_bad_real_server` (`@pytest.mark.opc`)
+
+**Expected Outcome**:
+- Workflow completes.
+- `prediction_confidence` is 14.
+- Comments contain `OPC UA session/channel error: BadSessionIdInvalid`.
+
+#### 3.2.5 OPC write blocked during reconnect (confidence 14)
+**Summary**: While reconnect holds the repository connection lock, writes fail fast with `reconnect_in_progress`.
+
+**Sources**:
+- Mock: `e2e/test_predictions_batch_format_export.py::test_scenario_3_2_5_opc_reconnect_in_progress_mock`
+- Real server: `e2e/test_opc_real_server.py::test_scenario_3_2_5_opc_write_blocked_during_reconnect_real_server` (`@pytest.mark.opc`)
+
+**Expected Outcome**:
+- Workflow completes.
+- `prediction_confidence` is 14.
+- Comments contain `OPC UA reconnect in progress`.
+
 ### 3.3.1 Combined Optional Outputs (PI + OPC)
 **Summary**: Both external output channels are enabled together.
 

e2e/test_opc_real_server.py

@@ -0,0 +1,201 @@
+"""
+E2E tests for OPC export using an in-process asyncua server and real OpcRepository.
+
+Covers scenarios 3.1.2, 3.2.2, 3.2.4, and 3.2.5 from e2e/scenarios.md.
+Mock-based OPC tests remain in test_predictions_batch_format_export.py.
+"""
+
+import asyncio
+import threading
+import time
+
+import pytest
+from temporalio.testing import WorkflowEnvironment
+from temporalio.worker import Worker
+
+from e2e.helpers import assert_prediction, insert_sample_data, make_workflow_id, start_and_await_workflow
+from e2e.opc_test_server import UNKNOWN_NODE_ID, OpcE2ETestServer, build_opc_output_config
+from e2e.test_predictions_batch_format_export import get_base_input_data
+from laborious.activities.activities import Activities
+from laborious.activities.opc import OPC_RECONNECT_IN_PROGRESS_COMMENT
+from laborious.utils.repository.opc_repository import OpcRepository
+from laborious.workflows.predictions_batch import PredictionsBatch
+
+
+def _slow_reconnect_under_lock(repo: OpcRepository, hold_seconds: float = 0.75) -> None:
+    """
+    Hold the connection lock briefly so concurrent writes see reconnect_in_progress.
+
+    Args:
+        repo (OpcRepository): Connected repository.
+        hold_seconds (float): Time to keep the lock before reconnecting.
+    """
+    with repo._connection_lock:
+        time.sleep(hold_seconds)
+        repo._reconnect_locked()
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+@pytest.mark.opc
+async def test_scenario_3_1_2_export_with_opc_only_real_server(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_real_opc: Worker,
+    test_activities_real_opc: Activities,
+    opc_e2e_server: OpcE2ETestServer,
+    postgres_engine,
+):
+    """
+    Scenario 3.1.2 (real OPC): connect, write prediction and confidence, verify server values.
+    """
+    client = temporal_test_env.client
+    model_id = 412
+    insert_sample_data(postgres_engine, model_id, [23.5, 78.2])
+
+    input_data = get_base_input_data(model_id)
+    input_data['opc_output_config'] = build_opc_output_config(opc_e2e_server.node_ids)
+    input_data['pi_web_api_output_config'] = None
+
+    await start_and_await_workflow(
+        client,
+        PredictionsBatch.run,
+        input_data,
+        make_workflow_id('test-opc-real-happy'),
+    )
+
+    test_activities_real_opc.pi_web_api_client.write_value.assert_not_called()
+    assert await opc_e2e_server.read_prediction() == pytest.approx(0.5)
+    assert await opc_e2e_server.read_confidence() == pytest.approx(0.0)
+    assert_prediction(postgres_engine, model_id)
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+@pytest.mark.opc
+async def test_scenario_3_2_2_opc_write_error_real_server(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_real_opc: Worker,
+    test_activities_real_opc: Activities,
+    opc_e2e_server: OpcE2ETestServer,
+    postgres_engine,
+):
+    """
+    Scenario 3.2.2 (real OPC): unknown NodeId yields generic write failure (confidence 12).
+    """
+    client = temporal_test_env.client
+    model_id = 422
+    insert_sample_data(postgres_engine, model_id, [23.5, 78.2])
+
+    node_ids = opc_e2e_server.node_ids
+    input_data = get_base_input_data(model_id)
+    input_data['opc_output_config'] = build_opc_output_config(
+        node_ids,
+        prediction_tag=UNKNOWN_NODE_ID,
+        confidence_tag=UNKNOWN_NODE_ID,
+    )
+    input_data['pi_web_api_output_config'] = None
+
+    await start_and_await_workflow(
+        client,
+        PredictionsBatch.run,
+        input_data,
+        make_workflow_id('test-opc-real-bad-node'),
+    )
+
+    assert_prediction(
+        postgres_engine,
+        model_id,
+        prediction_confidence=12,
+        comments='Some data could not be written to OPC servers',
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+@pytest.mark.opc
+async def test_scenario_3_2_4_opc_session_bad_real_server(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_real_opc: Worker,
+    test_activities_real_opc: Activities,
+    opc_e2e_server: OpcE2ETestServer,
+    postgres_engine,
+):
+    """
+    Scenario 3.2.4 (real OPC): server PreWrite fault injects BadSessionIdInvalid (confidence 14).
+    """
+    client = temporal_test_env.client
+    model_id = 424
+    insert_sample_data(postgres_engine, model_id, [23.5, 78.2])
+
+    opc_e2e_server.set_session_bad_on_write(True)
+    try:
+        input_data = get_base_input_data(model_id)
+        input_data['opc_output_config'] = build_opc_output_config(
+            opc_e2e_server.node_ids,
+            prediction_only=True,
+        )
+        input_data['pi_web_api_output_config'] = None
+
+        await start_and_await_workflow(
+            client,
+            PredictionsBatch.run,
+            input_data,
+            make_workflow_id('test-opc-real-session-bad'),
+        )
+    finally:
+        opc_e2e_server.set_session_bad_on_write(False)
+
+    assert_prediction(
+        postgres_engine,
+        model_id,
+        prediction_confidence=14,
+        comments_contains='OPC UA session/channel error: BadSessionIdInvalid',
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+@pytest.mark.opc
+async def test_scenario_3_2_5_opc_write_blocked_during_reconnect_real_server(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_real_opc: Worker,
+    test_activities_real_opc: Activities,
+    opc_e2e_server: OpcE2ETestServer,
+    postgres_engine,
+):
+    """
+    Scenario 3.2.5 (real OPC): writes rejected while reconnect holds the connection lock.
+    """
+    client = temporal_test_env.client
+    model_id = 425
+    insert_sample_data(postgres_engine, model_id, [23.5, 78.2])
+
+    repo = test_activities_real_opc.opc_repository['1']
+    repo._session_ready.clear()
+    reconnect_thread = threading.Thread(
+        target=_slow_reconnect_under_lock,
+        args=(repo,),
+        daemon=True,
+    )
+    reconnect_thread.start()
+
+    input_data = get_base_input_data(model_id)
+    input_data['opc_output_config'] = build_opc_output_config(opc_e2e_server.node_ids)
+    input_data['pi_web_api_output_config'] = None
+
+    try:
+        await start_and_await_workflow(
+            client,
+            PredictionsBatch.run,
+            input_data,
+            make_workflow_id('test-opc-real-reconnect-block'),
+        )
+    finally:
+        reconnect_thread.join(timeout=5.0)
+
+    assert_prediction(
+        postgres_engine,
+        model_id,
+        prediction_confidence=14,
+        comments_contains=OPC_RECONNECT_IN_PROGRESS_COMMENT,
+    )

e2e/test_predictions_batch_format_export.py

@@ -4,7 +4,7 @@ End-to-end tests for PredictionsBatch workflow - Format and Export scenarios.
 
 from decimal import Decimal
 from typing import Any, cast
-from unittest.mock import ANY, call
+from unittest.mock import call
 
 import pytest
 from sientia_do.notifications.models import NotificationLevel
@@ -123,7 +123,6 @@ async def test_scenario_3_1_1_default_prediction_export(
                 'addr_1',
                 0,
                 'float',
-                ANY,
                 {
                     'model_id': 311,
                     'model_name': 'test_model',
@@ -135,7 +134,6 @@ async def test_scenario_3_1_1_default_prediction_export(
                 'addr_2',
                 2,
                 'float',
-                ANY,
                 {
                     'model_id': 311,
                     'model_name': 'test_model',
@@ -222,20 +220,28 @@ async def test_scenario_3_1_2_export_with_opc_only(
     opc_write_data = cast(Any, test_activities.opc_repository['1'].write_data)
     opc_write_data.assert_has_calls(
         [
-            call('addr_1', 0.5, 'float', ANY,
+            call(
-            {
+                'addr_1',
-                'model_id': 312,
+                0.5,
-                'model_name': 'test_model',
+                'float',
-                'schedule_name': 'test-schedule',
+                {
-                'workflow_name': 'predictions_batch',
+                    'model_id': 312,
-            }),
+                    'model_name': 'test_model',
-            call('addr_2', 0, 'float', ANY,
+                    'schedule_name': 'test-schedule',
-            {
+                    'workflow_name': 'predictions_batch',
-                'model_id': 312,
+                },
-                'model_name': 'test_model',
+            ),
-                'schedule_name': 'test-schedule',
+            call(
-                'workflow_name': 'predictions_batch',
+                'addr_2',
-            }),
+                0,
+                'float',
+                {
+                    'model_id': 312,
+                    'model_name': 'test_model',
+                    'schedule_name': 'test-schedule',
+                    'workflow_name': 'predictions_batch',
+                },
+            ),
         ]
     )
 
@@ -470,20 +476,28 @@ async def test_scenario_3_1_5_export_without_transformed_data(
     opc_write_data = cast(Any, test_activities.opc_repository['1'].write_data)
     opc_write_data.assert_has_calls(
         [
-            call('addr_1', 0.5, 'float', ANY,
+            call(
-            {
+                'addr_1',
-                'model_id': 315,
+                0.5,
-                'model_name': 'test_model',
+                'float',
-                'schedule_name': 'test-schedule',
+                {
-                'workflow_name': 'predictions_batch',
+                    'model_id': 315,
-            }),
+                    'model_name': 'test_model',
-            call('addr_2', 0, 'float', ANY,
+                    'schedule_name': 'test-schedule',
-            {
+                    'workflow_name': 'predictions_batch',
-                'model_id': 315,
+                },
-                'model_name': 'test_model',
+            ),
-                'schedule_name': 'test-schedule',
+            call(
-                'workflow_name': 'predictions_batch',
+                'addr_2',
-            }),
+                0,
+                'float',
+                {
+                    'model_id': 315,
+                    'model_name': 'test_model',
+                    'schedule_name': 'test-schedule',
+                    'workflow_name': 'predictions_batch',
+                },
+            ),
         ]
     )
 
@@ -618,6 +632,109 @@ async def test_scenario_3_2_2_opc_write_error(
     )
 
 
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_scenario_3_2_4_opc_session_bad_mock(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker: Worker,
+    test_activities: Activities,
+    postgres_engine,
+):
+    """
+    Scenario 3.2.4 (mock): Tier-1 session error maps to confidence 14.
+    """
+    client = temporal_test_env.client
+    model_id = 324
+
+    insert_sample_data(postgres_engine, model_id, [23.5, 78.2])
+
+    opc_write_data = cast(Any, test_activities.opc_repository['1'].write_data)
+    opc_write_data.return_value = (
+        False,
+        {
+            'opc_error_kind': 'session_bad',
+            'opc_status': 'BadSessionIdInvalid',
+            'notification_id': 'OPC_WRITE_DATA_ERROR_1',
+            'message': 'OPC session invalid',
+            'block': 'opc_repository',
+            'level': NotificationLevel.ERROR,
+            'attachment_content': 'BadSessionIdInvalid',
+        },
+    )
+
+    input_data = get_base_input_data(model_id)
+    input_data['opc_output_config'] = {
+        '1': {
+            'prediction_tags': {'addr_1': {'data_type': 'float'}},
+            'confidence_tags': {'addr_2': {'data_type': 'float'}},
+        }
+    }
+    input_data['pi_web_api_output_config'] = None
+
+    await start_and_await_workflow(
+        client, PredictionsBatch.run, input_data, make_workflow_id('test-opc-session-bad-mock')
+    )
+
+    assert_prediction(
+        postgres_engine,
+        model_id,
+        prediction_confidence=14,
+        comments_contains='OPC UA session/channel error: BadSessionIdInvalid',
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_scenario_3_2_5_opc_reconnect_in_progress_mock(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker: Worker,
+    test_activities: Activities,
+    postgres_engine,
+):
+    """
+    Scenario 3.2.5 (mock): reconnect_in_progress maps to confidence 14.
+    """
+    client = temporal_test_env.client
+    model_id = 325
+
+    insert_sample_data(postgres_engine, model_id, [23.5, 78.2])
+
+    opc_write_data = cast(Any, test_activities.opc_repository['1'].write_data)
+    opc_write_data.return_value = (
+        False,
+        {
+            'opc_error_kind': 'reconnect_in_progress',
+            'notification_id': 'OPC_WRITE_DATA_ERROR_1',
+            'message': 'OPC reconnect in progress',
+            'block': 'opc_repository',
+            'level': NotificationLevel.ERROR,
+            'attachment_content': 'reconnect',
+        },
+    )
+
+    input_data = get_base_input_data(model_id)
+    input_data['opc_output_config'] = {
+        '1': {
+            'prediction_tags': {'addr_1': {'data_type': 'float'}},
+            'confidence_tags': {'addr_2': {'data_type': 'float'}},
+        }
+    }
+    input_data['pi_web_api_output_config'] = None
+
+    await start_and_await_workflow(
+        client,
+        PredictionsBatch.run,
+        input_data,
+        make_workflow_id('test-opc-reconnect-mock'),
+    )
+
+    assert_prediction(
+        postgres_engine,
+        model_id,
+        prediction_confidence=14,
+        comments_contains='OPC UA reconnect in progress',
+    )
+
 
 @pytest.mark.asyncio
 @pytest.mark.integration

laborious/activities/opc.py

@@ -15,6 +15,45 @@ with workflow.unsafe.imports_passed_through():
     from laborious.utils.repository.opc_repository import OpcRepository
 
 OPC_WRITTING_ERROR_CONFIDENCE = 12
+OPC_SESSION_BAD_CONFIDENCE = 14
+OPC_SESSION_BAD_COMMENT_PREFIX = 'OPC UA session/channel error:'
+OPC_WRITTING_ERROR_MESSAGE = 'Some data could not be written to OPC servers'
+OPC_RECONNECT_IN_PROGRESS_COMMENT = 'OPC UA reconnect in progress'
+OPC_COMMENT_SEPARATOR = ' | '
+
+
+def _opc_session_bad_comment(opc_status: str | None) -> str:
+    status = opc_status or 'Unknown'
+    return f'{OPC_SESSION_BAD_COMMENT_PREFIX} {status}'
+
+
+def _apply_opc_write_error(
+    error_info: dict[str, Any] | None,
+    session_bad_seen: bool,
+    session_bad_status: str | None,
+    reconnect_in_progress_seen: bool,
+) -> tuple[bool, str | None, bool]:
+    """
+    Update session/reconnect flags from an OPC write error payload.
+
+    Args:
+        error_info: Repository error details, or None when the write succeeded.
+        session_bad_seen: Whether a session_bad error was seen so far.
+        session_bad_status: Last known OPC status for session errors.
+        reconnect_in_progress_seen: Whether reconnect_in_progress was seen so far.
+
+    Return:
+        Updated (session_bad_seen, session_bad_status, reconnect_in_progress_seen).
+    """
+    if not error_info:
+        return session_bad_seen, session_bad_status, reconnect_in_progress_seen
+
+    kind = error_info.get('opc_error_kind')
+    if kind == 'session_bad':
+        return True, error_info.get('opc_status', session_bad_status), reconnect_in_progress_seen
+    if kind == 'reconnect_in_progress':
+        return session_bad_seen, session_bad_status, True
+    return session_bad_seen, session_bad_status, reconnect_in_progress_seen
 
 
 class OPC(SientiaMonitoring):
@@ -67,13 +106,14 @@ class OPC(SientiaMonitoring):
                 logger=self.logger,
                 notification_handler=self.notification_handler,
                 metrics_controller=self.metrics_controller,
+                reconnection_interval=server.get('reconnection_interval', 60),
                 server_uri=server['server_uri'],
                 cert_path=server['cert_path'],
                 private_key_path=server['private_key_path'],
                 server_cert_path=server['server_cert_path'],
             )
-            ok, err = self.opc_repository[opc_id].connect()
+            is_connected, error_data = self.opc_repository[opc_id].connect()
-            if not ok:
+            if not is_connected:
                 self.send_notification(
                     metadata={
                         'model_id': '-',
@@ -81,11 +121,11 @@ class OPC(SientiaMonitoring):
                         'workflow_name': '-',
                         'schedule_name': 'INITIALIZATION',
                     },
-                    notification_id=f'OPC_CONNECTION_ERROR_{server.get("id", opc_id)}',
+                    notification_id=error_data['notification_id'],
-                    message=err.get('message', 'Failed to connect to OPC server'),
+                    message=error_data['message'],
-                    block='opc_repository',
+                    block=error_data['block'],
-                    level=NotificationLevel.ERROR,
+                    level=error_data.get('level', NotificationLevel.ERROR),
-                    attachment_content=err.get('attachment_content', traceback.format_exc()),
+                    attachment_content=error_data.get('attachment_content', None),
                 )
             else:
                 self.info(
@@ -101,19 +141,13 @@ class OPC(SientiaMonitoring):
         data_type: str,
         tag_type: str,
         metadata: dict[str, Any],
-    ) -> float | None:
+    ) -> tuple[float | None, dict[str, Any] | None]:
         """
         Write data to a specific OPC server tag with comprehensive error handling.
 
-        Args:
+        Return:
-            - server_id (str): The id of the OPC server.
+            tuple[float | None, dict[str, Any] | None]: Response time on success, or
-            - tag (str): The tag to write to.
+                (None, error info_data) on repository failure.
-            - data (Any): The data to write.
-            - data_type (str): The data type.
-            - tag_type (str): The tag type.
-
-        Returns:
-            - float | None: Response time in seconds if successful, None otherwise.
         """
 
         try:
@@ -129,8 +163,8 @@ class OPC(SientiaMonitoring):
                     level=info_data.get('level', NotificationLevel.ERROR),
                     attachment_content=info_data.get('attachment_content', None),
                 )
-                return None
+                return None, info_data
-            return info_data['response_time']
+            return info_data['response_time'], None
         except Exception as e:
             trace = traceback.format_exc()
             self.send_notification(
@@ -147,10 +181,6 @@ class OPC(SientiaMonitoring):
         """
         Validate that an OPC repository exists for the requested server identifier.
 
-        This guard prevents write attempts against unknown/uninitialized servers.
-        When the server is missing, it emits an error notification with the list
-        of available repositories to help operators diagnose configuration drift.
-
         Args:
          - server_id (str): OPC server identifier from workflow output config.
          - metadata (dict[str, Any]): Workflow metadata used for logs/alerts.
@@ -172,72 +202,116 @@ class OPC(SientiaMonitoring):
             return False
         return True
 
+    def _write_tags_from_config(
+        self,
+        server_id: str,
+        tags_config: dict[str, dict[str, Any]],
+        data: DataFrame,
+        data_column: str,
+        tag_type: str,
+        log_label: str,
+        metadata: dict[str, Any],
+    ) -> tuple[dict[str, float | None], bool, str | None, bool]:
+        """
+        Write a group of OPC tags and collect response times and error flags.
+
+        Args:
+            server_id: Target OPC server identifier.
+            tags_config: Tag name to configuration mapping.
+            data: DataFrame with prediction/confidence columns.
+            data_column: Column name whose first row value is written.
+            tag_type: Tag category passed to write_data ('prediction' or 'confidence').
+            log_label: Human-readable label for success logs.
+            metadata: Context metadata for logging and notifications.
+
+        Return:
+            (response_times, session_bad_seen, session_bad_status, reconnect_in_progress_seen)
+        """
+        response_times: dict[str, float | None] = {}
+        session_bad_seen = False
+        session_bad_status: str | None = None
+        reconnect_in_progress_seen = False
+
+        for tag, tag_config in tags_config.items():
+            response_time, error_info = self.write_data(
+                server_id=server_id,
+                tag=tag,
+                data=data.head(1)[data_column].values[0],
+                data_type=tag_config['data_type'],
+                tag_type=tag_type,
+                metadata=metadata,
+            )
+            session_bad_seen, session_bad_status, reconnect_in_progress_seen = (
+                _apply_opc_write_error(
+                    error_info,
+                    session_bad_seen,
+                    session_bad_status,
+                    reconnect_in_progress_seen,
+                )
+            )
+            if response_time is not None:
+                self.info(
+                    f'{log_label} written to OPC server {server_id} for tag {tag}.',
+                    metadata,
+                )
+            response_times[tag] = response_time
+
+        return response_times, session_bad_seen, session_bad_status, reconnect_in_progress_seen
+
     def manage_output_tags(
         self,
         server_id: str,
         config: dict[str, Any],
         data: DataFrame,
         metadata: dict[str, Any],
-    ) -> tuple[bool, dict[str, float | None]]:
+    ) -> tuple[bool, dict[str, float | None], bool, str | None, bool]:
         """
         Write prediction and confidence values for one OPC server configuration.
 
-        The method iterates through optional ``prediction_tags`` and
-        ``confidence_tags``, performs synchronous writes for each tag, collects
-        per-tag response times, and returns an aggregate success flag
-        (all tags successful) with a metrics-friendly response map.
-
-        Args:
-         - server_id (str): Target OPC server id.
-         - config (dict[str, Any]): Server output configuration containing optional
-           ``prediction_tags`` and ``confidence_tags`` sections.
-         - data (DataFrame): Prediction dataframe used as source values.
-         - metadata (dict[str, Any]): Workflow metadata for logging/notifications.
-
         Return:
-         tuple[bool, dict[str, float | None]]: Global success flag and response-time
+         tuple: success flag, per-tag response times, session_bad flags.
-         map per tag (``None`` for failed writes).
         """
-
         response_times: dict[str, float | None] = {}
+        session_bad_seen = False
+        session_bad_status: str | None = None
+        reconnect_in_progress_seen = False
 
-        if 'prediction_tags' in config:
+        tag_groups = (
-            for tag, tag_config in config['prediction_tags'].items():
+            ('prediction_tags', 'prediction', 'prediction', 'Prediction data'),
-                response_time = self.write_data(
+            ('confidence_tags', 'prediction_confidence', 'confidence', 'Confidence data'),
-                    server_id=server_id,
+        )
-                    tag=tag,
+        for config_key, data_column, tag_type, log_label in tag_groups:
-                    data=data.head(1)['prediction'].values[0],
+            if config_key not in config:
-                    data_type=tag_config['data_type'],
+                continue
-                    tag_type='prediction',
+            (
-                    metadata=metadata,
+                group_times,
-                )
+                group_session_bad,
-                if response_time is not None:
+                group_status,
-                    self.info(
+                group_reconnect,
-                        f'Prediction data written to OPC server {server_id} for tag {tag}.',
+            ) = self._write_tags_from_config(
-                        metadata,
+                server_id=server_id,
-                    )
+                tags_config=config[config_key],
-                response_times[tag] = response_time
+                data=data,
-
+                data_column=data_column,
-        if 'confidence_tags' in config:
+                tag_type=tag_type,
-            for tag, tag_config in config['confidence_tags'].items():
+                log_label=log_label,
-                response_time = self.write_data(
+                metadata=metadata,
-                    server_id=server_id,
+            )
-                    tag=tag,
+            response_times.update(group_times)
-                    data=data.head(1)['prediction_confidence'].values[0],
+            if group_session_bad:
-                    data_type=tag_config['data_type'],
+                session_bad_seen = True
-                    tag_type='confidence',
+                session_bad_status = group_status or session_bad_status
-                    metadata=metadata,
+            if group_reconnect:
-                )
+                reconnect_in_progress_seen = True
-                if response_time is not None:
-                    self.info(
-                        f'Confidence data written to OPC server {server_id} for tag {tag}.',
-                        metadata,
-                    )
-                response_times[tag] = response_time
 
         success = None not in response_times.values()
-
+        return (
-        return success, response_times
+            success,
+            response_times,
+            session_bad_seen,
+            session_bad_status,
+            reconnect_in_progress_seen,
+        )
 
     @activity.defn(name='write_opc_data')
     def write_opc_data(
@@ -246,19 +320,13 @@ class OPC(SientiaMonitoring):
         """
         Execute OPC writes across all configured servers and collect per-tag metrics.
 
-        For each server in ``opc_output_config``, this activity validates server
-        availability, writes enabled prediction/confidence tags, accumulates
-        response-time metrics, and then normalizes confidence/comments in the
-        returned prediction payload when at least one write fails.
-
         Args:
          - input_data (dict[str, Any]): Payload containing workflow metadata, data
            to write, and ``opc_output_config`` server/tag definitions.
 
         Return:
          tuple[dict[Hashable, Any], dict[str, dict[str, float | None]]]: Updated
-         prediction payload dict and nested metrics
+         prediction payload dict and nested metrics per server/tag.
-         ``{server_id: {tag_name: response_time_or_none}}``.
         """
         metadata = input_data['metadata']
         self.info('Writing data to OPC servers...', metadata)
@@ -267,20 +335,32 @@ class OPC(SientiaMonitoring):
         self.info(f'Data to write: {data.size} rows', metadata)
 
         success = True
+        session_bad_seen = False
+        session_bad_status: str | None = None
+        reconnect_in_progress_seen = False
 
-        metrics: dict[str, dict[str, float | None]] = {}
+        opc_metrics: dict[str, dict[str, float | None]] = {}
 
         for server_id, config in opc_output_config.items():
             if not self.validate_server(server_id, metadata):
                 success = False
                 continue
 
-            local_success, local_response_times = self.manage_output_tags(
+            (
-                server_id, config, data, metadata
+                local_success,
-            )
+                local_response_times,
-            metrics[server_id] = local_response_times
+                local_session_bad,
+                local_status,
+                local_reconnect_in_progress,
+            ) = self.manage_output_tags(server_id, config, data, metadata)
+            opc_metrics[server_id] = local_response_times
             local_count = len(local_response_times)
             success = success and local_success
+            if local_session_bad:
+                session_bad_seen = True
+                session_bad_status = local_status or session_bad_status
+            if local_reconnect_in_progress:
+                reconnect_in_progress_seen = True
 
             n_pred = len(config.get('prediction_tags') or {})
             n_conf = len(config.get('confidence_tags') or {})
@@ -289,34 +369,55 @@ class OPC(SientiaMonitoring):
                 metadata,
             )
 
-        return self.process_confidence(data, success, metadata), metrics
+        return (
+            self.process_confidence(
+                data,
+                success,
+                metadata,
+                session_bad=session_bad_seen,
+                opc_status=session_bad_status,
+                reconnect_in_progress=reconnect_in_progress_seen,
+            ),
+            opc_metrics,
+        )
 
     def process_confidence(
-        self, data: DataFrame, success: bool, metadata: dict[str, Any]
+        self,
+        data: DataFrame,
+        success: bool,
+        metadata: dict[str, Any],
+        *,
+        session_bad: bool = False,
+        opc_status: str | None = None,
+        reconnect_in_progress: bool = False,
     ) -> dict[Hashable, Any]:
         """
         Apply fallback confidence/comment values when OPC writes are not fully successful.
 
-        Args:
-         - data (DataFrame): Prediction dataframe to be returned to downstream steps.
-         - success (bool): Aggregate write status across all attempted OPC tags.
-         - metadata (dict[str, Any]): Workflow metadata used for debug logs.
-
         Return:
-         dict[Hashable, Any]: Serialized dataframe dict with original values on success,
+         dict[Hashable, Any]: Serialized dataframe dict with updated confidence/comments on failure.
-         or downgraded confidence/comment fields on failure.
         """
 
-        message = 'Some data could not be written to OPC servers'
-
         if not success:
-            data['prediction_confidence'] = OPC_WRITTING_ERROR_CONFIDENCE
+            comment_parts: list[str] = []
-            data['comments'] = message
+            confidence = OPC_WRITTING_ERROR_CONFIDENCE
+
+            if session_bad:
+                comment_parts.append(_opc_session_bad_comment(opc_status))
+                confidence = OPC_SESSION_BAD_CONFIDENCE
+            if reconnect_in_progress:
+                comment_parts.append(OPC_RECONNECT_IN_PROGRESS_COMMENT)
+                confidence = OPC_SESSION_BAD_CONFIDENCE
+            if not comment_parts:
+                comment_parts.append(OPC_WRITTING_ERROR_MESSAGE)
+
+            comments = OPC_COMMENT_SEPARATOR.join(comment_parts)
+            data['prediction_confidence'] = confidence
+            data['comments'] = comments
             self.debug(
-                f'{message}, setting confidence to  {OPC_WRITTING_ERROR_CONFIDENCE}.',
+                f'OPC write issues, confidence={confidence}, comments={comments}',
                 metadata,
             )
-
         else:
             self.debug('Data written to OPC servers successfully.', metadata)
 
@@ -325,9 +426,6 @@ class OPC(SientiaMonitoring):
     def close(self) -> None:
         """
         Disconnect all tracked OPC repositories and clear in-memory references.
-
-        This method should be called during worker shutdown to ensure every
-        synchronous OPC session is explicitly closed before process exit.
         """
 
         for opc in self.opc_repository.values():

laborious/metrics.py

@@ -89,6 +89,40 @@ OPC_CONNECTION_STATUS = Gauge(
     ['pod_id', 'server_name', 'server_url'],
 )
 
+_OPC_SESSION_DEBUG_LABELS = ['pod_id', 'server_name', 'runtime', 'opc_server_id', 'session_id']
+
+OPC_SESSION_CREATED_TOTAL = Counter(
+    'opc_session_created_total',
+    'OPC UA sessions established (after successful connect)',
+    _OPC_SESSION_DEBUG_LABELS,
+)
+
+OPC_SESSION_CLOSED_TOTAL = Counter(
+    'opc_session_closed_total',
+    'OPC UA client disconnects completed (session tear-down initiated)',
+    _OPC_SESSION_DEBUG_LABELS,
+)
+
+OPC_SESSION_REVISED_TIMEOUT_MS = Gauge(
+    'opc_session_revised_timeout_milliseconds',
+    'Server-revised OPC UA session timeout (RevisedSessionTimeout) in ms after connect',
+    _OPC_SESSION_DEBUG_LABELS,
+)
+
+OPC_WRITE_ATTEMPT_LABELS = [*_OPC_SESSION_DEBUG_LABELS, 'model_id', 'model_name', 'result']
+
+OPC_WRITE_ATTEMPTS_TOTAL = Counter(
+    'opc_write_attempts_total',
+    'OPC UA write attempts with session and outcome (result=OK or exception class name)',
+    OPC_WRITE_ATTEMPT_LABELS,
+)
+
+OPC_WRITE_INTER_ARRIVAL_OVER_SESSION_TIMEOUT_TOTAL = Counter(
+    'opc_write_inter_arrival_over_session_timeout_total',
+    'Successful writes where seconds since the previous successful write exceeded RevisedSessionTimeout (ms)',
+    _OPC_SESSION_DEBUG_LABELS,
+)
+
 # ================== Model metrics ==================
 
 MODEL_READ_LAG = Histogram(

laborious/utils/connectors_config.py

@@ -3,8 +3,6 @@ from os import getenv
 from typing import Any
 
 
-
-
 def build_mlflow_config() -> dict[str, Any]:
     """
     Read MLflow tracking and registry credentials from the environment.

laborious/utils/repository/opc_repository.py

@@ -1,12 +1,14 @@
 """
-Synchronous OPC UA client repository using asyncua (opcua-asyncio) ``sync`` API.
+Synchronous OPC UA client repository using asyncua ``sync`` API.
 
-``asyncua.sync.Client`` runs the asyncio client on a background thread so callers
+``asyncua.sync.Client`` runs the asyncio stack on a background thread so Temporal
-stay synchronous. Connect/disconnect, optional Basic256 security, session checks,
+activities and other callers stay blocking while preserving the same session
-and typed writes mirror the previous python-opcua integration.
+lifecycle, security policy, reconnect semantics, and write error classification
+as the async ``origin/main`` implementation.
 """
 
 import json
+import threading
 import time
 import traceback
 from datetime import datetime
@@ -16,6 +18,7 @@ from typing import Any
 from asyncua import ua
 from asyncua.crypto import security_policies
 from asyncua.sync import Client
+from asyncua.ua.uaerrors import UaStatusCodeError
 from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
 from sientia_do.notifications.models import NotificationLevel
 from sientia_do.observability.logger import Logger
@@ -24,6 +27,119 @@ from sientia_do.observability.sientia_monitoring import SientiaMonitoring
 
 from laborious import metrics
 
+# Requested session and secure channel lifetime (ms) before server revision; 10 minutes.
+OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS = 10 * 60 * 1000
+
+
+class OpcClientAlreadyExistsError(RuntimeError):
+    """Raised when _create_client is called while self.client is already set."""
+
+
+class OpcSessionAlreadyConnectedError(RuntimeError):
+    """Raised when _open_session is called while a UA session is already open."""
+
+
+class OpcClientNotInitializedError(RuntimeError):
+    """Raised when _open_session is called before _create_client."""
+
+
+RECONNECTABLE_OPC_BAD_NAMES: frozenset[str] = frozenset(
+    {
+        'BadSessionIdInvalid',
+        'BadSessionClosed',
+        'BadSessionNotActivated',
+        'BadSecureChannelIdInvalid',
+        'BadSecureChannelClosed',
+        'BadSecureChannelTokenUnknown',
+        'BadTcpSecureChannelUnknown',
+        'BadServerNotConnected',
+        'BadConnectionClosed',
+        'BadDisconnect',
+        'BadConnectionRejected',
+        'BadCommunicationError',
+        'BadRequestInterrupted',
+        'BadUnknownResponse',
+        'BadTimeout',
+        'BadRequestTimeout',
+        'BadSequenceNumberInvalid',
+        'BadSequenceNumberUnknown',
+        'BadSecurityModeInsufficient',
+        'BadRequestHeaderInvalid',
+        'BadInvalidState',
+    }
+)
+
+
+def _opc_authentication_token_str(client: Client | None) -> str:
+    """
+    Serialize the current OPC UA authentication token (session handle) for logging and metrics.
+
+    Return:
+        str: Token string, or "unknown" if unavailable.
+    """
+    if client is None:
+        return 'unknown'
+    try:
+        proto = client.aio_obj.uaclient.protocol
+        if proto is None:
+            return 'unknown'
+        tok = getattr(proto, 'authentication_token', None)
+        if tok is None:
+            return 'unknown'
+        return str(tok)
+    except Exception:
+        return 'unknown'
+
+
+def _opc_status_from_exception(exc: BaseException) -> str:
+    """
+    Resolve OPC UA status name from an exception, including chained UaStatusCodeError causes.
+
+    Args:
+        exc (BaseException): Raised error from asyncua.
+
+    Return:
+        str: Status class name or generic Python exception name.
+    """
+    current: BaseException | None = exc
+    while current is not None:
+        if isinstance(current, UaStatusCodeError):
+            return type(current).__name__
+        current = current.__cause__
+    return type(exc).__name__
+
+
+def is_reconnectable_opcua_bad(exc: BaseException) -> bool:
+    """
+    Return whether the exception is a Tier-1 OPC UA Bad* that should trigger reconnect.
+
+    Args:
+        exc (BaseException): Raised error from get_node or write_value.
+
+    Return:
+        bool: True if reconnect should be scheduled.
+    """
+    return _opc_status_from_exception(exc) in RECONNECTABLE_OPC_BAD_NAMES
+
+
+def _model_labels_from_write_metadata(metadata: dict[str, Any] | None) -> dict[str, str]:
+    """
+    Extract model_id and model_name from write metadata for Prometheus labels.
+
+    Args:
+        metadata (dict[str, Any] | None): Context passed into write_data; may omit keys.
+
+    Return:
+        dict[str, str]: Labels model_id and model_name, defaulting to "unknown".
+    """
+    if not metadata:
+        return {'model_id': 'unknown', 'model_name': 'unknown'}
+    return {
+        'model_id': str(metadata.get('model_id', 'unknown')),
+        'model_name': str(metadata.get('model_name', 'unknown')),
+    }
+
+
 data_type_map = {
     'float': {
         'converter': float,
@@ -52,11 +168,8 @@ class OpcRepository(SientiaMonitoring):
     """
     Synchronous OPC UA repository for connect/disconnect and typed writes.
 
-    Attributes:
+    Uses ``asyncua.sync.Client`` with the same session metrics, Tier-1 Bad* reconnect,
-        url: OPC UA endpoint URL.
+    and structured write error payloads as the async repository on ``origin/main``.
-        id: Server identifier used in metrics and notifications.
-        server_name: Human-readable server name for labels.
-        client: Active ``asyncua.sync.Client`` while connected.
     """
 
     def __init__(
@@ -80,10 +193,10 @@ class OpcRepository(SientiaMonitoring):
         self.cert_path = cert_path
         self.private_key_path = private_key_path
         self.server_cert_path = server_cert_path
-        self.error_count = 0
         self.reconnection_interval = reconnection_interval
         self.last_reconnection_time: datetime | None = None
         self.disconnection_interval = 10.0
+        self.notification_handler = notification_handler
         self.client: Client | None = None
 
         SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
@@ -94,15 +207,63 @@ class OpcRepository(SientiaMonitoring):
             'workflow_name': 'opc_repository',
             'schedule_name': '-',
         }
+        self._last_write_mono: float | None = None
+        self._connection_lock = threading.Lock()
+        self._session_ready = threading.Event()
+        self._reconnect_thread: threading.Thread | None = None
+
+    def _opc_debug_tags(self, session_id: str) -> dict[str, str]:
+        return {
+            'pod_id': str(getattr(self, 'pod_id', 'unknown')),
+            'server_name': self.server_name,
+            'runtime': str(getattr(self, 'runtime', 'unknown')),
+            'opc_server_id': self.id,
+            'session_id': session_id,
+        }
+
+    def _is_session_open(self) -> bool:
+        """
+        Return whether the asyncua client has an open transport session.
+
+        Return:
+            bool: True when protocol exists and is not closed.
+        """
+        if self.client is None:
+            return False
+        try:
+            proto = self.client.aio_obj.uaclient.protocol
+            return proto is not None and proto.state != 'closed'
+        except Exception:
+            return False
+
+    def _reconnection_window_elapsed(self) -> bool:
+        """
+        Return whether enough time has passed since the last reconnect attempt.
+
+        Return:
+            bool: True if a new reconnect is allowed.
+        """
+        if self.last_reconnection_time is None:
+            return True
+        return (
+            datetime.now() - self.last_reconnection_time
+        ).total_seconds() > self.reconnection_interval
+
+    def _not_connected_error(self) -> dict[str, Any]:
+        return {
+            'notification_id': f'OPC_CONNECTION_NOT_READY_{self.id}',
+            'message': f'OPC server {self.id} is not connected',
+            'block': 'opc_repository',
+            'level': NotificationLevel.WARNING,
+        }
 
     def set_security(self) -> None:
         """
-        Configure Basic256 security policy, certificates, and long channel/session timeouts.
+        Configure certificates and timeouts on the sync asyncua client.
 
         Raises:
-            ValueError: If certificate paths are missing or client is not initialized.
+            ValueError: If cert paths or client are missing.
         """
-
         if self.cert_path is None or self.private_key_path is None:
             raise ValueError(
                 'Certificate and private key paths must be provided for secure connection.'
@@ -115,7 +276,7 @@ class OpcRepository(SientiaMonitoring):
         if self.client is None:
             raise ValueError('Client must be initialized before setting security')
 
-        self.client.application_uri = self.server_uri or self.client.application_uri
+        self.client.application_uri = self.server_uri
         self.info('Setting security...', self.metadata)
         self.client.set_security(
             security_policies.SecurityPolicyBasic256,
@@ -124,75 +285,105 @@ class OpcRepository(SientiaMonitoring):
             None,
             str(server_cert) if server_cert else None,
         )
-        self.client.aio_obj.secure_channel_timeout = 10000000
+        aio = self.client.aio_obj
-        self.client.aio_obj.session_timeout = 10000000
+        aio.secure_channel_timeout = OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS
+        aio.session_timeout = OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS
 
-    def connect(self) -> tuple[bool, dict[str, Any]]:
+    def _create_client(self) -> None:
         """
-        Create the synchronous client, optionally apply security, and connect to the server.
+        Instantiate the sync Client and apply security when configured.
 
-        Return:
+        Caller must hold _connection_lock. Does not open a UA session.
-            tuple[bool, dict[str, Any]]: Success flag and error payload when False.
+
+        Raises:
+            OpcClientAlreadyExistsError: If self.client is already set.
         """
+        if self.client is not None:
+            raise OpcClientAlreadyExistsError(
+                f'OPC client already exists for server {self.id}; '
+                'call disconnect() before creating a new client'
+            )
 
         self.client = Client(self.url, timeout=10)
-
+        aio = self.client.aio_obj
-        self.client.aio_obj.name = self.pod_id
+        if hasattr(aio, 'watchdog_intervall'):
-        self.client.aio_obj.description = self.pod_id
+            aio.watchdog_intervall = 50
+        aio.name = self.pod_id
+        aio.description = self.pod_id
         pod_uri = self.pod_id.replace('-', ':')
         self.client.application_uri = pod_uri
-        self.client.aio_obj.product_uri = pod_uri
+        aio.product_uri = pod_uri
-
         if self.cert_path:
             self.set_security()
-        self.info(
-            f'Starting connection to OPC server {self.id}:{self.server_name}...', self.metadata
-        )
-        return self.try_connect()
 
-    def try_connect(self) -> tuple[bool, dict[str, Any]]:
+    def _open_session(self) -> tuple[bool, dict[str, Any]]:
         """
-        Perform the TCP/session handshake and emit connection metrics.
+        Open the OPC UA session on the existing client.
+
+        Caller must hold _connection_lock.
+
+        Raises:
+            OpcClientNotInitializedError: If self.client is None.
+            OpcSessionAlreadyConnectedError: If a session is already open.
 
         Return:
-            tuple[bool, dict[str, Any]]: Success flag and structured error when False.
+            tuple[bool, dict[str, Any]]: Success flag and error payload on connect failure.
         """
+        if self.client is None:
+            raise OpcClientNotInitializedError(
+                f'OPC client is not initialized for server {self.id}; '
+                'call _create_client() before opening a session'
+            )
+        if self._is_session_open():
+            raise OpcSessionAlreadyConnectedError(
+                f'OPC session already connected for server {self.id}; '
+                'call disconnect() before connecting again'
+            )
 
         tags = {
             'pod_id': self.pod_id,
             'server_name': self.server_name,
         }
         self.emit_metric_sync(metrics.OPC_CONNECTIONS_TOTAL, tags)
+
         try:
-            self.last_reconnection_time = datetime.now()
-            if self.client is None:
-                return False, {
-                    'notification_id': f'OPC_CONNECTION_ERROR_{self.id}',
-                    'message': 'Client is not initialized',
-                    'block': 'opc_repository',
-                    'level': NotificationLevel.ERROR,
-                }
             self.client.connect()
 
+            aio = self.client.aio_obj
+            session_id = _opc_authentication_token_str(self.client)
+            revised_session_timeout_ms = int(aio.session_timeout)
+            revised_secure_channel_timeout_ms = int(aio.secure_channel_timeout)
+            self.info(
+                f'OPC new session connected opc_server_id={self.id} session_id={session_id} '
+                f'revised_session_timeout_ms={revised_session_timeout_ms} '
+                f'revised_secure_channel_timeout_ms={revised_secure_channel_timeout_ms}',
+                self.metadata,
+            )
+            self.emit_metric_sync(
+                metrics.OPC_SESSION_CREATED_TOTAL, self._opc_debug_tags(session_id)
+            )
+            self.emit_metric_sync(
+                metric_object=metrics.OPC_SESSION_REVISED_TIMEOUT_MS,
+                method='set',
+                tags=self._opc_debug_tags(session_id),
+                value=revised_session_timeout_ms,
+            )
             self.emit_metric_sync(
                 metric_object=metrics.OPC_CONNECTION_STATUS,
                 method='set',
-                tags={
+                tags={**tags, 'server_url': self.url},
-                    **tags,
-                    'server_url': self.url,
-                },
                 value=1,
             )
 
+            self._last_write_mono = None
+            self._session_ready.set()
             return True, {}
-        except Exception as e:
-            self.disconnect()
 
+        except Exception as e:
+            self._disconnect_locked()
             trace = traceback.format_exc()
             self.error(trace, self.metadata)
-
             self.emit_metric_sync(metrics.OPC_CONNECTIONS_FAILED, tags)
-
             return False, {
                 'notification_id': f'OPC_CONNECTION_ERROR_{self.id}',
                 'message': f'Failed to connect to OPC server: {e}',
@@ -201,20 +392,38 @@ class OpcRepository(SientiaMonitoring):
                 'attachment_content': trace,
             }
 
-    def disconnection_fallback(self) -> list[dict[str, Any]]:
+    def _connect_locked(self) -> tuple[bool, dict[str, Any]]:
         """
-        Retry disconnect up to five times with linear backoff.
+        Create the client when absent, then open a UA session.
+
+        Caller must hold _connection_lock.
+
+        Raises:
+            OpcSessionAlreadyConnectedError: If a session is already open.
 
         Return:
-            list[dict[str, Any]]: Empty on success, otherwise error records per attempt.
+            tuple[bool, dict[str, Any]]: Result from _open_session on connect failure.
         """
+        if self._is_session_open():
+            raise OpcSessionAlreadyConnectedError(
+                f'OPC session already connected for server {self.id}; '
+                'call disconnect() before connecting again'
+            )
+        if self.client is None:
+            self._create_client()
+        return self._open_session()
 
+    def _disconnection_fallback(self) -> list[dict[str, Any]]:
+        """
+        Try up to five times to disconnect from the OPC UA server.
+        """
         assert self.client is not None
-        error_stack = []
+        error_stack: list[dict[str, Any]] = []
         for i in range(5):
             try:
                 self.info(
-                    f'Disconnecting from OPC UA server, attempt {i + 1} of 5', self.metadata
+                    f'Disconnecting from OPC UA server, attempt {i + 1} of 5',
+                    self.metadata,
                 )
                 self.client.disconnect()
                 return []
@@ -233,15 +442,26 @@ class OpcRepository(SientiaMonitoring):
                 time.sleep(self.disconnection_interval * i)
         return error_stack
 
-    def disconnect(self) -> None:
+    def _disconnect_locked(self) -> None:
         """
-        Tear down the UA session and reset connection metrics.
+        Tear down the current session and client.
+
+        Caller must hold _connection_lock.
         """
+        self._last_write_mono = None
+        self._session_ready.clear()
 
         if self.client is None:
             return
 
-        errors = self.disconnection_fallback()
+        session_id = _opc_authentication_token_str(self.client)
+        self.info(
+            f'OPC disconnecting opc_server_id={self.id} session_id={session_id}',
+            self.metadata,
+        )
+        self.emit_metric_sync(metrics.OPC_SESSION_CLOSED_TOTAL, self._opc_debug_tags(session_id))
+
+        errors = self._disconnection_fallback()
         if errors:
             self.send_notification(
                 metadata=self.metadata,
@@ -252,9 +472,8 @@ class OpcRepository(SientiaMonitoring):
                 attachment_content=json.dumps(errors, indent=4),
             )
         else:
-            self.warning(
+            self.warning(f'Disconnected from OPC server {self.id} successfully', self.metadata)
-                f'Disconnected from OPC server {self.id} successfully', self.metadata
+
-            )
         self.emit_metric_sync(
             metric_object=metrics.OPC_CONNECTION_STATUS,
             method='set',
@@ -265,139 +484,313 @@ class OpcRepository(SientiaMonitoring):
             },
             value=0,
         )
-
         self.client = None
 
-    def _session_alive(self) -> bool:
+    def _reconnect_locked(self) -> tuple[bool, dict[str, Any]]:
         """
-        Best-effort check that the synchronous client still has a working session.
+        Close the current session and open a new one.
+
+        Caller must hold _connection_lock. Records last_reconnection_time for interval gating.
 
         Return:
-            bool: True if a root browse succeeds, False otherwise.
+            tuple[bool, dict[str, Any]]: Result from _connect_locked after teardown.
         """
+        self.last_reconnection_time = datetime.now()
+        self._disconnect_locked()
+        return self._connect_locked()
 
-        if self.client is None:
+    def connect(self) -> tuple[bool, dict[str, Any]]:
-            return False
+        """
-        try:
+        Open an OPC UA session under the connection lock (worker initialization).
-            self.client.get_root_node()
+        """
-            return True
+        with self._connection_lock:
-        except Exception:
+            self.info(
-            return False
+                f'Starting connection to OPC server {self.id}:{self.server_name}...',
+                self.metadata,
+            )
+            return self._connect_locked()
+
+    def disconnect(self) -> None:
+        """
+        Gracefully disconnect from the OPC server under the connection lock.
+        """
+        with self._connection_lock:
+            self._disconnect_locked()
 
     def validate_connection(self) -> tuple[bool, dict[str, Any]]:
         """
-        Ensure the UA session is usable; reconnect when outside the backoff window.
+        Read-only check that the asyncua protocol is open.
+
+        Caller must ensure _session_ready before writing. Does not connect or reconnect.
 
         Return:
-            tuple[bool, dict[str, Any]]: Whether the session is ready and optional error payload.
+            tuple[bool, dict[str, Any]]: (True, {}) when open, otherwise (False, error).
         """
-
+        if self._is_session_open():
-        if self.client is None:
-            return self.connect()
-
-        try:
-            if not self._session_alive():
-                self.error(f'OPC server {self.id} is not connected', self.metadata)
-                if (
-                    self.last_reconnection_time is None
-                    or (datetime.now() - self.last_reconnection_time).total_seconds()
-                    > self.reconnection_interval
-                ):
-                    self.disconnect()
-                    self.info(
-                        f'Trying to reconnect to OPC server {self.id}...', self.metadata
-                    )
-                    return self.connect()
-
-                return False, {
-                    'notification_id': f'OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{self.id}',
-                    'message': (
-                        f'OPC server {self.id} is not connected, waiting for next reconnection window...'
-                    ),
-                    'block': 'opc_repository',
-                    'level': NotificationLevel.WARNING,
-                }
             return True, {}
-        except Exception as e:
+        self.error(f'OPC server {self.id} is not connected', self.metadata)
-            trace = traceback.format_exc()
+        return False, self._not_connected_error()
-            message = f'Failed to validate connection to OPC server: {e}'
+
-            self.error(message, self.metadata)
+    def _start_reconnect_on_bad(self, opc_status: str, session_id: str) -> None:
-            return False, {
+        """
-                'notification_id': f'OPC_CONNECTION_CHECK_ERROR_{self.id}',
+        Schedule a background reconnect if interval and thread state allow it.
-                'message': message,
+
-                'block': 'opc_repository',
+        Args:
-                'level': NotificationLevel.ERROR,
+            opc_status (str): OPC UA status name that triggered reconnect.
-                'attachment_content': trace,
+            session_id (str): Session token before failure.
-            }
+        """
+        if not self._reconnection_window_elapsed():
+            self.warning(
+                f'OPC reconnect skipped reason=reconnection_window opc_server_id={self.id} '
+                f'opc_status={opc_status}',
+                self.metadata,
+            )
+            return
+        if self._reconnect_thread is not None and self._reconnect_thread.is_alive():
+            self.warning(
+                f'OPC reconnect skipped reason=in_progress opc_server_id={self.id} '
+                f'opc_status={opc_status}',
+                self.metadata,
+            )
+            return
+
+        self._session_ready.clear()
+        self.info(
+            f'OPC reconnect scheduled after opc_status={opc_status} opc_server_id={self.id} '
+            f'old_session_id={session_id}',
+            self.metadata,
+        )
+        self._reconnect_thread = threading.Thread(
+            target=self._run_reconnect_on_bad,
+            args=(opc_status, session_id),
+            daemon=True,
+        )
+        self._reconnect_thread.start()
+
+    def _run_reconnect_on_bad(self, opc_status: str, session_id: str) -> None:
+        """
+        Tear down and re-establish the OPC UA session under the connection lock.
+
+        Args:
+            opc_status (str): OPC UA status that triggered reconnect.
+            session_id (str): Previous session token string.
+        """
+        try:
+            with self._connection_lock:
+                self.info(
+                    f'OPC reconnect started opc_status={opc_status} opc_server_id={self.id} '
+                    f'old_session_id={session_id}',
+                    self.metadata,
+                )
+                self._reconnect_locked()
+        except Exception:
+            self.error(
+                f'OPC reconnect task failed opc_server_id={self.id} opc_status={opc_status}',
+                self.metadata,
+            )
+            self.error(traceback.format_exc(), self.metadata)
+
+    def _log_write_inter_arrival(self, session_id: str, node: str) -> None:
+        """
+        Log elapsed wall time since the previous successful OPC write on this repository.
+
+        Args:
+            session_id (str): Current OPC UA session token string.
+            node (str): Node id written in this operation.
+        """
+        now = time.monotonic()
+        if self._last_write_mono is not None:
+            delta_s = now - self._last_write_mono
+            self.info(
+                f'OPC write inter-arrival_s={delta_s:.6f} opc_server_id={self.id} '
+                f'session_id={session_id} node={node}',
+                self.metadata,
+            )
+            if self.client is not None:
+                session_timeout_ms = float(self.client.aio_obj.session_timeout)
+                if session_timeout_ms > 0 and delta_s > (session_timeout_ms / 1000.0):
+                    self.emit_metric_sync(
+                        metrics.OPC_WRITE_INTER_ARRIVAL_OVER_SESSION_TIMEOUT_TOTAL,
+                        self._opc_debug_tags(session_id),
+                    )
+        self._last_write_mono = now
+
+    def _emit_opc_write_metric(
+        self, session_id: str, result: str, metadata: dict[str, Any] | None
+    ) -> None:
+        self.emit_metric_sync(
+            metrics.OPC_WRITE_ATTEMPTS_TOTAL,
+            {
+                **self._opc_debug_tags(session_id),
+                **_model_labels_from_write_metadata(metadata),
+                'result': result,
+            },
+        )
+
+    def _write_failure_payload(
+        self,
+        notification_id: str,
+        message: str,
+        level: NotificationLevel = NotificationLevel.ERROR,
+        attachment_content: str | None = None,
+        opc_error_kind: str | None = None,
+        opc_status: str | None = None,
+    ) -> dict[str, Any]:
+        payload: dict[str, Any] = {
+            'notification_id': notification_id,
+            'message': message,
+            'block': 'opc_repository',
+            'level': level,
+        }
+        if attachment_content is not None:
+            payload['attachment_content'] = attachment_content
+        if opc_error_kind is not None:
+            payload['opc_error_kind'] = opc_error_kind
+        if opc_status is not None:
+            payload['opc_status'] = opc_status
+        return payload
+
+    def _handle_tier1_bad(
+        self,
+        exc: BaseException,
+        session_id: str,
+        node: str,
+        metadata: dict[str, Any],
+        phase: str,
+    ) -> tuple[bool, dict[str, Any]]:
+        """
+        Record metrics/logs and schedule reconnect after a Tier-1 Bad* error.
+
+        Args:
+            exc (BaseException): Tier-1 OPC UA error.
+            session_id (str): Session token at failure time.
+            node (str): Node id being written.
+            metadata (dict[str, Any]): Write context.
+            phase (str): get_node or write_value.
+
+        Return:
+            tuple[bool, dict[str, Any]]: Always (False, error payload).
+        """
+        opc_status = _opc_status_from_exception(exc)
+        trace = traceback.format_exc()
+        self.error(trace, metadata)
+        self._emit_opc_write_metric(session_id, opc_status, metadata)
+        self.error(
+            f'OPC write failed opc_status={opc_status} opc_server_id={self.id} '
+            f'session_id={session_id} model_id={metadata.get("model_id", "unknown")} '
+            f'model_name={metadata.get("model_name", "unknown")} node={node} phase={phase}',
+            metadata,
+        )
+        self._start_reconnect_on_bad(opc_status, session_id)
+        return False, self._write_failure_payload(
+            notification_id=f'OPC_WRITE_DATA_ERROR_{self.id}',
+            message=f'Failed to {phase} on OPC server: {exc} | metadata: {metadata}',
+            attachment_content=trace,
+            opc_error_kind='session_bad',
+            opc_status=opc_status,
+        )
+
+    @staticmethod
+    def _write_node_value(
+        node_obj: Any,
+        ua_data: ua.DataValue,
+        data: Any,
+        variant_type: ua.VariantType,
+    ) -> None:
+        """
+        Write a DataValue to a node, falling back to set_value when write_value is unavailable.
+
+        Args:
+            node_obj: Sync or async node wrapper from asyncua.
+            ua_data (ua.DataValue): Encoded value for write_value.
+            data: Scalar converted value for set_value fallback.
+            variant_type (ua.VariantType): OPC UA type for set_value fallback.
+        """
+        if hasattr(node_obj, 'write_value'):
+            try:
+                node_obj.write_value(ua_data)
+                return
+            except (AttributeError, TypeError):
+                pass
+        node_obj.set_value(data, variant_type)
 
     def write_data(
         self, node: str, value: Any, data_type: str, metadata: dict[str, Any]
     ) -> tuple[bool, dict[str, Any]]:
         """
-        Write a typed value to an OPC UA node after validating connectivity.
+        Write data to OPC server with a single attempt and Tier-1 Bad* reconnect scheduling.
-
-        Args:
-            node: Node id string accepted by ``Client.get_node``.
-            value: Scalar value to encode.
-            data_type: Key into ``data_type_map`` (e.g. float, str).
-            metadata: Workflow metadata for error context.
-
-        Return:
-            tuple[bool, dict[str, Any]]: Success flag and either ``response_time`` or error fields.
         """
+        if not self._session_ready.is_set():
+            self._emit_opc_write_metric('unknown', 'ReconnectInProgress', metadata)
+            self.warning(
+                f'OPC write rejected reconnect_in_progress opc_server_id={self.id} '
+                f'model_id={metadata.get("model_id", "unknown")} '
+                f'model_name={metadata.get("model_name", "unknown")}',
+                metadata,
+            )
+            return False, {
+                'notification_id': f'OPC_WRITE_RECONNECT_IN_PROGRESS_{self.id}',
+                'message': f'OPC write skipped: reconnect in progress | metadata: {metadata}',
+                'block': 'opc_repository',
+                'level': NotificationLevel.WARNING,
+                'opc_error_kind': 'reconnect_in_progress',
+            }
 
         is_connected, error = self.validate_connection()
-
         if not is_connected:
+            self._emit_opc_write_metric('unknown', 'NotConnected', metadata)
             return False, error
 
         start_time = time.time()
+        session_id = _opc_authentication_token_str(self.client)
 
         try:
             assert self.client is not None
             node_obj = self.client.get_node(node)
         except Exception as e:
+            if is_reconnectable_opcua_bad(e):
+                return self._handle_tier1_bad(e, session_id, node, metadata, 'get_node')
             trace = traceback.format_exc()
             self.error(trace, metadata)
-            self.error_count += 1
+            self._emit_opc_write_metric(session_id, f'GetNodeError:{type(e).__name__}', metadata)
-            return False, {
+            return False, self._write_failure_payload(
-                'notification_id': f'OPC_WRITE_GET_NODE_ERROR_{self.id}',
+                notification_id=f'OPC_WRITE_GET_NODE_ERROR_{self.id}',
-                'message': f'Failed to get node from OPC server: {e} | metadata: {metadata}',
+                message=f'Failed to get node from OPC server: {e} | metadata: {metadata}',
-                'block': 'opc_repository',
+                attachment_content=trace,
-                'level': NotificationLevel.ERROR,
+            )
-                'attachment_content': trace,
-            }
 
         if data_type not in data_type_map:
-            return False, {
+            self._emit_opc_write_metric(session_id, 'UnsupportedDataType', metadata)
-                'notification_id': f'OPC_WRITE_DATA_TYPE_ERROR_{self.id}',
+            return False, self._write_failure_payload(
-                'message': f'Unsupported data type: {data_type} | metadata: {metadata}',
+                notification_id=f'OPC_WRITE_DATA_TYPE_ERROR_{self.id}',
-                'block': 'opc_repository',
+                message=f'Unsupported data type: {data_type} | metadata: {metadata}',
-                'level': NotificationLevel.ERROR,
+            )
-            }
 
         data = data_type_map[data_type]['converter'](value)
         self.info(f'Writing {data} - {type(data)} to {node}', metadata)
         variant_type = data_type_map[data_type]['opc_type']
+        ua_data = ua.DataValue(
+            ua.Variant(data, variant_type),
+        )
 
         try:
-            node_obj.set_value(data, variant_type)
+            self._write_node_value(node_obj, ua_data, data, variant_type)
-
             end_time = time.time()
             response_time = end_time - start_time
-
         except Exception as e:
+            if is_reconnectable_opcua_bad(e):
+                return self._handle_tier1_bad(e, session_id, node, metadata, 'write_value')
             trace = traceback.format_exc()
             self.error(trace, metadata)
-            self.error_count += 1
+            self._emit_opc_write_metric(session_id, type(e).__name__, metadata)
-            return False, {
+            return False, self._write_failure_payload(
-                'notification_id': f'OPC_WRITE_DATA_ERROR_{self.id}',
+                notification_id=f'OPC_WRITE_DATA_ERROR_{self.id}',
-                'message': f'Failed to write data to OPC server: {e} | metadata: {metadata}',
+                message=f'Failed to write data to OPC server: {e} | metadata: {metadata}',
-                'block': 'opc_repository',
+                attachment_content=trace,
-                'level': NotificationLevel.ERROR,
+            )
-                'attachment_content': trace,
+
-            }
+        self._emit_opc_write_metric(session_id, 'OK', metadata)
-        self.error_count = 0
+        self._log_write_inter_arrival(session_id, node)
 
         return True, {
             'response_time': response_time,

pyproject.toml

@@ -121,6 +121,7 @@ markers = [
     "asyncio: marks tests as async",
     "integration: marks tests as integration tests",
     "unit: marks tests as unit tests",
+    "opc: marks E2E tests that use in-process asyncua + real OpcRepository",
 ]
 
 [tool.coverage.run]

tests/laborious/activities/test_mlflow.py

@@ -377,6 +377,7 @@ def test_retrain_model_success_data_success_retrain(
     mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow
 ):
     mock_mkdtemp.return_value = 'tmp'
+    mock_to_datetime.side_effect = lambda idx, **kwargs: pd.DatetimeIndex(idx)
 
     mv_alias = MagicMock()
     mv_alias.run_id = 'source-run'
@@ -470,6 +471,7 @@ def test_retrain_model_always_uses_retrain_even_with_full_retrain_flag(
     mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow
 ):
     mock_mkdtemp.return_value = 'tmp'
+    mock_to_datetime.side_effect = lambda idx, **kwargs: pd.DatetimeIndex(idx)
     mv_alias = MagicMock(run_id='src')
     mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv_alias
     wrapper = MagicMock()

tests/laborious/activities/test_model_metrics.py

@@ -229,7 +229,6 @@ def test_calculate_drift_empty_drift_df(model_metrics_activity):
 
 def test_calculate_drift_empty_after_timestamp_filter(model_metrics_activity):
     """Rows dropped by target-window alignment yield an empty export list, not an insufficient-data error."""
-    ts_target = Timestamp('2023-05-26 11:12:27')
     drift_df = _sample_drift_metrics_df(Timestamp('2020-01-01'))
     model_metrics_activity.get_drift_metrics = MagicMock(return_value=drift_df)
 
@@ -271,7 +270,7 @@ def test_calculate_drift_drift_insufficient_data_error_from_lib(
 
     lib_msg = (
         '[MODEL_METRICS_DRIFT_INSUFFICIENT_DATA] Drift analysis produced no time chunks '
-        '(chunk_period=\'min\', analysis_rows=1).'
+        "(chunk_period='min', analysis_rows=1)."
     )
     model_metrics_activity.get_drift_metrics = MagicMock(
         side_effect=DriftInsufficientDataError(lib_msg, analysis_rows=1)

tests/laborious/activities/test_opc.py

@@ -5,7 +5,16 @@ from pandas import DataFrame
 from pytest import mark
 from sientia_do.notifications.models import NotificationLevel
 
-from laborious.activities.opc import OPC
+from laborious.activities.opc import (
+    OPC,
+    OPC_COMMENT_SEPARATOR,
+    OPC_RECONNECT_IN_PROGRESS_COMMENT,
+    OPC_SESSION_BAD_COMMENT_PREFIX,
+    OPC_SESSION_BAD_CONFIDENCE,
+    OPC_WRITTING_ERROR_CONFIDENCE,
+    OPC_WRITTING_ERROR_MESSAGE,
+    _apply_opc_write_error,
+)
 
 metadata = {
     'metadata': {
@@ -35,20 +44,26 @@ def test__init__():
 def test_init_opc(mock_send_notification, mock_opc_repository):
     mock_logger = MagicMock()
     mock_metrics_controller = MagicMock()
-    server1 = MagicMock()
+    server1 = MagicMock(
-    server1.connect.return_value = (True, {})
+        connect=MagicMock(return_value=(True, {})), write_data=MagicMock(return_value=(True, {}))
-    server2 = MagicMock()
+    )
-    server2.connect.return_value = (True, {})
+    server2 = MagicMock(
-    server3 = MagicMock()
+        connect=MagicMock(return_value=(True, {})), write_data=MagicMock(return_value=(True, {}))
-    server3.connect.return_value = (
+    )
-        False,
+    server3 = MagicMock(
-        {
+        connect=MagicMock(
-            'notification_id': 'OPC_CONNECTION_ERROR_server3',
+            return_value=(
-            'message': 'Failed to connect to OPC server: Test error',
+                False,
-            'block': 'opc_repository',
+                {
-            'level': NotificationLevel.ERROR,
+                    'notification_id': 'OPC_CONNECTION_ERROR_server3',
-            'attachment_content': 'Test error',
+                    'message': 'Failed to connect to OPC server: Test error',
-        },
+                    'block': 'opc_repository',
+                    'level': NotificationLevel.ERROR,
+                    'attachment_content': 'Test error',
+                },
+            )
+        ),
+        write_data=MagicMock(return_value=(True, {})),
     )
     mock_opc_repository.side_effect = [server1, server2, server3]
     mock_notification_handler = MagicMock()
@@ -105,12 +120,13 @@ def test_init_opc(mock_send_notification, mock_opc_repository):
                 server_name='server1',
                 url='http://localhost:8080',
                 logger=mock_logger,
-                notification_handler=mock_notification_handler,
-                metrics_controller=mock_metrics_controller,
                 server_uri='opc.tcp://localhost:4840',
                 cert_path='',
                 private_key_path='',
                 server_cert_path='',
+                notification_handler=mock_notification_handler,
+                reconnection_interval=60,
+                metrics_controller=mock_metrics_controller,
             ),
         ]
     )
@@ -121,12 +137,13 @@ def test_init_opc(mock_send_notification, mock_opc_repository):
                 server_name='server2',
                 url='http://localhost:8080',
                 logger=mock_logger,
-                notification_handler=mock_notification_handler,
-                metrics_controller=mock_metrics_controller,
                 server_uri='opc.tcp://localhost:4840',
                 cert_path='',
                 private_key_path='',
                 server_cert_path='',
+                notification_handler=mock_notification_handler,
+                reconnection_interval=60,
+                metrics_controller=mock_metrics_controller,
             )
         ]
     )
@@ -154,40 +171,40 @@ def test_init_opc(mock_send_notification, mock_opc_repository):
 
 
 @pytest.fixture
-def opc():
+@patch('laborious.activities.opc.OpcRepository')
-    with patch('laborious.activities.opc.OpcRepository') as mock_opc_repository:
+def opc(mock_opc_repository):
-        mock_opc_repository.return_value.write_data = MagicMock(return_value=(True, {}))
+    servers = {
-        mock_opc_repository.return_value.connect = MagicMock(return_value=(True, {}))
+        'server1': {
-        servers = {
+            'id': 'server1',
-            'server1': {
+            'server_name': 'server1',
-                'id': 'server1',
+            'url': 'http://localhost:8080',
-                'server_name': 'server1',
+            'server_uri': 'opc.tcp://localhost:4840',
-                'url': 'http://localhost:8080',
+            'cert_path': '',
-                'server_uri': 'opc.tcp://localhost:4840',
+            'private_key_path': '',
-                'cert_path': '',
+            'server_cert_path': '',
-                'private_key_path': '',
+            'reconnection_interval': 60,
-                'server_cert_path': '',
-                'reconnection_interval': 60,
-            }
         }
+    }
 
-        opc_instance = OPC(
+    mock_opc_repository.return_value.write_data = MagicMock(return_value=(True, {}))
-            opc_servers=servers,
+    mock_opc_repository.return_value.connect = MagicMock(return_value=(True, {}))
-            logger=MagicMock(),
+    opc = OPC(
-            notification_handler=MagicMock(),
+        opc_servers=servers,
-            metrics_controller=MagicMock(),
+        logger=MagicMock(),
-        )
+        notification_handler=MagicMock(),
-        opc_instance.init_opc()
+        metrics_controller=MagicMock(),
-        opc_instance.send_notification = MagicMock()
+    )
-        opc_instance.emit_metric_sync = MagicMock()
+    opc.init_opc()
-        yield opc_instance
+    opc.send_notification = MagicMock()
+    opc.emit_metric_sync = MagicMock()
+    return opc
 
 
 WRITE_DATA_CASES = [
     ('tag1', 'int', 50),
     ('tag2', 'float', 50.5),
     ('tag3', 'bool', True),
-    ('tag4', 'str', 'test'),
+    ('tag4', 'string', 'test'),
 ]
 
 
@@ -195,7 +212,7 @@ WRITE_DATA_CASES = [
 def test_write_data_success(opc, tag, data_type, data):
     opc.opc_repository['server1'].write_data.return_value = (True, {'response_time': 0.1})
 
-    result = opc.write_data(
+    response_time, error_info = opc.write_data(
         server_id='server1',
         tag=tag,
         data=data,
@@ -203,10 +220,9 @@ def test_write_data_success(opc, tag, data_type, data):
         tag_type='prediction',
         metadata=metadata,
     )
-    assert result == 0.1
+    assert response_time == 0.1
-    opc.opc_repository['server1'].write_data.assert_called_once_with(
+    assert error_info is None
-        tag, data, data_type, metadata
+    opc.opc_repository['server1'].write_data.assert_called_once_with(tag, data, data_type, metadata)
-    )
 
 
 def test_write_data_failed(opc):
@@ -221,7 +237,7 @@ def test_write_data_failed(opc):
         },
     )
 
-    result = opc.write_data(
+    response_time, error_info = opc.write_data(
         server_id='server1',
         tag='tag1',
         data=50,
@@ -229,7 +245,8 @@ def test_write_data_failed(opc):
         tag_type='prediction',
         metadata=metadata,
     )
-    assert result is None
+    assert response_time is None
+    assert error_info is not None
 
     opc.send_notification.assert_called_once_with(
         metadata=metadata,
@@ -268,16 +285,205 @@ def test_write_data_exception(opc):
         raise AssertionError('Expected an exception to be raised')
 
 
-def test_manage_output_tags_success(opc):
+@mark.parametrize(
-    opc.write_data = MagicMock(return_value=0.1)
+    'error_info,initial_seen,initial_status,initial_reconnect,expected',
+    [
+        (None, False, None, False, (False, None, False)),
+        ({}, False, None, False, (False, None, False)),
+        (
+            {'opc_error_kind': 'session_bad', 'opc_status': 'BadSessionIdInvalid'},
+            False,
+            None,
+            False,
+            (True, 'BadSessionIdInvalid', False),
+        ),
+        (
+            {'opc_error_kind': 'session_bad', 'opc_status': 'NewStatus'},
+            True,
+            'OldStatus',
+            False,
+            (True, 'NewStatus', False),
+        ),
+        (
+            {'opc_error_kind': 'session_bad'},
+            True,
+            'KeptStatus',
+            False,
+            (True, 'KeptStatus', False),
+        ),
+        (
+            {'opc_error_kind': 'reconnect_in_progress'},
+            False,
+            None,
+            False,
+            (False, None, True),
+        ),
+        (
+            {'opc_error_kind': 'other'},
+            True,
+            'Status',
+            True,
+            (True, 'Status', True),
+        ),
+    ],
+)
+def test_apply_opc_write_error(
+    error_info, initial_seen, initial_status, initial_reconnect, expected
+):
+    result = _apply_opc_write_error(
+        error_info,
+        initial_seen,
+        initial_status,
+        initial_reconnect,
+    )
+    assert result == expected
 
+
+def test_write_tags_from_config_prediction_success(opc):
+    opc.write_data = MagicMock(return_value=(0.1, None))
+    data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
+    tags_config = {'tag1': {'data_type': 'float'}}
+
+    response_times, session_bad, opc_status, reconnect = opc._write_tags_from_config(
+        server_id='server1',
+        tags_config=tags_config,
+        data=data,
+        data_column='prediction',
+        tag_type='prediction',
+        log_label='Prediction data',
+        metadata=metadata['metadata'],
+    )
+
+    assert response_times == {'tag1': 0.1}
+    assert session_bad is False
+    assert opc_status is None
+    assert reconnect is False
+    opc.write_data.assert_called_once_with(
+        server_id='server1',
+        tag='tag1',
+        data=0.75,
+        data_type='float',
+        tag_type='prediction',
+        metadata=metadata['metadata'],
+    )
+
+
+def test_write_tags_from_config_confidence_success(opc):
+    opc.write_data = MagicMock(return_value=(0.2, None))
+    data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
+    tags_config = {'tag2': {'data_type': 'float'}}
+
+    response_times, session_bad, opc_status, reconnect = opc._write_tags_from_config(
+        server_id='server1',
+        tags_config=tags_config,
+        data=data,
+        data_column='prediction_confidence',
+        tag_type='confidence',
+        log_label='Confidence data',
+        metadata=metadata['metadata'],
+    )
+
+    assert response_times == {'tag2': 0.2}
+    assert session_bad is False
+    assert opc_status is None
+    assert reconnect is False
+    opc.write_data.assert_called_once_with(
+        server_id='server1',
+        tag='tag2',
+        data=0.95,
+        data_type='float',
+        tag_type='confidence',
+        metadata=metadata['metadata'],
+    )
+
+
+def test_write_tags_from_config_write_failure(opc):
+    opc.write_data = MagicMock(return_value=(None, {}))
+    data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
+
+    response_times, session_bad, opc_status, reconnect = opc._write_tags_from_config(
+        server_id='server1',
+        tags_config={'tag1': {'data_type': 'float'}},
+        data=data,
+        data_column='prediction',
+        tag_type='prediction',
+        log_label='Prediction data',
+        metadata=metadata['metadata'],
+    )
+
+    assert response_times == {'tag1': None}
+    assert session_bad is False
+    assert opc_status is None
+    assert reconnect is False
+
+
+def test_write_tags_from_config_session_bad(opc):
+    opc.write_data = MagicMock(
+        return_value=(
+            None,
+            {
+                'opc_error_kind': 'session_bad',
+                'opc_status': 'BadSessionIdInvalid',
+            },
+        )
+    )
+    data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
+
+    response_times, session_bad, opc_status, reconnect = opc._write_tags_from_config(
+        server_id='server1',
+        tags_config={'tag1': {'data_type': 'float'}},
+        data=data,
+        data_column='prediction',
+        tag_type='prediction',
+        log_label='Prediction data',
+        metadata=metadata['metadata'],
+    )
+
+    assert response_times == {'tag1': None}
+    assert session_bad is True
+    assert opc_status == 'BadSessionIdInvalid'
+    assert reconnect is False
+
+
+def test_write_tags_from_config_reconnect_in_progress(opc):
+    opc.write_data = MagicMock(
+        return_value=(
+            None,
+            {'opc_error_kind': 'reconnect_in_progress'},
+        )
+    )
+    data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
+
+    response_times, session_bad, opc_status, reconnect = opc._write_tags_from_config(
+        server_id='server1',
+        tags_config={'tag1': {'data_type': 'float'}},
+        data=data,
+        data_column='prediction',
+        tag_type='prediction',
+        log_label='Prediction data',
+        metadata=metadata['metadata'],
+    )
+
+    assert response_times == {'tag1': None}
+    assert session_bad is False
+    assert opc_status is None
+    assert reconnect is True
+
+
+def test_manage_output_tags_success(opc):
+    opc._write_tags_from_config = MagicMock(
+        side_effect=[
+            ({'tag1': 0.1}, False, None, False),
+            ({'tag2': 0.1}, False, None, False),
+        ]
+    )
     data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
     config = {
         'prediction_tags': {'tag1': {'data_type': 'float'}},
         'confidence_tags': {'tag2': {'data_type': 'float'}},
     }
 
-    output_data, opc_metrics = opc.manage_output_tags(
+    output_data, opc_metrics, session_bad, opc_status, reconnect = opc.manage_output_tags(
         server_id='server1',
         config=config,
         data=data,
@@ -286,86 +492,57 @@ def test_manage_output_tags_success(opc):
 
     assert output_data is True
     assert opc_metrics == {'tag1': 0.1, 'tag2': 0.1}
-    opc.write_data.assert_has_calls(
+    assert session_bad is False
-        [
+    assert opc_status is None
-            call(
+    assert reconnect is False
-                server_id='server1',
+    assert opc._write_tags_from_config.call_count == 2
-                tag='tag1',
+
-                data=0.75,
+
-                data_type='float',
+def test_manage_output_tags_failed(opc):
-                tag_type='prediction',
+    opc._write_tags_from_config = MagicMock(
-                metadata=metadata['metadata'],
+        side_effect=[
-            ),
+            ({'tag1': 0.1}, False, None, False),
-            call(
+            ({'tag2': None}, False, None, False),
-                server_id='server1',
-                tag='tag2',
-                data=0.95,
-                data_type='float',
-                tag_type='confidence',
-                metadata=metadata['metadata'],
-            ),
         ]
     )
-
-
-@mark.parametrize('side_effect', [[0.1, None], [None, 0.2]])
-def test_manage_output_tags_failed(opc, side_effect):
-    opc.write_data = MagicMock(side_effect=side_effect)
     data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
     config = {
         'prediction_tags': {'tag1': {'data_type': 'float'}},
         'confidence_tags': {'tag2': {'data_type': 'float'}},
     }
-    output_data, opc_metrics = opc.manage_output_tags(
+
+    output_data, opc_metrics, _, _, _ = opc.manage_output_tags(
         server_id='server1',
         config=config,
         data=data,
         metadata=metadata['metadata'],
     )
+
     assert output_data is False
-    assert opc_metrics == {'tag1': side_effect[0], 'tag2': side_effect[1]}
+    assert opc_metrics == {'tag1': 0.1, 'tag2': None}
-    opc.write_data.assert_has_calls(
-        [
-            call(
-                server_id='server1',
-                tag='tag1',
-                data=0.75,
-                data_type='float',
-                tag_type='prediction',
-                metadata=metadata['metadata'],
-            ),
-            call(
-                server_id='server1',
-                tag='tag2',
-                data=0.95,
-                data_type='float',
-                tag_type='confidence',
-                metadata=metadata['metadata'],
-            ),
-        ]
-    )
 
 
 def test_manage_output_tags_do_nothing(opc):
-    opc.write_data = MagicMock(return_value=0.1)
+    opc._write_tags_from_config = MagicMock()
     data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
-    config = {
+    config = {'_invalid_key': {'tag1': {'data_type': 'float'}}}
-        '_invalid_key': {'tag1': {'data_type': 'float'}},
+
-    }
+    output_data, opc_metrics, _, _, _ = opc.manage_output_tags(
-    output_data, opc_metrics = opc.manage_output_tags(
         server_id='server1',
         config=config,
         data=data,
         metadata=metadata['metadata'],
     )
+
     assert output_data is True
     assert opc_metrics == {}
-    opc.write_data.assert_not_called()
+    opc._write_tags_from_config.assert_not_called()
 
 
 @patch('laborious.activities.opc.DataFrame')
 def test_write_opc_data_success(mock_dataframe, opc):
-    input_data: dict[str, object] = {
+    # Arrange
+    input_data = {
         **metadata,
         'data': {'prediction': [0.75], 'prediction_confidence': [0.95]},
         'opc_output_config': {
@@ -375,19 +552,21 @@ def test_write_opc_data_success(mock_dataframe, opc):
             }
         },
     }
-    opc_output_config = input_data['opc_output_config']
-    assert isinstance(opc_output_config, dict)
 
-    opc.manage_output_tags = MagicMock(return_value=(True, {'tag1': 0.1, 'tag2': 0.2}))
+    # Act
+    opc.manage_output_tags = MagicMock(
+        return_value=(True, {'tag1': 0.1, 'tag2': 0.2}, False, None, False)
+    )
 
     opc.process_confidence = MagicMock(return_value={'data': 'data'})
     output_data, opc_metrics = opc.write_opc_data(input_data)
 
+    # Assert
     assert output_data == {'data': 'data'}
     assert opc_metrics == {'server1': {'tag1': 0.1, 'tag2': 0.2}}
     opc.manage_output_tags.assert_called_once_with(
         'server1',
-        opc_output_config['server1'],
+        input_data['opc_output_config']['server1'],
         mock_dataframe.return_value,
         metadata['metadata'],
     )
@@ -395,10 +574,14 @@ def test_write_opc_data_success(mock_dataframe, opc):
         mock_dataframe.return_value,
         True,
         metadata['metadata'],
+        session_bad=False,
+        opc_status=None,
+        reconnect_in_progress=False,
     )
 
 
 def test_write_opc_data_empty_config(opc):
+    # Arrange
     input_data = {
         **metadata,
         'data': {'prediction': [0.75], 'prediction_confidence': [0.95]},
@@ -406,8 +589,10 @@ def test_write_opc_data_empty_config(opc):
         'opc_output_config': {'server1': {'prediction_tags': {}, 'confidence_tags': {}}},
     }
 
+    # Act
     opc.write_opc_data(input_data)
 
+    # Assert
     opc.opc_repository['server1'].write_data.assert_not_called()
 
 
@@ -424,8 +609,10 @@ def test_write_opc_data_no_validate_server(opc):
         },
     }
 
+    # Act
     opc.write_opc_data(input_data)
 
+    # Assert
     opc.opc_repository['server1'].write_data.assert_not_called()
 
 
@@ -437,18 +624,97 @@ def test_write_opc_data_no_validate_server(opc):
     ],
 )
 def test_process_confidence(opc, data, success, expected):
-    result = opc.process_confidence(data, success, metadata)
+    result = opc.process_confidence(data, success, metadata['metadata'])
-
     assert result['prediction_confidence'][0] == expected
 
 
+def test_process_confidence_session_bad(opc):
+    data = DataFrame({'prediction_confidence': [0.9]})
+    result = opc.process_confidence(
+        data,
+        False,
+        metadata['metadata'],
+        session_bad=True,
+        opc_status='BadSessionIdInvalid',
+    )
+    assert result['prediction_confidence'][0] == OPC_SESSION_BAD_CONFIDENCE
+    assert result['comments'][0].startswith(OPC_SESSION_BAD_COMMENT_PREFIX)
+    assert 'BadSessionIdInvalid' in result['comments'][0]
+
+
+def test_process_confidence_generic_failure(opc):
+    data = DataFrame({'prediction_confidence': [0.9]})
+    result = opc.process_confidence(data, False, metadata['metadata'])
+    assert result['prediction_confidence'][0] == OPC_WRITTING_ERROR_CONFIDENCE
+    assert result['comments'][0] == OPC_WRITTING_ERROR_MESSAGE
+
+
+def test_manage_output_tags_merges_error_flags(opc):
+    opc._write_tags_from_config = MagicMock(
+        side_effect=[
+            ({'tag1': None}, True, 'BadSessionIdInvalid', False),
+            ({'tag2': 0.2}, False, None, True),
+        ]
+    )
+    data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
+    config = {
+        'prediction_tags': {'tag1': {'data_type': 'float'}},
+        'confidence_tags': {'tag2': {'data_type': 'float'}},
+    }
+
+    (
+        success,
+        metrics,
+        session_bad_seen,
+        opc_status,
+        reconnect_in_progress,
+    ) = opc.manage_output_tags('server1', config, data, metadata['metadata'])
+
+    assert success is False
+    assert session_bad_seen is True
+    assert reconnect_in_progress is True
+    assert opc_status == 'BadSessionIdInvalid'
+    assert metrics == {'tag1': None, 'tag2': 0.2}
+
+
+def test_process_confidence_reconnect_in_progress(opc):
+    data = DataFrame({'prediction_confidence': [0.9]})
+    result = opc.process_confidence(
+        data,
+        False,
+        metadata['metadata'],
+        reconnect_in_progress=True,
+    )
+    assert result['prediction_confidence'][0] == OPC_SESSION_BAD_CONFIDENCE
+    assert result['comments'][0] == OPC_RECONNECT_IN_PROGRESS_COMMENT
+
+
+def test_process_confidence_concatenates_multiple_comments(opc):
+    data = DataFrame({'prediction_confidence': [0.9]})
+    session_comment = f'{OPC_SESSION_BAD_COMMENT_PREFIX} BadSessionIdInvalid'
+
+    result = opc.process_confidence(
+        data,
+        False,
+        metadata['metadata'],
+        session_bad=True,
+        opc_status='BadSessionIdInvalid',
+        reconnect_in_progress=True,
+    )
+
+    assert result['prediction_confidence'][0] == OPC_SESSION_BAD_CONFIDENCE
+    assert result['comments'][0] == OPC_COMMENT_SEPARATOR.join(
+        [session_comment, OPC_RECONNECT_IN_PROGRESS_COMMENT]
+    )
+
+
 def test_validate_server(opc):
     assert opc.validate_server('server1', metadata) is True
     assert opc.validate_server('server2', metadata) is False
 
 
 def test_close(opc):
-    disconnect_mock = MagicMock(return_value=None)
+    repo = opc.opc_repository['server1']
-    opc.opc_repository['server1'].disconnect = disconnect_mock
+    repo.disconnect = MagicMock(return_value=True)
     opc.close()
-    disconnect_mock.assert_called_once()
+    repo.disconnect.assert_called_once()

tests/laborious/utils/repository/test_opc_repository.py

@@ -1,12 +1,20 @@
+import concurrent.futures
 import json
 from datetime import datetime
 from unittest.mock import ANY, MagicMock, Mock, patch
 
 import pytest
 from asyncua.crypto import security_policies
+from asyncua.ua.uaerrors import BadNodeIdUnknown, BadSessionIdInvalid
 from sientia_do.notifications.models import NotificationLevel
 
-from laborious.utils.repository.opc_repository import OpcRepository
+from laborious.utils.repository.opc_repository import (
+    OpcClientAlreadyExistsError,
+    OpcClientNotInitializedError,
+    OpcRepository,
+    OpcSessionAlreadyConnectedError,
+    is_reconnectable_opcua_bad,
+)
 
 
 @pytest.fixture
@@ -31,7 +39,13 @@ def opc_repository(mock_logger):
     )
     repository.disconnection_interval = 0.1
     repository.send_notification = MagicMock()
+    repository.send_notification = MagicMock()
     repository.emit_metric_sync = MagicMock()
+    repository.info = MagicMock()
+    repository.error = MagicMock()
+    repository.warning = MagicMock()
+    repository.debug = MagicMock()
+    repository._session_ready.set()
     return repository
 
 
@@ -39,7 +53,12 @@ def opc_repository(mock_logger):
 def mock_client():
     with patch('laborious.utils.repository.opc_repository.Client') as mock:
         client_instance = MagicMock()
-        client_instance.aio_obj = MagicMock()
+        aio = MagicMock()
+        client_instance.aio_obj = aio
+        aio.uaclient = MagicMock()
+        aio.uaclient.protocol = MagicMock(state='closed')
+        aio.session_timeout = 600_000
+        aio.secure_channel_timeout = 600_000
         mock.return_value = client_instance
         yield client_instance
 
@@ -65,13 +84,13 @@ def test_init(opc_repository):
     assert opc_repository.reconnection_interval == 60
     assert opc_repository.client is None
     assert opc_repository.last_reconnection_time is None
-    assert opc_repository.error_count == 0
 
 
 def test_set_security(opc_repository, mock_client):
     opc_repository.client = mock_client
     opc_repository.set_security()
 
+    mock_client.application_uri = 'urn:test:server'
     mock_client.set_security.assert_called_once_with(
         security_policies.SecurityPolicyBasic256,
         '/path/to/cert.pem',
@@ -79,64 +98,115 @@ def test_set_security(opc_repository, mock_client):
         None,
         '/path/to/server_cert.pem',
     )
-    assert mock_client.aio_obj.secure_channel_timeout == 10000000
+    assert mock_client.aio_obj.secure_channel_timeout == 600_000
-    assert mock_client.aio_obj.session_timeout == 10000000
+    assert mock_client.aio_obj.session_timeout == 600_000
 
 
 def test_set_security_missing_certificates(opc_repository):
     opc_repository.cert_path = None
     opc_repository.private_key_path = None
 
-    with pytest.raises(ValueError, match='Certificate and private key paths'):
+    try:
         opc_repository.set_security()
+    except ValueError as e:
+        assert str(e) == 'Certificate and private key paths must be provided for secure connection.'
 
 
 def test_set_security_missing_client(opc_repository):
     opc_repository.client = None
-    with pytest.raises(ValueError, match='Client must be initialized'):
+    try:
         opc_repository.set_security()
+    except ValueError as e:
+        assert str(e) == 'Client must be initialized before setting security'
 
 
 def test_connect_with_security(opc_repository, mock_client):
-    opc_repository.try_connect = MagicMock(return_value=(True, {}))
+    opc_repository._create_client = MagicMock()
+    opc_repository._open_session = MagicMock(return_value=(True, {}))
     result = opc_repository.connect()
 
-    opc_repository.try_connect.assert_called_once()
+    opc_repository._create_client.assert_called_once()
-    assert opc_repository.client == mock_client
+    opc_repository._open_session.assert_called_once()
     assert result == (True, {})
 
 
 def test_connect_without_security(opc_repository, mock_client):
     opc_repository.cert_path = None
-    opc_repository.try_connect = MagicMock(return_value=(True, {}))
+    opc_repository._create_client = MagicMock()
+    opc_repository._open_session = MagicMock(return_value=(True, {}))
     opc_repository.set_security = MagicMock()
     result = opc_repository.connect()
 
-    opc_repository.try_connect.assert_called_once()
+    opc_repository._create_client.assert_called_once()
+    opc_repository._open_session.assert_called_once()
     opc_repository.set_security.assert_not_called()
-    assert opc_repository.client == mock_client
     assert result == (True, {})
 
 
-def test_try_connect_success(opc_repository):
+def test_connect_raises_when_session_already_open(opc_repository, mock_client):
-    opc_repository.last_reconnection_time = None
+    opc_repository.client = mock_client
+    proto = MagicMock()
+    proto.state = 'open'
+    mock_client.aio_obj.uaclient.protocol = proto
+
+    with pytest.raises(OpcSessionAlreadyConnectedError, match='disconnect'):
+        opc_repository.connect()
+
+
+def test_create_client_raises_when_client_exists(opc_repository, mock_client):
+    opc_repository.client = mock_client
+
+    with pytest.raises(OpcClientAlreadyExistsError, match='already exists'):
+        opc_repository._create_client()
+
+
+def test_open_session_success(opc_repository):
+    closed_proto = MagicMock()
+    closed_proto.state = 'closed'
     opc_repository.client = MagicMock()
-    result = opc_repository.try_connect()
+    aio = MagicMock()
+    opc_repository.client.aio_obj = aio
+    aio.uaclient = MagicMock(protocol=closed_proto)
+    aio.session_timeout = 600_000
+    aio.secure_channel_timeout = 600_000
+
+    open_proto = MagicMock()
+    open_proto.state = 'open'
+    open_proto.authentication_token = 'tok'
+
+    def connect_side_effect():
+        aio.uaclient.protocol = open_proto
+
+    opc_repository.client.connect = MagicMock(side_effect=connect_side_effect)
+
+    result = opc_repository._open_session()
 
     opc_repository.client.connect.assert_called_once()
-    assert opc_repository.last_reconnection_time is not None
     assert result == (True, {})
+    assert opc_repository._session_ready.is_set()
 
 
-def test_try_connect_fail(opc_repository):
+def test_open_session_raises_when_already_connected(opc_repository, mock_client):
-    opc_repository.last_reconnection_time = None
+    opc_repository.client = mock_client
-    opc_repository.disconnect = MagicMock()
+    proto = MagicMock()
+    proto.state = 'open'
+    mock_client.aio_obj.uaclient.protocol = proto
+
+    with pytest.raises(OpcSessionAlreadyConnectedError, match='disconnect'):
+        opc_repository._open_session()
+
+
+def test_open_session_fail(opc_repository):
+    opc_repository._disconnect_locked = MagicMock()
     opc_repository.client = MagicMock()
-    opc_repository.client.connect.side_effect = Exception('Test error')
+    aio = MagicMock()
+    opc_repository.client.aio_obj = aio
+    aio.uaclient = MagicMock(protocol=MagicMock(state='closed'))
+    opc_repository.client.connect = MagicMock(side_effect=Exception('Test error'))
 
-    is_connected, error_data = opc_repository.try_connect()
+    is_connected, error_data = opc_repository._open_session()
 
-    opc_repository.disconnect.assert_called_once()
+    opc_repository._disconnect_locked.assert_called_once()
     opc_repository.client.connect.assert_called_once()
     assert is_connected is False
     assert error_data['notification_id'] == f'OPC_CONNECTION_ERROR_{opc_repository.id}'
@@ -146,29 +216,17 @@ def test_try_connect_fail(opc_repository):
     assert error_data['attachment_content'] is not None
 
 
-def test_try_connect_no_client(opc_repository):
+def test_open_session_raises_when_no_client(opc_repository):
     opc_repository.client = None
-    result = opc_repository.try_connect()
-    assert result == (
-        False,
-        {
-            'notification_id': f'OPC_CONNECTION_ERROR_{opc_repository.id}',
-            'message': 'Client is not initialized',
-            'block': 'opc_repository',
-            'level': NotificationLevel.ERROR,
-        },
-    )
 
-
+    with pytest.raises(OpcClientNotInitializedError, match='not initialized'):
-def test_session_alive_returns_false_when_client_none(opc_repository):
+        opc_repository._open_session()
-    opc_repository.client = None
-    assert opc_repository._session_alive() is False
 
 
 def test_disconnection_fallback_success(opc_repository, mock_client):
     opc_repository.client = mock_client
-    mock_client.disconnect.return_value = None
+    mock_client.disconnect.return_value = True
-    result = opc_repository.disconnection_fallback()
+    result = opc_repository._disconnection_fallback()
 
     mock_client.disconnect.assert_called_once()
     assert result == []
@@ -177,7 +235,7 @@ def test_disconnection_fallback_success(opc_repository, mock_client):
 def test_disconnection_fallback_fail(opc_repository, mock_client):
     opc_repository.client = mock_client
     mock_client.disconnect.side_effect = Exception('Test error')
-    result = opc_repository.disconnection_fallback()
+    result = opc_repository._disconnection_fallback()
     assert result == [
         {'attempt': 1, 'error': 'Test error', 'traceback': ANY},
         {'attempt': 2, 'error': 'Test error', 'traceback': ANY},
@@ -190,10 +248,10 @@ def test_disconnection_fallback_fail(opc_repository, mock_client):
 
 def test_disconnect(opc_repository, mock_client):
     opc_repository.client = mock_client
-    opc_repository.disconnection_fallback = MagicMock(return_value=[])
+    opc_repository._disconnection_fallback = MagicMock(return_value=[])
     opc_repository.disconnect()
 
-    opc_repository.disconnection_fallback.assert_called_once()
+    opc_repository._disconnection_fallback.assert_called_once()
     assert opc_repository.client is None
 
 
@@ -204,12 +262,12 @@ def test_disconnect_no_client(opc_repository):
 
 def test_disconnect_error(opc_repository, mock_client):
     opc_repository.client = mock_client
-    opc_repository.disconnection_fallback = MagicMock(
+    opc_repository._disconnection_fallback = MagicMock(
         return_value=[{'attempt': 1, 'error': 'Test error', 'traceback': 'text'}]
     )
     opc_repository.disconnect()
 
-    opc_repository.disconnection_fallback.assert_called_once()
+    opc_repository._disconnection_fallback.assert_called_once()
     opc_repository.send_notification.assert_called_once_with(
         metadata=opc_repository.metadata,
         notification_id=f'OPC_DISCONNECTION_ERROR_{opc_repository.id}',
@@ -225,67 +283,25 @@ def test_disconnect_error(opc_repository, mock_client):
 
 def test_validate_connection_none_client(opc_repository):
     opc_repository.client = None
-    opc_repository.connect = MagicMock(return_value=(True, {}))
     response = opc_repository.validate_connection()
-    assert response == (True, {})
+    assert response == (False, opc_repository._not_connected_error())
-    opc_repository.connect.assert_called_once()
 
 
-def test_validate_connection_disconnect_raises(opc_repository):
+def test_validate_connection_session_not_open(opc_repository):
-    """Outer except path when reconnect cleanup fails mid-validation."""
-
     opc_repository.client = MagicMock()
-    opc_repository._session_alive = MagicMock(return_value=False)
+    opc_repository.client.aio_obj.uaclient.protocol = None
-    opc_repository.last_reconnection_time = datetime(2020, 1, 1, 0, 0, 0)
-    opc_repository.disconnect = MagicMock(side_effect=RuntimeError('disconnect failed'))
 
     response = opc_repository.validate_connection()
 
-    assert response[0] is False
+    assert response == (False, opc_repository._not_connected_error())
-    assert response[1]['notification_id'] == f'OPC_CONNECTION_CHECK_ERROR_{opc_repository.id}'
+    opc_repository.error.assert_called_once()
-    assert 'disconnect failed' in response[1]['message']
-
-
-@patch('laborious.utils.repository.opc_repository.datetime')
-def test_validate_connection_lost_not_time_to_reconnect(_mock_datetime, opc_repository):
-    _mock_datetime.now = MagicMock(return_value=datetime(2025, 1, 1, 0, 0, 0))
-    opc_repository.error_count = 0
-    opc_repository.client = MagicMock()
-    opc_repository.client.get_root_node.side_effect = RuntimeError('down')
-    opc_repository.last_reconnection_time = datetime(2025, 1, 1, 0, 0, 0)
-    opc_repository.connect = MagicMock(return_value=(True, {}))
-
-    response = opc_repository.validate_connection()
-    opc_repository.connect.assert_not_called()
-    assert response == (
-        False,
-        {
-            'notification_id': f'OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{opc_repository.id}',
-            'message': f'OPC server {opc_repository.id} is not connected, waiting for next reconnection window...',
-            'block': 'opc_repository',
-            'level': NotificationLevel.WARNING,
-        },
-    )
-
-
-@patch('laborious.utils.repository.opc_repository.datetime')
-def test_validate_connection_lost_time_to_reconnect(mock_datetime, opc_repository):
-    mock_datetime.now = MagicMock(return_value=datetime(2025, 1, 1, 1, 0, 0))
-    opc_repository.error_count = 0
-    opc_repository.client = MagicMock()
-    opc_repository.client.get_root_node.side_effect = RuntimeError('down')
-    opc_repository.last_reconnection_time = datetime(2025, 1, 1, 0, 0, 0)
-    opc_repository.connect = MagicMock(return_value=(True, {}))
-
-    response = opc_repository.validate_connection()
-    opc_repository.connect.assert_called_once()
-    assert response == opc_repository.connect.return_value
 
 
 def test_validate_connection_success(opc_repository):
     opc_repository.client = MagicMock()
-    opc_repository.error_count = 0
+    proto = MagicMock()
-    opc_repository.client.get_root_node.return_value = MagicMock()
+    proto.state = 'open'
+    opc_repository.client.aio_obj.uaclient.protocol = proto
 
     output = opc_repository.validate_connection()
     assert output == (True, {})
@@ -293,28 +309,22 @@ def test_validate_connection_success(opc_repository):
 
 def test_write_data_validate_connection_do_nothing(opc_repository):
     opc_repository.validate_connection = MagicMock(return_value=(True, {}))
-    opc_repository.client = MagicMock()
+    opc_repository.client = MagicMock(get_node=MagicMock())
     mock_node = MagicMock()
     opc_repository.client.get_node.return_value = mock_node
 
-    result = opc_repository.write_data(
+    result = opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata'])
-        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
-    )
 
     opc_repository.validate_connection.assert_called_once()
     opc_repository.client.get_node.assert_called_once_with('ns=2;s=TestNode')
-    mock_node.set_value.assert_called_once()
     assert result == (True, {'response_time': ANY})
 
 
 def test_write_data_validate_connection_failed(opc_repository):
     opc_repository.validate_connection = MagicMock(return_value=(False, {}))
     opc_repository.client = MagicMock()
-    opc_repository.error_count = 0
 
-    result = opc_repository.write_data(
+    result = opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata'])
-        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
-    )
 
     opc_repository.validate_connection.assert_called_once()
     opc_repository.client.get_node.assert_not_called()
@@ -324,7 +334,6 @@ def test_write_data_validate_connection_failed(opc_repository):
 def test_write_data_get_node_failed(opc_repository):
     opc_repository.validate_connection = MagicMock(return_value=(True, {}))
     opc_repository.client = MagicMock()
-    opc_repository.error_count = 0
     opc_repository.client.get_node = MagicMock(side_effect=Exception('Test error'))
 
     is_success, error_data = opc_repository.write_data(
@@ -335,6 +344,13 @@ def test_write_data_get_node_failed(opc_repository):
     opc_repository.client.get_node.assert_called_once_with('ns=2;s=TestNode')
     assert is_success is False
     assert error_data['notification_id'] == f'OPC_WRITE_GET_NODE_ERROR_{opc_repository.id}'
+    assert (
+        error_data['message']
+        == "Failed to get node from OPC server: Test error | metadata: {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}"
+    )
+    assert error_data['block'] == 'opc_repository'
+    assert error_data['level'] == NotificationLevel.ERROR
+    assert error_data['attachment_content'] is not None
 
 
 def test_write_data_invalid_data_type(opc_repository, mock_client):
@@ -352,6 +368,13 @@ def test_write_data_invalid_data_type(opc_repository, mock_client):
 
     assert is_success is False
     assert error_data['notification_id'] == f'OPC_WRITE_DATA_TYPE_ERROR_{opc_repository.id}'
+    assert (
+        error_data['message']
+        == "Unsupported data type: invalid_type | metadata: {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}"
+    )
+    assert error_data['block'] == 'opc_repository'
+    assert error_data['level'] == NotificationLevel.ERROR
+    assert error_data.get('attachment_content') is None
 
 
 def test_write_data(opc_repository, mock_client):
@@ -360,12 +383,10 @@ def test_write_data(opc_repository, mock_client):
     mock_node = MagicMock()
     mock_client.get_node = MagicMock(return_value=mock_node)
 
-    result = opc_repository.write_data(
+    result = opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata'])
-        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
-    )
 
     mock_client.get_node.assert_called_once_with('ns=2;s=TestNode')
-    mock_node.set_value.assert_called_once()
+    mock_node.write_value.assert_called_once()
     assert result == (True, {'response_time': ANY})
 
 
@@ -373,9 +394,8 @@ def test_write_data_write_value_failed(opc_repository, mock_client):
     opc_repository.validate_connection = MagicMock(return_value=(True, {}))
     opc_repository.client = mock_client
     mock_node = MagicMock()
-    opc_repository.error_count = 0
     mock_client.get_node = MagicMock(return_value=mock_node)
-    mock_node.set_value.side_effect = Exception('Test error')
+    mock_node.write_value.side_effect = Exception('Test error')
 
     is_success, error_data = opc_repository.write_data(
         'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
@@ -383,6 +403,106 @@ def test_write_data_write_value_failed(opc_repository, mock_client):
 
     opc_repository.validate_connection.assert_called_once()
     mock_client.get_node.assert_called_once_with('ns=2;s=TestNode')
-    mock_node.set_value.assert_called_once()
+    mock_node.write_value.assert_called_once()
     assert is_success is False
     assert error_data['notification_id'] == f'OPC_WRITE_DATA_ERROR_{opc_repository.id}'
+    assert (
+        error_data['message']
+        == "Failed to write data to OPC server: Test error | metadata: {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}"
+    )
+    assert error_data['block'] == 'opc_repository'
+    assert error_data['level'] == NotificationLevel.ERROR
+    assert error_data['attachment_content'] is not None
+
+
+def test_is_reconnectable_opcua_bad():
+    assert is_reconnectable_opcua_bad(BadSessionIdInvalid()) is True
+    assert is_reconnectable_opcua_bad(BadNodeIdUnknown()) is False
+    assert is_reconnectable_opcua_bad(Exception('other')) is False
+
+
+def test_write_data_bad_session_id_invalid_schedules_reconnect(opc_repository, mock_client):
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
+    opc_repository.client = mock_client
+    opc_repository._start_reconnect_on_bad = MagicMock()
+    mock_node = MagicMock()
+    mock_client.get_node = MagicMock(return_value=mock_node)
+    mock_node.write_value.side_effect = BadSessionIdInvalid()
+
+    is_success, error_data = opc_repository.write_data(
+        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
+    )
+
+    mock_node.write_value.assert_called_once()
+    opc_repository._start_reconnect_on_bad.assert_called_once()
+    assert is_success is False
+    assert error_data['opc_error_kind'] == 'session_bad'
+    assert error_data['opc_status'] == 'BadSessionIdInvalid'
+
+
+def test_write_data_reconnect_in_progress_immediate(opc_repository):
+    opc_repository._session_ready.clear()
+    opc_repository._reconnect_thread = MagicMock()
+    opc_repository._reconnect_thread.is_alive.return_value = True
+    opc_repository.validate_connection = MagicMock()
+
+    is_success, error_data = opc_repository.write_data(
+        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
+    )
+
+    opc_repository.validate_connection.assert_not_called()
+    assert is_success is False
+    assert error_data['opc_error_kind'] == 'reconnect_in_progress'
+
+
+def test_start_reconnect_on_bad_skips_within_interval(opc_repository):
+    opc_repository.last_reconnection_time = datetime.now()
+    opc_repository.reconnection_interval = 3600
+
+    opc_repository._start_reconnect_on_bad('BadSessionIdInvalid', 'tok')
+
+    assert opc_repository._reconnect_thread is None
+
+
+def test_parallel_bad_writes_single_reconnect_task(opc_repository, mock_client):
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
+    opc_repository.client = mock_client
+    opc_repository.reconnection_interval = 0
+    opc_repository.last_reconnection_time = None
+    mock_node = MagicMock()
+    mock_client.get_node = MagicMock(return_value=mock_node)
+    mock_node.write_value.side_effect = BadSessionIdInvalid()
+    opc_repository._start_reconnect_on_bad = MagicMock()
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
+        futures = [
+            executor.submit(
+                opc_repository.write_data,
+                node,
+                value,
+                'float',
+                metadata['metadata'],
+            )
+            for node, value in (('ns=2;s=TestNode', 1.0), ('ns=2;s=TestNode2', 2.0))
+        ]
+        results = [future.result() for future in futures]
+
+    assert 1 <= opc_repository._start_reconnect_on_bad.call_count <= 2
+    assert mock_node.write_value.call_count == 2
+    error_kinds = [r[1].get('opc_error_kind') for r in results]
+    assert error_kinds.count('session_bad') >= 1
+    assert all(k in ('session_bad', 'reconnect_in_progress') for k in error_kinds)
+
+
+@patch('laborious.utils.repository.opc_repository.datetime')
+def test_reconnect_locked_sets_last_reconnection_time(mock_datetime, opc_repository):
+    mock_datetime.now = MagicMock(return_value=datetime(2025, 1, 1, 12, 0, 0))
+    opc_repository._disconnect_locked = MagicMock()
+    opc_repository._connect_locked = MagicMock(return_value=(True, {}))
+
+    result = opc_repository._reconnect_locked()
+
+    opc_repository._disconnect_locked.assert_called_once()
+    opc_repository._connect_locked.assert_called_once()
+    assert result == (True, {})
+    assert opc_repository.last_reconnection_time == datetime(2025, 1, 1, 12, 0, 0)

tests/laborious/utils/test_connectors_config.py

@@ -10,8 +10,7 @@ from laborious.utils.connectors_config import (
 
 def test_build_mlflow_config_with_env_vars():
     # Arrange
-    environ['MLFLOW_HOST'] = 'http://test-host'
+    environ['MLFLOW_URL'] = 'http://test-host:8080'
-    environ['MLFLOW_PORT'] = '8080'
     environ['MLFLOW_USERNAME'] = 'test-user'
     environ['MLFLOW_PASSWORD'] = 'test-pass'
 
@@ -25,8 +24,7 @@ def test_build_mlflow_config_with_env_vars():
 
 
 def test_build_mlflow_config_host_already_has_port():
-    environ['MLFLOW_HOST'] = 'http://tracker.example.com:443'
+    environ['MLFLOW_URL'] = 'http://tracker.example.com:443'
-    environ['MLFLOW_PORT'] = '8080'
     environ['MLFLOW_USERNAME'] = 'u'
     environ['MLFLOW_PASSWORD'] = 'p'
 
@@ -38,8 +36,7 @@ def test_build_mlflow_config_host_already_has_port():
 def test_build_mlflow_config_with_defaults():
     # Arrange
     # Clear any existing env vars
-    environ.pop('MLFLOW_HOST', None)
+    environ.pop('MLFLOW_URL', None)
-    environ.pop('MLFLOW_PORT', None)
     environ.pop('MLFLOW_USERNAME', None)
     environ.pop('MLFLOW_PASSWORD', None)