Replace-experiment-client-with-ert-storage-api

config.yml

@@ -0,0 +1,13 @@
+config_path: config.yml
+controls:
+  - max: 0.1
+    min: 0.0
+    name: my_control
+    perturbation_magnitude: 0.01
+    variables:
+      - {initial_guess: 0.1, name: test}
+model:
+  realizations: [0]
+  realizations_weights: [1.0]
+objective_functions:
+  - {name: my_objective}

everest_output/logs/everest-log-config-yml-2026-08-27T0849+0200.txt

@@ -0,0 +1,2 @@
+2026-08-27 08:49:56,002 - everest - MainThread - DEBUG - No definitions node found in configuration file
+2026-08-27 08:49:56,104 - everest - MainThread - DEBUG - No definitions node found in configuration file

everest_output/logs/everest-log-config-yml-2026-08-27T0856+0200.txt

@@ -0,0 +1,2 @@
+2026-08-27 08:56:16,747 - everest - MainThread - DEBUG - No definitions node found in configuration file
+2026-08-27 08:56:16,851 - everest - MainThread - DEBUG - No definitions node found in configuration file

src/ert/services/ert_client.py

@@ -2,7 +2,13 @@
 
 import io
 import json
+import logging
+import queue
+import ssl
 import threading
+import time
+import traceback
+from base64 import b64encode
 from collections import OrderedDict
 from collections.abc import Callable
 from copy import deepcopy
@@ -15,6 +21,16 @@
 import numpy as np
 import numpy.typing as npt
 import pandas as pd
+from pydantic import ValidationError
+from websockets.exceptions import ConnectionClosedError
+from websockets.sync.client import connect
+
+from _ert.threading import ErtThread
+from ert.run_models.event import (
+    StatusEvents,
+    status_event_from_json,
+)
+from everest.strings import EverEndpoints
 
 from .shared_client import ErtClientConnectionInfo, Methods, SharedClient
 
@@ -24,6 +40,8 @@
 _PARQUET = {"accept": "application/x-parquet"}
 _EXPERIMENT_SERVER = "/experiment_server"
 
+logger = logging.getLogger(__name__)
+
 
 def _escape(value: str) -> str:
     """Keys may contain slashes, and the server decodes the path segment once."""
@@ -141,7 +159,7 @@ def version(self) -> str:
         return str(self._get("/version").json())
 
     def experiments(self) -> list[dict[str, Any]]:
-        return list(self._get("/experiments").json())
+        return self._get("/experiments").json()
 
     def ensemble(self, ensemble_id: str) -> dict[str, Any]:
         return dict(self._get(f"/ensembles/{ensemble_id}").json())
@@ -231,41 +249,120 @@ def experiment_server_is_running(self) -> bool:
         return response.status_code == httpx.codes.OK
 
     def experiment_ids(self) -> list[str]:
-        response = self._experiment_server_get("experiments")
+        response = self._experiment_server_get(EverEndpoints.EXPERIMENTS)
         return list(response.json()["experiment_ids"])
 
     def experiment_status(self, experiment_id: str) -> dict[str, Any]:
-        return dict(self._experiment_server_get(f"status/{experiment_id}").json())
+        return dict(
+            self._experiment_server_get(
+                f"{EverEndpoints.STATUS}/{experiment_id}"
+            ).json()
+        )
 
-    def experiment_config_path(self, experiment_id: str) -> dict[str, Any]:
-        return dict(self._experiment_server_get(f"config_path/{experiment_id}").json())
+    def experiment_config(self, experiment_id: str) -> dict[str, str]:
+        return self._experiment_server_get(
+            f"{EverEndpoints.CONFIG_PATH}/{experiment_id}"
+        ).json()
 
     def experiment_start_time(self, experiment_id: str) -> int:
-        return int(self._experiment_server_get(f"start_time/{experiment_id}").text)
+        return int(
+            self._experiment_server_get(
+                f"{EverEndpoints.START_TIME}/{experiment_id}"
+            ).text
+        )
+
+    def setup_event_queue_from_ws_endpoint(
+        self,
+        experiment_id: str,
+        refresh_interval: float = 0.01,
+        open_timeout: float = 30,
+        websocket_recv_timeout: float = 1.0,
+    ) -> tuple[queue.SimpleQueue[StatusEvents], ErtThread]:
+        """Return a queue of experiment events and the thread that fills it.
+
+        The caller owns the thread and must start it.
+        """
+        event_queue: queue.SimpleQueue[StatusEvents] = queue.SimpleQueue()
+
+        url = (
+            self.conn_info.base_url.replace("https://", "wss://")
+            + f"{_EXPERIMENT_SERVER}/{EverEndpoints.EVENTS}/{experiment_id}"
+        )
+        username, password = self._auth
+        credentials = b64encode(f"{username}:{password}".encode()).decode()
+
+        def passthrough_ws_events() -> None:
+            try:  # ruff: ignore[too-many-statements-in-try-clause]
+                with connect(
+                    url,
+                    ssl=self._ssl_context,
+                    open_timeout=open_timeout,
+                    additional_headers={"Authorization": f"Basic {credentials}"},
+                ) as websocket:
+                    while True:
+                        try:
+                            message = websocket.recv(timeout=websocket_recv_timeout)
+                        except TimeoutError:
+                            message = None
+                        if message:
+                            try:
+                                event_queue.put(status_event_from_json(message))
+                            except ValidationError as e:
+                                logger.error(
+                                    "Error when processing event %s", exc_info=e
+                                )
+
+                        time.sleep(refresh_interval)
+            except ConnectionClosedError:
+                logger.debug("Connection closed by server")
+            except Exception:
+                logger.debug(traceback.format_exc())
+
+        monitor_thread = ErtThread(
+            name="ert_storage_api_event_monitor",
+            target=passthrough_ws_events,
+            daemon=True,
+        )
+
+        return event_queue, monitor_thread
 
     def start_experiment(self, config: dict[str, Any]) -> str:
         response = self._request(
             "POST",
-            f"{_EXPERIMENT_SERVER}/start_experiment",
+            f"{_EXPERIMENT_SERVER}/{EverEndpoints.START_EXPERIMENT}",
             auth=self._auth,
             json=config,
         )
         return str(_checked(response).json()["experiment_id"])
 
-    def stop_experiment_server(self) -> None:
-        _checked(self._request("POST", f"{_EXPERIMENT_SERVER}/stop", auth=self._auth))
+    def stop_experiment_server(self) -> bool:
+        return (
+            self._request(
+                "POST",
+                f"{_EXPERIMENT_SERVER}/{EverEndpoints.STOP}",
+                auth=self._auth,
+            ).status_code
+            == 200
+        )
 
     def runpath_exists(self, paths: list[str]) -> bool:
         response = self._request(
             "POST",
-            f"{_EXPERIMENT_SERVER}/runpath",
+            f"{_EXPERIMENT_SERVER}/{EverEndpoints.RUNPATH}",
             auth=self._auth,
             json={"paths": paths},
         )
         return response.status_code == httpx.codes.OK
 
     # <-------------- Internals -------------->
 
+    @property
+    def _ssl_context(self) -> ssl.SSLContext | None:
+        cert = self._client.conn_info.cert
+        if not isinstance(cert, str):
+            return None
+        return ssl.create_default_context(cafile=cert)
+
     @property
     def _auth(self) -> tuple[str, str]:
         """Experiment-server routes authenticate with HTTP Basic, not the token

src/everest/bin/everest_script.py

@@ -22,7 +22,6 @@
 from ert.utils import makedirs_if_needed
 from everest.config import EverestConfig, ServerConfig
 from everest.detached import (
-    start_experiment,
     start_server,
     wait_for_server,
 )
@@ -164,7 +163,7 @@ def _build_args_parser() -> argparse.ArgumentParser:
 async def run_everest(options: argparse.Namespace) -> None:
 
     try:
-        ErtClient.for_project(
+        ErtClient.get_client(
             Path(ServerConfig.get_session_dir(options.config.output_dir)),
             connect_timeout=1,
         )
@@ -228,7 +227,7 @@ async def directory_is_nonempty(path: Path) -> bool:
     print("Waiting for server ...")
     logger.debug("Waiting for response from everserver")
     wait_start_time: float = time.monotonic()
-    client = ErtClient.for_project(
+    client = ErtClient.get_client(
         Path(ServerConfig.get_session_dir(options.config.output_dir))
     )
     wait_for_server(client, timeout=600)
@@ -238,11 +237,7 @@ async def directory_is_nonempty(path: Path) -> bool:
         f"waiting for {time.monotonic() - wait_start_time:g} seconds. "
         "Starting experiment"
     )
-
-    experiment_id = start_experiment(
-        server_context=ServerConfig.get_server_context_from_conn_info(client.conn_info),
-        config=options.config,
-    )
+    experiment_id = client.start_experiment(options.config)
 
     # blocks until the run is finished
     if options.gui: