From 34965c2b68c590da28bb740781a7ab0cae654566 Mon Sep 17 00:00:00 2001
From: JianYan <15005893916@163.com>
Date: Thu, 10 Sep 2026 14:34:19 +0800
Subject: [PATCH 1/4] fix(client): clean up provider when session startup fails
 (#1144)

---
 src/openenv/core/env_client.py         |  6 +++-
 tests/test_core/test_generic_client.py | 40 ++++++++++++++++++++++++--
 2 files changed, 42 insertions(+), 4 deletions(-)

diff --git a/src/openenv/core/env_client.py b/src/openenv/core/env_client.py
index 8e7f58ddc..85dbff71f 100644
--- a/src/openenv/core/env_client.py
+++ b/src/openenv/core/env_client.py
@@ -415,7 +415,11 @@ def _start_provider_if_needed(self) -> None:
         self._set_base_url(base_url)
 
     def _create_session_client(self) -> "EnvClient[Any, Any, Any]":
-        self._start_provider_if_needed()
+        try:
+            self._start_provider_if_needed()
+        except Exception:
+            self._stop_provider_best_effort()
+            raise
         if self._base_url is None:
             raise RuntimeError("EnvClient has no base URL.")
 
diff --git a/tests/test_core/test_generic_client.py b/tests/test_core/test_generic_client.py
index 7cec264a7..9e0539183 100644
--- a/tests/test_core/test_generic_client.py
+++ b/tests/test_core/test_generic_client.py
@@ -134,7 +134,10 @@ def stop_container(self):
             await client.connect()
 
     @pytest.mark.asyncio
-    async def test_new_session_reuses_provider_server(self, mock_provider):
+    @pytest.mark.parametrize("connect_parent", [False, True])
+    async def test_new_session_reuses_provider_server(
+        self, mock_provider, connect_parent
+    ):
         """Child sessions connect to the same server without owning the provider."""
         websockets = []
 
@@ -145,7 +148,8 @@ async def fake_ws_connect(*args, **kwargs):
 
         with patch("openenv.core.env_client.ws_connect", side_effect=fake_ws_connect):
             client = GenericEnvClient(provider=mock_provider)
-            await client.connect()
+            if connect_parent:
+                await client.connect()
             session = await client.new_session()
 
             assert isinstance(session, GenericEnvClient)
@@ -153,13 +157,43 @@ async def fake_ws_connect(*args, **kwargs):
             assert session._provider is None
             assert session._base_url == "http://localhost:8000"
             assert session._ws_url == "ws://localhost:8000/ws"
-            assert len(websockets) == 2
+            assert len(websockets) == 1 + int(connect_parent)
             mock_provider.start_container.assert_called_once_with()
 
             await client.close()
 
         mock_provider.stop_container.assert_called_once_with()
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "start_method,stop_method",
+        [("start_container", "stop_container"), ("start", "stop")],
+    )
+    @pytest.mark.parametrize("cleanup_fails", [False, True])
+    async def test_new_session_stops_provider_when_readiness_fails(
+        self, start_method, stop_method, cleanup_fails
+    ):
+        provider = Mock(spec=[start_method, stop_method, "wait_for_ready"])
+        start = getattr(provider, start_method)
+        stop = getattr(provider, stop_method)
+        start.return_value = "http://localhost:8000"
+        error = TimeoutError("environment never became ready")
+        provider.wait_for_ready.side_effect = error
+        if cleanup_fails:
+            stop.side_effect = RuntimeError("cleanup failed")
+        client = GenericEnvClient(provider=provider)
+
+        with patch("openenv.core.env_client.ws_connect", AsyncMock()) as connect:
+            with pytest.raises(TimeoutError) as exc_info:
+                await client.new_session()
+
+        assert exc_info.value is error
+        start.assert_called_once_with()
+        stop.assert_called_once_with()
+        connect.assert_not_called()
+        assert client.base_url is None
+        assert client._child_clients == []
+
     @pytest.mark.asyncio
     async def test_close_stops_provider_when_child_close_raises(self, mock_provider):
         """Parent cleanup continues even when a child session close fails."""

From 15773f3c9686d04ae978becad4995fee8bdc305b Mon Sep 17 00:00:00 2001
From: JianYan <15005893916@163.com>
Date: Fri, 18 Sep 2026 21:57:19 +0800
Subject: [PATCH 2/4] fix(client): make session provider cleanup retry-safe
 (#1144)

---
 src/openenv/core/env_client.py         |  97 ++++++++++--------
 tests/test_core/test_generic_client.py | 133 +++++++++++++++++++++++++
 2 files changed, 187 insertions(+), 43 deletions(-)

diff --git a/src/openenv/core/env_client.py b/src/openenv/core/env_client.py
index 85dbff71f..1a0e0f053 100644
--- a/src/openenv/core/env_client.py
+++ b/src/openenv/core/env_client.py
@@ -359,6 +359,7 @@ def __init__(
         self._websocket_ping_interval_s = websocket_ping_interval_s
         self._websocket_ping_timeout_s = websocket_ping_timeout_s
         self._provider = provider
+        self._provider_stopped = False
         self._start_provider_on_connect = base_url is None
         self._child_clients: list[EnvClient[Any, Any, Any]] = []
         self._ws: Optional[ClientConnection] = None
@@ -405,9 +406,11 @@ def _start_provider_if_needed(self) -> None:
                     f"{required}. Start the provider manually and pass base_url, "
                     "or configure a provider with a constructor-owned image/source."
                 )
+            self._provider_stopped = False
             base_url = self._provider.start_container()
             self._provider.wait_for_ready(base_url)
         elif hasattr(self._provider, "start"):
+            self._provider_stopped = False
             base_url = self._provider.start()
             self._provider.wait_for_ready()
         else:
@@ -415,35 +418,38 @@ def _start_provider_if_needed(self) -> None:
         self._set_base_url(base_url)
 
     def _create_session_client(self) -> "EnvClient[Any, Any, Any]":
+        # Match _start_provider_if_needed's startup condition. A provider with
+        # an existing URL may already be serving the parent or other children.
+        starts_provider_here = self._provider is not None and self._ws_url is None
         try:
             self._start_provider_if_needed()
+            if self._base_url is None:
+                raise RuntimeError("EnvClient has no base URL.")
+
+            signature = inspect.signature(type(self))
+            accepts_kwargs = any(
+                parameter.kind == inspect.Parameter.VAR_KEYWORD
+                for parameter in signature.parameters.values()
+            )
+            candidate_kwargs = {
+                "base_url": self._base_url,
+                "connect_timeout_s": self._connect_timeout,
+                "message_timeout_s": self._message_timeout,
+                "max_message_size_mb": self._max_message_size / (1024 * 1024),
+                "websocket_ping_interval_s": self._websocket_ping_interval_s,
+                "websocket_ping_timeout_s": self._websocket_ping_timeout_s,
+                "mode": self._mode,
+            }
+            constructor_kwargs = {}
+            for name, value in candidate_kwargs.items():
+                if accepts_kwargs or name in signature.parameters:
+                    constructor_kwargs[name] = value
+
+            return type(self)(**constructor_kwargs)
         except Exception:
-            self._stop_provider_best_effort()
+            if starts_provider_here:
+                self._stop_provider_best_effort()
             raise
-        if self._base_url is None:
-            raise RuntimeError("EnvClient has no base URL.")
-
-        signature = inspect.signature(type(self))
-        accepts_kwargs = any(
-            parameter.kind == inspect.Parameter.VAR_KEYWORD
-            for parameter in signature.parameters.values()
-        )
-        candidate_kwargs = {
-            "base_url": self._base_url,
-            "connect_timeout_s": self._connect_timeout,
-            "message_timeout_s": self._message_timeout,
-            "max_message_size_mb": self._max_message_size / (1024 * 1024),
-            "websocket_ping_interval_s": self._websocket_ping_interval_s,
-            "websocket_ping_timeout_s": self._websocket_ping_timeout_s,
-            "mode": self._mode,
-        }
-        constructor_kwargs = {}
-        for name, value in candidate_kwargs.items():
-            if accepts_kwargs or name in signature.parameters:
-                constructor_kwargs[name] = value
-
-        client = type(self)(**constructor_kwargs)
-        return client
 
     async def new_session(self) -> "EnvClient[Any, Any, Any]":
         """
@@ -1063,21 +1069,32 @@ async def _close_async(self) -> None:
                     if deferred_cancellation is None:
                         deferred_cancellation = exc
         finally:
-            try:
-                if self._provider is not None:
-                    # Handle both ContainerProvider and RuntimeProvider
-                    if hasattr(self._provider, "stop_container"):
-                        self._provider.stop_container()
-                    elif hasattr(self._provider, "stop"):
-                        self._provider.stop()
-            finally:
-                if self._start_provider_on_connect:
-                    self._base_url = None
-                    self._ws_url = None
+            self._stop_provider()
 
         if deferred_cancellation is not None:
             raise deferred_cancellation
 
+    def _stop_provider(self) -> None:
+        """Stop the provider once, retaining it for retries and later startup."""
+        try:
+            provider = self._provider
+            if provider is None or self._provider_stopped:
+                return
+
+            if hasattr(provider, "stop_container"):
+                provider.stop_container()
+            elif hasattr(provider, "stop"):
+                provider.stop()
+            else:
+                return
+
+            # Only a successful stop discharges our cleanup responsibility.
+            self._provider_stopped = True
+        finally:
+            if self._start_provider_on_connect:
+                self._base_url = None
+                self._ws_url = None
+
     def _stop_provider_best_effort(self) -> None:
         """Stop the underlying provider directly, ignoring any errors.
 
@@ -1086,14 +1103,8 @@ def _stop_provider_best_effort(self) -> None:
         after the provider started but before the connection is established, so
         routing cleanup through the (possibly broken) sync loop is not an option.
         """
-        provider = self._provider
-        if provider is None:
-            return
         with suppress(Exception):
-            if hasattr(provider, "stop_container"):
-                provider.stop_container()
-            elif hasattr(provider, "stop"):
-                provider.stop()
+            self._stop_provider()
 
     async def __aenter__(self) -> "EnvClient":
         """Enter async context manager, ensuring connection is established."""
diff --git a/tests/test_core/test_generic_client.py b/tests/test_core/test_generic_client.py
index 9e0539183..9004faa29 100644
--- a/tests/test_core/test_generic_client.py
+++ b/tests/test_core/test_generic_client.py
@@ -54,6 +54,139 @@ def mock_provider():
 # ============================================================================
 
 
+@pytest.fixture(params=[("start_container", "stop_container"), ("start", "stop")])
+def session_provider(request):
+    start_name, stop_name = request.param
+    provider = Mock(spec=[start_name, stop_name, "wait_for_ready"])
+    start = getattr(provider, start_name)
+    stop = getattr(provider, stop_name)
+    start.return_value = "http://localhost:8000"
+    return provider, start, stop
+
+
+class FailingSessionClient(GenericEnvClient):
+    def __init__(self, base_url=None, provider=None, **kwargs):
+        if base_url is not None:
+            raise ValueError("child constructor failed")
+        super().__init__(base_url=base_url, provider=provider, **kwargs)
+
+
+class TestSessionProviderCleanup:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("failure", ["readiness", "constructor"])
+    async def test_failed_session_is_cleaned_up_only_once(
+        self, session_provider, failure
+    ):
+        provider, start, stop = session_provider
+        stop.side_effect = [None, RuntimeError("already stopped")]
+        if failure == "readiness":
+            error = TimeoutError("not ready")
+            provider.wait_for_ready.side_effect = error
+            client = GenericEnvClient(provider=provider)
+        else:
+            client = FailingSessionClient(provider=provider)
+
+        with patch("openenv.core.env_client.ws_connect", AsyncMock()) as connect:
+            with pytest.raises((TimeoutError, ValueError)) as caught:
+                await client.new_session()
+
+        if failure == "readiness":
+            assert caught.value is error
+        else:
+            assert str(caught.value) == "child constructor failed"
+        start.assert_called_once_with()
+        stop.assert_called_once_with()
+        connect.assert_not_called()
+        assert client.base_url is None
+        assert client._ws_url is None
+        assert client._child_clients == []
+        await client.close()
+        await client.close()
+        stop.assert_called_once_with()
+
+    @pytest.mark.asyncio
+    async def test_failed_cleanup_can_be_retried(self, session_provider):
+        provider, start, stop = session_provider
+        error = TimeoutError("not ready")
+        provider.wait_for_ready.side_effect = error
+        stop.side_effect = [RuntimeError("cleanup failed"), None]
+        client = GenericEnvClient(provider=provider)
+        with pytest.raises(TimeoutError) as caught:
+            await client.new_session()
+        assert caught.value is error
+        assert client._provider is provider
+        stop.assert_called_once_with()
+        await client.close()
+        assert stop.call_count == 2
+        await client.close()
+        assert stop.call_count == 2
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("connect_parent", [False, True])
+    async def test_constructor_failure_preserves_running_provider(
+        self, session_provider, connect_parent
+    ):
+        provider, start, stop = session_provider
+        parent = FailingSessionClient(provider=provider)
+        ws = AsyncMock()
+        with patch("openenv.core.env_client.ws_connect", AsyncMock(return_value=ws)):
+            if connect_parent:
+                await parent.connect()
+            else:
+                parent._start_provider_if_needed()
+            with pytest.raises(ValueError, match="child constructor failed"):
+                await parent.new_session()
+            start.assert_called_once_with()
+            stop.assert_not_called()
+            assert parent.base_url == "http://localhost:8000"
+            if connect_parent:
+                assert parent._ws is ws
+                ws.close.assert_not_called()
+            await parent.close()
+        stop.assert_called_once_with()
+
+    @pytest.mark.asyncio
+    async def test_provider_can_restart_after_successful_cleanup(
+        self, session_provider
+    ):
+        provider, start, stop = session_provider
+        provider.wait_for_ready.side_effect = TimeoutError("not ready")
+        parent = GenericEnvClient(provider=provider)
+        with pytest.raises(TimeoutError):
+            await parent.new_session()
+        stop.assert_called_once_with()
+        provider.wait_for_ready.side_effect = None
+        with patch(
+            "openenv.core.env_client.ws_connect",
+            AsyncMock(return_value=AsyncMock()),
+        ):
+            child = await parent.new_session()
+            assert child.base_url == "http://localhost:8000"
+            assert start.call_count == 2
+            await parent.close()
+        assert stop.call_count == 2
+
+    @pytest.mark.parametrize("failure", ["readiness", "constructor"])
+    def test_sync_session_failure_is_cleaned_up_only_once(
+        self, session_provider, failure
+    ):
+        provider, start, stop = session_provider
+        stop.side_effect = [None, RuntimeError("already stopped")]
+        if failure == "readiness":
+            provider.wait_for_ready.side_effect = TimeoutError("not ready")
+            parent = GenericEnvClient(provider=provider)
+        else:
+            parent = FailingSessionClient(provider=provider)
+        client = parent.sync()
+        try:
+            with pytest.raises((TimeoutError, ValueError)):
+                client.new_session()
+            stop.assert_called_once_with()
+        finally:
+            client.close()
+        stop.assert_called_once_with()
+
+
 class TestGenericEnvClientInstantiation:
     """Test GenericEnvClient instantiation."""
 

From ba88f81b888e42334baf281ec9a713f4d1d0d810 Mon Sep 17 00:00:00 2001
From: JianYan <15005893916@163.com>
Date: Fri, 18 Sep 2026 22:22:43 +0800
Subject: [PATCH 3/4] fix(client): block restart until provider cleanup
 succeeds

Prevent a failed provider stop from being followed by another start that can overwrite the only resource handle. Add state-machine and real Docker regression coverage for sync and async lifecycle recovery.
---
 src/openenv/core/env_client.py                |  20 +-
 .../test_core/test_client_provider_cleanup.py | 296 ++++++++++++++++++
 .../test_client_provider_cleanup_docker.py    | 167 ++++++++++
 3 files changed, 479 insertions(+), 4 deletions(-)
 create mode 100644 tests/test_core/test_client_provider_cleanup.py
 create mode 100644 tests/test_core/test_client_provider_cleanup_docker.py

diff --git a/src/openenv/core/env_client.py b/src/openenv/core/env_client.py
index 1a0e0f053..acd8069f0 100644
--- a/src/openenv/core/env_client.py
+++ b/src/openenv/core/env_client.py
@@ -360,6 +360,7 @@ def __init__(
         self._websocket_ping_timeout_s = websocket_ping_timeout_s
         self._provider = provider
         self._provider_stopped = False
+        self._provider_cleanup_pending = False
         self._start_provider_on_connect = base_url is None
         self._child_clients: list[EnvClient[Any, Any, Any]] = []
         self._ws: Optional[ClientConnection] = None
@@ -392,6 +393,10 @@ def _set_base_url(self, base_url: str) -> None:
         self._ws_url = f"{ws_url}/ws"
 
     def _start_provider_if_needed(self) -> None:
+        # A missing URL does not mean the previous resource was released.
+        # Retry its cleanup before start can overwrite the provider's handle.
+        if self._provider_cleanup_pending:
+            self._stop_provider()
         if self._ws_url is not None:
             return
         if self._provider is None:
@@ -447,7 +452,7 @@ def _create_session_client(self) -> "EnvClient[Any, Any, Any]":
 
             return type(self)(**constructor_kwargs)
         except Exception:
-            if starts_provider_here:
+            if starts_provider_here and not self._provider_cleanup_pending:
                 self._stop_provider_best_effort()
             raise
 
@@ -567,7 +572,11 @@ async def _connect_async(self) -> "EnvClient":
         try:
             self._start_provider_if_needed()
         except Exception:
-            await self.close()
+            # A failed cleanup retry must propagate without attempting stop
+            # again. For a new startup failure, preserve its original error.
+            if not self._provider_cleanup_pending:
+                with suppress(Exception):
+                    await self.close()
             raise
 
         assert self._ws_url is not None
@@ -1082,13 +1091,16 @@ def _stop_provider(self) -> None:
                 return
 
             if hasattr(provider, "stop_container"):
-                provider.stop_container()
+                stop = provider.stop_container
             elif hasattr(provider, "stop"):
-                provider.stop()
+                stop = provider.stop
             else:
                 return
 
+            self._provider_cleanup_pending = True
+            stop()
             # Only a successful stop discharges our cleanup responsibility.
+            self._provider_cleanup_pending = False
             self._provider_stopped = True
         finally:
             if self._start_provider_on_connect:
diff --git a/tests/test_core/test_client_provider_cleanup.py b/tests/test_core/test_client_provider_cleanup.py
new file mode 100644
index 000000000..dfd92ce56
--- /dev/null
+++ b/tests/test_core/test_client_provider_cleanup.py
@@ -0,0 +1,296 @@
+# SPDX-License-Identifier: BSD-3-Clause
+
+"""Provider cleanup retries must not lose ownership of an allocated resource."""
+
+import json
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+from openenv.core.generic_client import GenericEnvClient
+from websockets.protocol import State
+
+
+class ResourceLedger:
+    """Model a provider whose next start overwrites its sole resource handle."""
+
+    def __init__(self, start_name, stop_name):
+        self.events = []
+        self.live_resources = set()
+        self.current_resource = None
+        self.allocations = 0
+        self.start_error = None
+        self.readiness_error = None
+        self.stop_error = None
+        self.provider = Mock(spec=[start_name, stop_name, "wait_for_ready"])
+        getattr(self.provider, start_name).side_effect = self.start
+        getattr(self.provider, stop_name).side_effect = self.stop
+        self.provider.wait_for_ready.side_effect = self.wait_for_ready
+
+    def start(self):
+        self.allocations += 1
+        self.current_resource = self.allocations
+        self.live_resources.add(self.current_resource)
+        self.events.append(("start", self.current_resource))
+        if self.start_error is not None:
+            # A provider can acquire a resource before startup raises.
+            raise self.start_error
+        return f"http://localhost:{8000 + self.current_resource}"
+
+    def wait_for_ready(self, *args):
+        self.events.append(("ready", self.current_resource))
+        if self.readiness_error is not None:
+            raise self.readiness_error
+
+    def stop(self):
+        self.events.append(("stop", self.current_resource))
+        if self.stop_error is not None:
+            raise self.stop_error
+        # An accidental repeated stop fails, as with non-idempotent providers.
+        self.live_resources.remove(self.current_resource)
+
+
+@pytest.fixture(params=[("start_container", "stop_container"), ("start", "stop")])
+def ledger(request):
+    return ResourceLedger(*request.param)
+
+
+@pytest.fixture(params=["async", "sync"])
+def execution_mode(request):
+    return request.param
+
+
+@pytest.fixture
+def client_type():
+    class SessionClient(GenericEnvClient):
+        child_error = None
+
+        def __init__(self, base_url=None, provider=None, **kwargs):
+            if base_url is not None and self.child_error is not None:
+                raise self.child_error
+            super().__init__(base_url=base_url, provider=provider, **kwargs)
+
+    return SessionClient
+
+
+@pytest.fixture
+def websocket_connect(monkeypatch, ledger):
+    async def connect(*args, **kwargs):
+        websocket = Mock(spec=["send", "recv", "close", "state"])
+        websocket.state = State.OPEN
+        websocket.send = AsyncMock()
+        websocket.recv = AsyncMock(
+            return_value=json.dumps(
+                {
+                    "type": "response",
+                    "data": {
+                        "observation": {"resource": ledger.current_resource},
+                        "done": False,
+                    },
+                }
+            )
+        )
+        websocket.close = AsyncMock()
+        return websocket
+
+    mocked_connect = AsyncMock(side_effect=connect)
+    monkeypatch.setattr("openenv.core.env_client.ws_connect", mocked_connect)
+    return mocked_connect
+
+
+def make_client(client_type, ledger, execution_mode):
+    client = client_type(provider=ledger.provider)
+    return client.sync() if execution_mode == "sync" else client
+
+
+async def invoke(client, execution_mode, method, *args):
+    result = getattr(client, method)(*args)
+    return await result if execution_mode == "async" else result
+
+
+def fail_setup(ledger, client_type, failure):
+    error = ValueError(f"{failure} failed")
+    if failure == "start":
+        ledger.start_error = error
+    elif failure == "readiness":
+        ledger.readiness_error = error
+    else:
+        client_type.child_error = error
+    return error
+
+
+def allow_setup(ledger, client_type):
+    ledger.start_error = None
+    ledger.readiness_error = None
+    client_type.child_error = None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("failure", ["start", "readiness", "constructor"])
+@pytest.mark.parametrize("retry_method", ["connect", "new_session"])
+async def test_pending_cleanup_blocks_allocation_and_retries_stop_once(
+    ledger, execution_mode, client_type, websocket_connect, failure, retry_method
+):
+    client = make_client(client_type, ledger, execution_mode)
+    original_error = fail_setup(ledger, client_type, failure)
+    cleanup_error = RuntimeError("resource is still running")
+    ledger.stop_error = cleanup_error
+    try:
+        with pytest.raises(ValueError) as caught:
+            await invoke(client, execution_mode, "new_session")
+        assert caught.value is original_error
+        assert ledger.events.count(("stop", 1)) == 1
+        allow_setup(ledger, client_type)
+
+        # Repeated user retries may retry stop, but never allocate or connect.
+        for _ in range(2):
+            previous_events = list(ledger.events)
+            with pytest.raises(RuntimeError) as caught:
+                await invoke(client, execution_mode, retry_method)
+            assert caught.value is cleanup_error
+            assert ledger.events == previous_events + [("stop", 1)]
+            assert ledger.live_resources == {1}
+            websocket_connect.assert_not_called()
+    finally:
+        ledger.stop_error = None
+        await invoke(client, execution_mode, "close")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("failure", ["start", "readiness", "constructor"])
+async def test_recovered_cleanup_stops_original_resource_before_restart(
+    ledger, execution_mode, client_type, websocket_connect, failure
+):
+    client = make_client(client_type, ledger, execution_mode)
+    original_error = fail_setup(ledger, client_type, failure)
+    ledger.stop_error = RuntimeError("temporary cleanup failure")
+    try:
+        with pytest.raises(ValueError) as caught:
+            await invoke(client, execution_mode, "new_session")
+        assert caught.value is original_error
+        allow_setup(ledger, client_type)
+        ledger.stop_error = None
+        previous_events = list(ledger.events)
+
+        child = await invoke(client, execution_mode, "new_session")
+
+        assert ledger.events == previous_events + [
+            ("stop", 1),
+            ("start", 2),
+            ("ready", 2),
+        ]
+        assert ledger.live_resources == {2}
+        assert child.base_url == "http://localhost:8002"
+        websocket_connect.assert_awaited_once()
+        await invoke(client, execution_mode, "close")
+        await invoke(client, execution_mode, "close")
+        assert ledger.events.count(("stop", 2)) == 1
+        assert ledger.live_resources == set()
+    finally:
+        ledger.stop_error = None
+        await invoke(client, execution_mode, "close")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("retry_method", ["connect", "new_session"])
+async def test_failed_explicit_close_blocks_restart_until_cleanup_succeeds(
+    ledger, execution_mode, client_type, websocket_connect, retry_method
+):
+    client = make_client(client_type, ledger, execution_mode)
+    cleanup_error = RuntimeError("close failed")
+    try:
+        await invoke(client, execution_mode, "connect")
+        ledger.stop_error = cleanup_error
+        with pytest.raises(RuntimeError) as caught:
+            await invoke(client, execution_mode, "close")
+        assert caught.value is cleanup_error
+        previous_events = list(ledger.events)
+
+        with pytest.raises(RuntimeError) as caught:
+            await invoke(client, execution_mode, retry_method)
+        assert caught.value is cleanup_error
+        assert ledger.events == previous_events + [("stop", 1)]
+        assert websocket_connect.await_count == 1
+        assert ledger.live_resources == {1}
+
+        ledger.stop_error = None
+        previous_events = list(ledger.events)
+        await invoke(client, execution_mode, retry_method)
+        assert ledger.events == previous_events + [
+            ("stop", 1),
+            ("start", 2),
+            ("ready", 2),
+        ]
+        await invoke(client, execution_mode, "close")
+        assert ledger.live_resources == set()
+    finally:
+        ledger.stop_error = None
+        await invoke(client, execution_mode, "close")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("failure", ["start", "readiness"])
+async def test_connect_preserves_setup_error_when_cleanup_also_fails(
+    ledger, execution_mode, client_type, websocket_connect, failure
+):
+    client = make_client(client_type, ledger, execution_mode)
+    original_error = fail_setup(ledger, client_type, failure)
+    ledger.stop_error = RuntimeError("cleanup failed")
+    try:
+        with pytest.raises(ValueError) as caught:
+            await invoke(client, execution_mode, "connect")
+        assert caught.value is original_error
+        assert ledger.events.count(("stop", 1)) == 1
+        assert ledger.live_resources == {1}
+        websocket_connect.assert_not_called()
+    finally:
+        ledger.stop_error = None
+        await invoke(client, execution_mode, "close")
+
+
+@pytest.mark.asyncio
+async def test_sibling_constructor_failure_keeps_live_child_usable(
+    ledger, execution_mode, client_type, websocket_connect
+):
+    client = make_client(client_type, ledger, execution_mode)
+    try:
+        # The provider is owned by a parent with no WebSocket of its own.
+        child = await invoke(client, execution_mode, "new_session")
+        original_error = fail_setup(ledger, client_type, "constructor")
+        with pytest.raises(ValueError) as caught:
+            await invoke(client, execution_mode, "new_session")
+        assert caught.value is original_error
+        assert ledger.events == [("start", 1), ("ready", 1)]
+        result = await invoke(child, execution_mode, "step", {})
+        assert result.observation == {"resource": 1}
+        websocket_connect.assert_awaited_once()
+        await invoke(client, execution_mode, "close")
+        assert ledger.live_resources == set()
+    finally:
+        await invoke(client, execution_mode, "close")
+
+
+@pytest.mark.asyncio
+async def test_multiple_failure_recovery_cycles_leave_no_orphaned_resources(
+    ledger, execution_mode, client_type, websocket_connect
+):
+    client = make_client(client_type, ledger, execution_mode)
+    try:
+        for cycle, failure in enumerate(["start", "readiness", "constructor"]):
+            original_error = fail_setup(ledger, client_type, failure)
+            ledger.stop_error = RuntimeError("cleanup temporarily unavailable")
+            with pytest.raises(ValueError) as caught:
+                await invoke(client, execution_mode, "new_session")
+            assert caught.value is original_error
+            allow_setup(ledger, client_type)
+            ledger.stop_error = None
+
+            await invoke(client, execution_mode, "new_session")
+            current_resource = 2 * cycle + 2
+            assert ledger.live_resources == {current_resource}
+            await invoke(client, execution_mode, "close")
+            await invoke(client, execution_mode, "close")
+            assert ledger.live_resources == set()
+            assert ledger.events.count(("stop", current_resource)) == 1
+    finally:
+        ledger.stop_error = None
+        await invoke(client, execution_mode, "close")
diff --git a/tests/test_core/test_client_provider_cleanup_docker.py b/tests/test_core/test_client_provider_cleanup_docker.py
new file mode 100644
index 000000000..46efba70e
--- /dev/null
+++ b/tests/test_core/test_client_provider_cleanup_docker.py
@@ -0,0 +1,167 @@
+# SPDX-License-Identifier: BSD-3-Clause
+
+"""Opt-in EnvClient cleanup tests with real, fault-injected Docker resources.
+
+Run with OPENENV_DOCKER_INTEGRATION=1 and a locally cached Alpine image:
+    PYTHONPATH=src:envs uv run pytest tests/test_core/test_client_provider_cleanup_docker.py -v
+
+OPENENV_DOCKER_TEST_IMAGE defaults to alpine:latest. No images are pulled,
+ports published, or host directories mounted. These tests verify container
+ownership and cleanup, not the environment server or WebSocket protocol.
+"""
+
+import os
+import subprocess
+from unittest.mock import AsyncMock, patch
+from uuid import uuid4
+
+import pytest
+from openenv.core.containers.runtime.providers import ContainerProvider
+from openenv.core.generic_client import GenericEnvClient
+
+pytestmark = [
+    pytest.mark.docker,
+    pytest.mark.integration,
+    pytest.mark.skipif(
+        os.environ.get("OPENENV_DOCKER_INTEGRATION") != "1",
+        reason="Set OPENENV_DOCKER_INTEGRATION=1 to run real Docker lifecycle tests",
+    ),
+]
+
+
+def docker(*args):
+    return subprocess.run(
+        ["docker", *args],
+        capture_output=True,
+        text=True,
+        check=True,
+        timeout=30,
+    ).stdout.strip()
+
+
+class DockerLifecycleProvider(ContainerProvider):
+    """Single-handle provider with injectable readiness and stop failures."""
+
+    def __init__(self, image):
+        self.image = image
+        self.label = f"openenv-cleanup-test={uuid4().hex}"
+        self.container_id = None
+        self.started = []
+        self.stop_calls = 0
+        self.fail_readiness = False
+        self.fail_stop = True
+
+    def start_container(self):
+        self.container_id = docker(
+            "run",
+            "--detach",
+            "--pull=never",
+            "--network=none",
+            "--label",
+            self.label,
+            self.image,
+            "sleep",
+            "300",
+        )
+        self.started.append(self.container_id)
+        # Every tested operation fails before opening a WebSocket.
+        return "http://localhost:8000"
+
+    def wait_for_ready(self, base_url, timeout_s=30.0):
+        assert (
+            docker("inspect", "--format={{.State.Running}}", self.container_id)
+            == "true"
+        )
+        if self.fail_readiness:
+            raise TimeoutError("injected readiness failure")
+
+    def stop_container(self):
+        self.stop_calls += 1
+        if self.fail_stop:
+            raise OSError("injected Docker cleanup failure")
+        docker("stop", "--time=1", self.container_id)
+        docker("rm", self.container_id)
+        self.container_id = None
+
+    def remaining_containers(self):
+        return docker(
+            "ps", "--all", "--quiet", "--no-trunc", "--filter", f"label={self.label}"
+        ).splitlines()
+
+
+@pytest.fixture
+def docker_provider():
+    image = os.environ.get("OPENENV_DOCKER_TEST_IMAGE", "alpine:latest")
+    # Opting in with a missing daemon/image is a failure, not a silent skip.
+    docker("image", "inspect", image)
+    provider = DockerLifecycleProvider(image)
+    try:
+        yield provider
+    finally:
+        # Independent of the provider's single handle: remove even resources
+        # orphaned by the regression, and only this fixture's unique label.
+        remaining = provider.remaining_containers()
+        if remaining:
+            docker("rm", "--force", *remaining)
+        assert provider.remaining_containers() == []
+
+
+class ConstructorFailureClient(GenericEnvClient):
+    def __init__(self, base_url=None, **kwargs):
+        if base_url is not None:
+            raise ValueError("injected constructor failure")
+        super().__init__(base_url=base_url, **kwargs)
+
+
+@pytest.mark.parametrize("failure", ["readiness", "constructor"])
+@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"])
+@pytest.mark.asyncio
+async def test_failed_cleanup_never_orphans_a_real_container(
+    docker_provider, failure, sync
+):
+    provider = docker_provider
+    provider.fail_readiness = failure == "readiness"
+    client_type = (
+        GenericEnvClient if provider.fail_readiness else ConstructorFailureClient
+    )
+    parent = client_type(provider=provider)
+    client = parent.sync() if sync else parent
+
+    async def invoke(method):
+        result = getattr(client, method)()
+        if not sync:
+            return await result
+        return result
+
+    with patch("openenv.core.env_client.ws_connect", AsyncMock()) as connect:
+        try:
+            with pytest.raises((TimeoutError, ValueError), match="injected"):
+                await invoke("new_session")
+            original = provider.container_id
+            assert provider.remaining_containers() == [original]
+
+            with pytest.raises(Exception) as retry_error:
+                await invoke("new_session")
+            assert provider.started == [original]
+            assert provider.stop_calls == 2
+            assert provider.remaining_containers() == [original]
+            assert str(retry_error.value) == "injected Docker cleanup failure"
+            assert isinstance(retry_error.value, OSError)
+
+            provider.fail_stop = False
+            with pytest.raises((TimeoutError, ValueError), match="injected"):
+                await invoke("new_session")
+            assert len(provider.started) == 2
+            assert provider.started[1] != original
+            assert provider.remaining_containers() == []
+            assert provider.stop_calls == 4
+
+            await invoke("close")
+            await invoke("close")
+            assert provider.stop_calls == 4
+            connect.assert_not_called()
+        finally:
+            provider.fail_stop = False
+            # On an unfixed client close can only release the last handle;
+            # the fixture finalizer independently catches older orphans.
+            await invoke("close")

From 5629394845a4c57371c0c4e67aa0bdf9a046701a Mon Sep 17 00:00:00 2001
From: JianYan <15005893916@163.com>
Date: Sun, 20 Sep 2026 15:53:19 +0800
Subject: [PATCH 4/4] fix(client): reject factory reconnect while cleanup is
 pending

---
 src/openenv/core/env_client.py                |  6 +++
 .../test_core/test_client_provider_cleanup.py | 42 ++++++++++++++++++-
 2 files changed, 47 insertions(+), 1 deletion(-)

diff --git a/src/openenv/core/env_client.py b/src/openenv/core/env_client.py
index acd8069f0..31298cada 100644
--- a/src/openenv/core/env_client.py
+++ b/src/openenv/core/env_client.py
@@ -396,6 +396,12 @@ def _start_provider_if_needed(self) -> None:
         # A missing URL does not mean the previous resource was released.
         # Retry its cleanup before start can overwrite the provider's handle.
         if self._provider_cleanup_pending:
+            if not self._start_provider_on_connect:
+                raise RuntimeError(
+                    "Provider cleanup is pending for this client with an existing "
+                    "base URL. Retry close() to finish cleanup, then create a new "
+                    "client instead of reconnecting to the old URL."
+                )
             self._stop_provider()
         if self._ws_url is not None:
             return
diff --git a/tests/test_core/test_client_provider_cleanup.py b/tests/test_core/test_client_provider_cleanup.py
index dfd92ce56..03437f323 100644
--- a/tests/test_core/test_client_provider_cleanup.py
+++ b/tests/test_core/test_client_provider_cleanup.py
@@ -26,7 +26,7 @@ def __init__(self, start_name, stop_name):
         getattr(self.provider, stop_name).side_effect = self.stop
         self.provider.wait_for_ready.side_effect = self.wait_for_ready
 
-    def start(self):
+    def start(self, *args, **kwargs):
         self.allocations += 1
         self.current_resource = self.allocations
         self.live_resources.add(self.current_resource)
@@ -294,3 +294,43 @@ async def test_multiple_failure_recovery_cycles_leave_no_orphaned_resources(
     finally:
         ledger.stop_error = None
         await invoke(client, execution_mode, "close")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("retry_method", ["connect", "new_session"])
+@pytest.mark.parametrize("cleanup_recovers", [False, True])
+async def test_factory_pending_cleanup_requires_explicit_close(
+    ledger, execution_mode, websocket_connect, retry_method, cleanup_recovers
+):
+    bootstrap = GenericEnvClient.from_env(
+        "audit/test-environment",
+        provider=ledger.provider,
+        use_docker=hasattr(ledger.provider, "start_container"),
+    )
+    client = bootstrap.sync() if execution_mode == "sync" else await bootstrap
+    try:
+        ledger.stop_error = RuntimeError("cleanup temporarily unavailable")
+        with pytest.raises(RuntimeError, match="cleanup temporarily unavailable"):
+            await invoke(client, execution_mode, "close")
+        if cleanup_recovers:
+            ledger.stop_error = None
+        previous_events = list(ledger.events)
+        websocket_connect.reset_mock()
+
+        # A factory URL cannot be reused after an implicit cleanup retry.
+        # Refuse before touching either the provider or the cached endpoint.
+        for _ in range(2):
+            with pytest.raises(RuntimeError, match=r"close\(\)"):
+                await invoke(client, execution_mode, retry_method)
+        assert ledger.events == previous_events
+        assert ledger.live_resources == {1}
+        websocket_connect.assert_not_called()
+
+        ledger.stop_error = None
+        await invoke(client, execution_mode, "close")
+        await invoke(client, execution_mode, "close")
+        assert ledger.events == previous_events + [("stop", 1)]
+        assert not ledger.live_resources
+    finally:
+        ledger.stop_error = None
+        await invoke(client, execution_mode, "close")