SIENTIAPDE-1811

Enhance OPC UA communication and metrics tracking - Updated README.md to include new OPC UA Communication section and detailed metrics for session and write diagnostics. - Added new metrics in laborious/metrics.py for tracking OPC UA session states and write attempts. - Refactored OPC activity in laborious/activities/opc.py to handle session errors and improve error reporting. - Updated e2e tests to cover new scenarios for OPC session/channel errors and reconnect handling. - Modified .gitignore to include relatorio files and mlruns directory. - Added ipykernel to requirements-dev.txt for Jupyter notebook support.
2026-05-15 15:28:28 -03:00
parent 4edf70c4a0
commit 7bf7848a93
8 changed files with 266 additions and 63 deletions
--- a/.gitignore
+++ b/.gitignore
@@ -55,4 +55,4 @@ mlruns/

 relatorio*
 openspec/*
-.cursor/*
+.cursor/*
--- a/README.md
+++ b/README.md
@@ -48,6 +48,7 @@ A comprehensive, Temporal-based ML orchestration system for industrial data proc
  - [Prediction Operation Metrics](#prediction-operation-metrics)
  - [OPC Export Metrics](#opc-export-metrics)
  - [Data Quality Metrics](#data-quality-metrics)
+- [OPC UA Communication](#opc-ua-communication)
 - [Configuration](#configuration-1)
  - [Environment Variables](#environment-variables)
  - [OPC Configuration](#opc-configuration)
@@ -164,7 +165,7 @@ Laborious uses a Temporal-based architecture with strong separation of concerns
 - `connectors_config.py`: Env-driven configuration builders
 - `models/minio_dataframe_payload.py`: MinIO-offloaded DataFrame payload model
 - ML models are loaded via `SientiaMLflowRepository` (wrapper-based, `@production` alias) constructed in `Activities` from `build_mlflow_config()`.
- `repository/opc_repository.py`: OPC communication and writes
+- `repository/opc_repository.py`: OPC UA client, writes, session recovery (see [OPC UA Communication](#opc-ua-communication))
 - `repository/minio_manager.py`: MinIO object storage operations
 - `filters/conditional_filters.py` and `filters/mlflow_filters.py`

@@ -778,6 +779,15 @@ The Laborious system exposes comprehensive Prometheus metrics for operational vi
  - Labels: `pod_id`, `model_name`, `workflow_name`, `opc_server_id`
  - Buckets: [0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0]

+OPC UA session and write diagnostics (Prometheus, `laborious/metrics.py`):
+
+- `opc_connections_initiated_total`, `opc_connections_failed_total`, `opc_connection_status`
+- `opc_session_created_total`, `opc_session_closed_total`, `opc_session_revised_timeout_milliseconds`
+- `opc_write_attempts_total` (label `result`: `OK` or exception name, e.g. `BadSessionIdInvalid`)
+- `opc_write_inter_arrival_over_session_timeout_total`
+
+See [OPC UA Communication](#opc-ua-communication) for semantics, concurrency, and confidence codes **12** / **14**.
+
 ### Data Quality Metrics
 - Filter pass/fail rates through notification system
 - MLFlow API response validation metrics
@@ -822,7 +832,7 @@ The Laborious system exposes comprehensive Prometheus metrics for operational vi
 | `OPC_CERT_PATH` | OPC client certificate path | `None` | No |
 | `OPC_PRIVATE_KEY_PATH` | OPC private key path | `None` | No |
 | `OPC_SERVER_CERT_PATH` | OPC server certificate path | `None` | No |
-| `OPC_RECONNECTION_INTERVAL` | OPC reconnection interval (ms) | `120` | No |
+| `OPC_RECONNECTION_INTERVAL` | Minimum seconds between OPC reconnects | `120` | No |
 | `PI_WEB_API_BASE_URL` | PI Web API server base URL | `None` | No |
 | `PI_WEB_API_AUTH_TYPE` | PI Web API authentication type (basic/bearer) | `None` | No |
 | `PI_WEB_API_AUTH_TOKEN` | PI Web API authentication token | `None` | No |
@@ -915,6 +925,12 @@ Legacy MinIO object layout (relative key):
 `training_datasets/{model_name}/{object_prefix}_{timestamp}.parquet` where `object_prefix` is sanitized
 (slashes replaced by underscores) to keep a stable model-level directory.

+## OPC UA Communication
+
+Full reference: **[docs/opc-communication.md](docs/opc-communication.md)** (connection lifecycle, Tier-1 `Bad*` reconnect, connection lock / session readiness, metrics, PostgreSQL confidence **12** vs **14**, tests).
+
+Implementation plan: [`.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md`](.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md).
+
 ### OPC Configuration

 For multiple OPC servers, use the `OPC_CONFIG` environment variable:
@@ -1180,8 +1196,10 @@ laborious/
   - Ensure proper connection pool configuration

 4. **OPC Connection Failures**
-   - Verify OPC server is accessible
+   - See [docs/opc-communication.md](docs/opc-communication.md)
+   - Verify OPC server is accessible and `OPC_RECONNECTION_INTERVAL` is appropriate
   - Check certificate and key file paths
+   - Correlate `opc_write_attempts_total` with `opc_session_*` metrics; count session errors via `prediction_confidence = 14`
   - Review OPC server logs for connection issues

 5. **PI Web API Connection Failures**
--- a/docs/opc-communication.md
+++ b/docs/opc-communication.md
@@ -12,7 +12,7 @@ Worker (long-lived)
        ├── 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*
+        └── background reconnect on Tier-1 Bad*, closed protocol, or stale session

 Temporal activity write_opc_data
  └── OPC.manage_output_tags → write_data per tag (sequential per activity)
@@ -26,9 +26,9 @@ One worker process holds one `OpcRepository` instance per configured server. Mul
 |-------|----------|
 | 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 |
+| Tier-1 Bad* / protocol closed / session not ready | `_start_reconnect(reason)` → `_run_reconnect` (thread) → `_reconnect_locked()` (respects `reconnection_interval`) |
+| Write | `write_data()` checks in-flight reconnect thread, `_session_ready`, validates, then one `get_node` + `write_value` |
+| Shutdown | `disconnect()` sets `_allow_reconnect = False`, then tears down session |

 ### Session and channel timeouts

@@ -44,8 +44,9 @@ To allow **multiple concurrent writes** when the session is healthy, but **block

 | 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_lock` (`threading.Lock`) | Held for the entire `disconnect` → `connect` path. Only one connection-maintenance task at a time. |
+| `_session_ready` (`threading.Event`) | Set when a session is ready for writes; cleared before reconnect starts and set again after a successful connect. |
+| `_allow_reconnect` | Cleared in `disconnect()` so shutdown does not spawn reconnect threads |

 **Connection methods (caller holds `_connection_lock` for `_*_locked` helpers):**

@@ -61,21 +62,20 @@ Public `connect()` / `disconnect()` acquire the lock and call `_connect_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).
+1. If a reconnect **thread** is alive → `reconnect_in_progress`.
+2. If `_session_ready` is cleared → schedule `SessionNotReady` reconnect; return `reconnect_in_progress` or `connection_lost`.
+3. If `validate_connection()` fails (protocol closed) → schedule `ProtocolClosed` reconnect; return `connection_lost`.
+4. Single `get_node` + `write_value` (no retry in the same call).

-**Reconnect path (`_run_reconnect_on_bad`):**
+**Reconnect path (`_run_reconnect`):**

-1. `_start_reconnect_on_bad` clears `_session_ready` and schedules the task when the interval allows.
-2. `async with _connection_lock:` → `_reconnect_locked()`.
+1. `_start_reconnect` clears `_session_ready` and starts a daemon thread when the interval allows and `_allow_reconnect` is true.
+2. `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.
+**asyncua note:** Concurrent `write_value` on the same session is only safe if the stack tolerates it. If production shows issues, serialize writes while keeping the connection lock semantics above.

 ## Tier-1 `Bad*` errors and reconnect

--- a/inter_arrival.py
+++ b/inter_arrival.py
@@ -0,0 +1,25 @@
+# %%
+
+# Load logs.txt
+with open('logs.txt', 'r') as file:
+    lines = file.readlines()
+
+# %%
+import re
+# Grep "inter-arrival_s=number" with regex
+intervals = []
+for line in lines:
+    match = re.search(r'inter-arrival_s=([0-9.]+)', line)
+    if match:
+        intervals.append(float(match.group(1)))
+# %%
+
+print(intervals)
+# %%
+import matplotlib.pyplot as plt
+plt.plot(intervals)
+plt.ylabel('Inter-arrival time (s)')
+plt.xlabel('Sample')
+plt.title('Inter-arrival time distribution')
+plt.show()
+# %%
--- a/laborious/utils/repository/opc_repository.py
+++ b/laborious/utils/repository/opc_repository.py
@@ -211,6 +211,7 @@ class OpcRepository(SientiaMonitoring):
        self._connection_lock = threading.Lock()
        self._session_ready = threading.Event()
        self._reconnect_thread: threading.Thread | None = None
+        self._allow_reconnect = True

    def _opc_debug_tags(self, session_id: str) -> dict[str, str]:
        return {
@@ -513,8 +514,12 @@ class OpcRepository(SientiaMonitoring):
    def disconnect(self) -> None:
        """
        Gracefully disconnect from the OPC server under the connection lock.
+
+        Disables background reconnect so late writes during worker shutdown do not
+        respawn sessions.
        """
        with self._connection_lock:
+            self._allow_reconnect = False
            self._disconnect_locked()

    def validate_connection(self) -> tuple[bool, dict[str, Any]]:
@@ -531,61 +536,82 @@ class OpcRepository(SientiaMonitoring):
        self.error(f'OPC server {self.id} is not connected', self.metadata)
        return False, self._not_connected_error()

-    def _start_reconnect_on_bad(self, opc_status: str, session_id: str) -> None:
+    def _reconnect_thread_in_progress(self) -> bool:
        """
-        Schedule a background reconnect if interval and thread state allow it.
+        Return whether a background reconnect thread is currently running.
+
+        Return:
+            bool: True when a reconnect thread exists and is alive.
+        """
+        return self._reconnect_thread is not None and self._reconnect_thread.is_alive()
+
+    def _start_reconnect(self, reason: str, session_id: str) -> None:
+        """
+        Schedule a background reconnect when allowed by interval and thread state.
+
+        Clears _session_ready before starting the thread. No-op when _allow_reconnect is
+        False, the reconnection window has not elapsed, or a reconnect is already running.

        Args:
-            opc_status (str): OPC UA status name that triggered reconnect.
+            reason (str): Trigger for reconnect (OPC status name or synthetic reason).
            session_id (str): Session token before failure.
        """
+        if not self._allow_reconnect:
+            return
        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}',
+                f'reconnect_reason={reason}',
                self.metadata,
            )
            return
-        if self._reconnect_thread is not None and self._reconnect_thread.is_alive():
+        if self._reconnect_thread_in_progress():
            self.warning(
                f'OPC reconnect skipped reason=in_progress opc_server_id={self.id} '
-                f'opc_status={opc_status}',
+                f'reconnect_reason={reason}',
                self.metadata,
            )
            return

        self._session_ready.clear()
        self.info(
-            f'OPC reconnect scheduled after opc_status={opc_status} opc_server_id={self.id} '
+            f'OPC reconnect scheduled reconnect_reason={reason} 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),
+            target=self._run_reconnect,
+            args=(reason, session_id),
            daemon=True,
        )
        self._reconnect_thread.start()

-    def _run_reconnect_on_bad(self, opc_status: str, session_id: str) -> None:
+    def _run_reconnect(self, reason: 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.
+            reason (str): Trigger for reconnect (OPC status or ProtocolClosed).
            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'OPC reconnect started reconnect_reason={reason} opc_server_id={self.id} '
                    f'old_session_id={session_id}',
                    self.metadata,
                )
-                self._reconnect_locked()
+                success, error = self._reconnect_locked()
+            if not success:
+                self.error(
+                    f'OPC reconnect failed reconnect_reason={reason} opc_server_id={self.id}',
+                    self.metadata,
+                )
+                if error:
+                    self.error(error.get('message', ''), self.metadata)
        except Exception:
            self.error(
-                f'OPC reconnect task failed opc_server_id={self.id} opc_status={opc_status}',
+                f'OPC reconnect task failed opc_server_id={self.id} reconnect_reason={reason}',
                self.metadata,
            )
            self.error(traceback.format_exc(), self.metadata)
@@ -681,7 +707,7 @@ class OpcRepository(SientiaMonitoring):
            f'model_name={metadata.get("model_name", "unknown")} node={node} phase={phase}',
            metadata,
        )
-        self._start_reconnect_on_bad(opc_status, session_id)
+        self._start_reconnect(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}',
@@ -714,32 +740,89 @@ class OpcRepository(SientiaMonitoring):
                pass
        node_obj.set_value(data, variant_type)

+    def _write_reconnect_in_progress(
+        self, metadata: dict[str, Any]
+    ) -> tuple[bool, dict[str, Any]]:
+        """
+        Fail a write because a background reconnect thread is already running.
+
+        Args:
+            metadata (dict[str, Any]): Write context passed through to the activity.
+
+        Return:
+            tuple[bool, dict[str, Any]]: (False, error info with opc_error_kind reconnect_in_progress).
+        """
+        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',
+        }
+
+    def _write_connection_lost(
+        self, metadata: dict[str, Any], opc_status: str
+    ) -> tuple[bool, dict[str, Any]]:
+        """
+        Fail a write after scheduling reconnect for a closed or stale session.
+
+        Args:
+            metadata (dict[str, Any]): Write context passed through to the activity.
+            opc_status (str): Synthetic reason (ProtocolClosed, SessionNotReady).
+
+        Return:
+            tuple[bool, dict[str, Any]]: (False, error info with opc_error_kind connection_lost).
+        """
+        self._emit_opc_write_metric('unknown', opc_status, metadata)
+        return False, self._write_failure_payload(
+            notification_id=f'OPC_WRITE_CONNECTION_LOST_{self.id}',
+            message=f'OPC write skipped: connection lost ({opc_status}) | metadata: {metadata}',
+            level=NotificationLevel.WARNING,
+            opc_error_kind='connection_lost',
+            opc_status=opc_status,
+        )
+
    def write_data(
        self, node: str, value: Any, data_type: str, metadata: dict[str, Any]
    ) -> tuple[bool, dict[str, Any]]:
        """
-        Write data to OPC server with a single attempt and Tier-1 Bad* reconnect scheduling.
-        """
-        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',
-            }
+        Write data to OPC server with a single attempt and background reconnect scheduling.

-        is_connected, error = self.validate_connection()
+        Reconnect is scheduled on Tier-1 Bad*, closed protocol, or stale session readiness.
+        There is no retry within the same call.
+
+        Args:
+            node (str): OPC UA node id to write.
+            value (Any): Value to convert and send.
+            data_type (str): Logical type key (float, int, bool, str, double).
+            metadata (dict[str, Any]): Activity context (model_id, model_name, etc.).
+
+        Return:
+            tuple[bool, dict[str, Any]]: (True, {response_time}) on success, or
+                (False, structured error info) on failure.
+        """
+        if self._reconnect_thread_in_progress():
+            return self._write_reconnect_in_progress(metadata)
+
+        if not self._session_ready.is_set():
+            session_id = _opc_authentication_token_str(self.client)
+            self._start_reconnect('SessionNotReady', session_id)
+            if self._reconnect_thread_in_progress():
+                return self._write_reconnect_in_progress(metadata)
+            return self._write_connection_lost(metadata, 'SessionNotReady')
+
+        is_connected, _error = self.validate_connection()
        if not is_connected:
-            self._emit_opc_write_metric('unknown', 'NotConnected', metadata)
-            return False, error
+            session_id = _opc_authentication_token_str(self.client)
+            self._start_reconnect('ProtocolClosed', session_id)
+            return self._write_connection_lost(metadata, 'ProtocolClosed')

        start_time = time.time()
        session_id = _opc_authentication_token_str(self.client)
--- a/laborious/worker/worker.py
+++ b/laborious/worker/worker.py
@@ -65,7 +65,7 @@ with workflow.unsafe.imports_passed_through():
    )
    from laborious.workflows.sub_workflows.prediction_process import PredictionProcess

-POD_ID = os.getenv('POD_ID')
+POD_ID = os.getenv('HOSTNAME')
 SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', '9091'))


--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -18,3 +18,4 @@ testcontainers[postgres,minio] # PostgreSQL and MinIO containers for E2E tests
 # Development Tools
 ipython>=8.12.0          # Enhanced Python shell
 ipdb>=0.13.13            # IPython debugger
+ipykernel==6.30.1        # IPython kernel for Jupyter notebooks
--- a/tests/laborious/utils/repository/test_opc_repository.py
+++ b/tests/laborious/utils/repository/test_opc_repository.py
@@ -253,6 +253,7 @@ def test_disconnect(opc_repository, mock_client):

    opc_repository._disconnection_fallback.assert_called_once()
    assert opc_repository.client is None
+    assert opc_repository._allow_reconnect is False


 def test_disconnect_no_client(opc_repository):
@@ -323,12 +324,19 @@ def test_write_data_validate_connection_do_nothing(opc_repository):
 def test_write_data_validate_connection_failed(opc_repository):
    opc_repository.validate_connection = MagicMock(return_value=(False, {}))
    opc_repository.client = MagicMock()
+    opc_repository._start_reconnect = MagicMock()

-    result = opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata'])
+    is_success, error_data = opc_repository.write_data(
+        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
+    )

    opc_repository.validate_connection.assert_called_once()
    opc_repository.client.get_node.assert_not_called()
-    assert result == (False, {})
+    opc_repository._start_reconnect.assert_called_once()
+    assert opc_repository._start_reconnect.call_args.args[0] == 'ProtocolClosed'
+    assert is_success is False
+    assert error_data['opc_error_kind'] == 'connection_lost'
+    assert error_data['opc_status'] == 'ProtocolClosed'


 def test_write_data_get_node_failed(opc_repository):
@@ -424,7 +432,7 @@ def test_is_reconnectable_opcua_bad():
 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()
+    opc_repository._start_reconnect = MagicMock()
    mock_node = MagicMock()
    mock_client.get_node = MagicMock(return_value=mock_node)
    mock_node.write_value.side_effect = BadSessionIdInvalid()
@@ -434,7 +442,7 @@ def test_write_data_bad_session_id_invalid_schedules_reconnect(opc_repository, m
    )

    mock_node.write_value.assert_called_once()
-    opc_repository._start_reconnect_on_bad.assert_called_once()
+    opc_repository._start_reconnect.assert_called_once()
    assert is_success is False
    assert error_data['opc_error_kind'] == 'session_bad'
    assert error_data['opc_status'] == 'BadSessionIdInvalid'
@@ -455,15 +463,83 @@ def test_write_data_reconnect_in_progress_immediate(opc_repository):
    assert error_data['opc_error_kind'] == 'reconnect_in_progress'


-def test_start_reconnect_on_bad_skips_within_interval(opc_repository):
+def test_start_reconnect_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')
+    opc_repository._start_reconnect('BadSessionIdInvalid', 'tok')

    assert opc_repository._reconnect_thread is None


+def test_write_data_protocol_closed_schedules_reconnect(opc_repository):
+    opc_repository.client = MagicMock()
+    opc_repository.client.aio_obj.uaclient.protocol = MagicMock(state='closed')
+    opc_repository._start_reconnect = MagicMock()
+
+    is_success, error_data = opc_repository.write_data(
+        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
+    )
+
+    opc_repository._start_reconnect.assert_called_once()
+    assert opc_repository._start_reconnect.call_args.args[0] == 'ProtocolClosed'
+    assert is_success is False
+    assert error_data['opc_error_kind'] == 'connection_lost'
+    assert error_data['opc_status'] == 'ProtocolClosed'
+
+
+def test_write_data_protocol_closed_skips_reconnect_within_interval(opc_repository):
+    opc_repository.client = MagicMock()
+    opc_repository.client.aio_obj.uaclient.protocol = MagicMock(state='closed')
+    opc_repository.last_reconnection_time = datetime.now()
+    opc_repository.reconnection_interval = 3600
+
+    is_success, error_data = opc_repository.write_data(
+        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
+    )
+
+    assert opc_repository._reconnect_thread is None
+    assert is_success is False
+    assert error_data['opc_error_kind'] == 'connection_lost'
+
+
+def test_write_data_after_failed_reconnect_schedules_again(opc_repository):
+    opc_repository._session_ready.clear()
+    opc_repository.reconnection_interval = 0
+    opc_repository.last_reconnection_time = None
+    opc_repository._reconnect_locked = MagicMock(
+        return_value=(False, {'message': 'connect failed'})
+    )
+
+    opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata'])
+    if opc_repository._reconnect_thread is not None:
+        opc_repository._reconnect_thread.join(timeout=2)
+    assert opc_repository._reconnect_locked.call_count == 1
+    assert not opc_repository._reconnect_thread_in_progress()
+
+    opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', metadata['metadata'])
+    if opc_repository._reconnect_thread is not None:
+        opc_repository._reconnect_thread.join(timeout=2)
+    assert opc_repository._reconnect_locked.call_count == 2
+
+
+def test_write_data_after_disconnect_does_not_schedule_reconnect(opc_repository, mock_client):
+    opc_repository.client = mock_client
+    proto = MagicMock()
+    proto.state = 'closed'
+    mock_client.aio_obj.uaclient.protocol = proto
+    opc_repository._disconnection_fallback = MagicMock(return_value=[])
+    opc_repository.disconnect()
+
+    is_success, error_data = opc_repository.write_data(
+        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
+    )
+
+    assert opc_repository._reconnect_thread is None
+    assert is_success is False
+    assert error_data['opc_error_kind'] == 'connection_lost'
+
+
 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
@@ -472,7 +548,7 @@ def test_parallel_bad_writes_single_reconnect_task(opc_repository, mock_client):
    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()
+    opc_repository._start_reconnect = MagicMock()

    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        futures = [
@@ -487,7 +563,7 @@ def test_parallel_bad_writes_single_reconnect_task(opc_repository, mock_client):
        ]
        results = [future.result() for future in futures]

-    assert 1 <= opc_repository._start_reconnect_on_bad.call_count <= 2
+    assert 1 <= opc_repository._start_reconnect.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