diff --git a/loopx/control_plane/turn_driver/dsh_operation_host.py b/loopx/control_plane/turn_driver/dsh_operation_host.py
index 6cffedf6de..f6dcfdc9cc 100644
--- a/loopx/control_plane/turn_driver/dsh_operation_host.py
+++ b/loopx/control_plane/turn_driver/dsh_operation_host.py
@@ -122,6 +122,29 @@ def _digest(path: Path) -> str:
         return hashlib.file_digest(stream, "sha256").hexdigest()
 
 
+def _write_owned_operation_patch(module: Path, socket_path: Path) -> Path:
+    """Write paired loader files; the caller owns their directory and lifetime."""
+    wrapper = socket_path.parent / "operation.mjs"
+    wrapper.write_text(
+        "import {apply} from " + json.dumps(module.as_uri()) + ";\n"
+        "export const name='loopx-owned-operation';\n"
+        "export const inject=['agents','tools','sessionProjections'];\n"
+        "export function applyOwned(ctx) { apply(ctx,{socketPath:"
+        + json.dumps(str(socket_path))
+        + "}); }\n"
+        "export {applyOwned as apply};\n",
+        encoding="utf-8",
+    )
+    owned_patch = socket_path.parent / "operation.patch.yml"
+    owned_patch.write_text(
+        "- insert:\n    - id: loopx-owned-operation\n      name: "
+        + json.dumps(wrapper.as_uri())
+        + "\n",
+        encoding="utf-8",
+    )
+    return owned_patch
+
+
 def _run_dsh_operation_host(
     request: Mapping[str, Any],
     *,
@@ -296,24 +319,7 @@ def native_start(host_turn_id: str) -> None:
         )
         owner.connection = connection
         owner.session_id = session_id
-        wrapper = connection.path.parent / "operation.mjs"
-        wrapper.write_text(
-            "import {apply} from " + json.dumps(module.as_uri()) + ";\n"
-            "export const name='loopx-owned-operation';\n"
-            "export const inject=['agents','tools','sessionProjections'];\n"
-            "export function applyOwned(ctx) { apply(ctx,{socketPath:"
-            + json.dumps(str(connection.path))
-            + "}); }\n"
-            "export {applyOwned as apply};\n",
-            encoding="utf-8",
-        )
-        owned_patch = connection.path.parent / "operation.patch.yml"
-        owned_patch.write_text(
-            "- insert:\n    - id: loopx-owned-operation\n      name: "
-            + json.dumps(wrapper.as_uri())
-            + "\n",
-            encoding="utf-8",
-        )
+        owned_patch = _write_owned_operation_patch(module, connection.path)
         config = build_sdk_config(
             provider=provider,
             model=model,
diff --git a/tests/control_plane/test_dsh_operation_sdk.py b/tests/control_plane/test_dsh_operation_sdk.py
index 963abbcb26..ce9f655e15 100644
--- a/tests/control_plane/test_dsh_operation_sdk.py
+++ b/tests/control_plane/test_dsh_operation_sdk.py
@@ -8,6 +8,7 @@
 import json
 import hashlib
 import os
+import shutil
 import threading
 from contextlib import ExitStack
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@@ -117,6 +118,14 @@ def test_real_sdk_native_prepare_confirmation_consume_report_and_profile_revocat
     runtime_bin = os.environ.get("DSH_NATIVE_OPERATION_RUNTIME")
     if connection_lifetime != "retained" and not runtime_bin:
         pytest.skip("explicit resume-capable runtime required")
+    # Exercise URI/JSON escaping through the actual SDK loader, including
+    # the frozen-request and persisted-session paths, without changing code.
+    if request_transport == "frozen":
+        escaped_module = tmp_path / 'native module "quoted" 雪'
+        escaped_module.mkdir()
+        for source in Path(module).parent.glob("*.js"):
+            shutil.copy2(source, escaped_module / source.name)
+        module = str(escaped_module / Path(module).name)
     service, store = fixtures.service(tmp_path)
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     projection = build_todo_runtime_shadow_projection(
@@ -497,3 +506,54 @@ def fail(notification):
         assert not host.connection.path.exists()
         with pytest.raises(ValueError, match="closed"):
             host.run(request(), **options)
+
+
+@pytest.mark.parametrize("failed_artifact", ["operation.mjs", "operation.patch.yml"])
+def test_artifact_write_failure_closes_connection_before_sdk_launch(
+    tmp_path, monkeypatch, failed_artifact
+):
+    sdk = pytest.importorskip("deepseek_harness")
+    service, store = fixtures.service(tmp_path)
+    module = tmp_path / "native-operation.js"
+    module.write_text("export function apply() {}\n")
+    writes = []
+    original_write = Path.write_text
+
+    def write(path, *args, **kwargs):
+        if path.name in {"operation.mjs", "operation.patch.yml"}:
+            writes.append(path.name)
+            if path.name == failed_artifact:
+                raise OSError("fixture artifact write failed")
+        return original_write(path, *args, **kwargs)
+
+    def forbidden_launch(*_args, **_kwargs):
+        pytest.fail("artifact failure must precede SDK process launch")
+
+    monkeypatch.setattr(Path, "write_text", write)
+    monkeypatch.setattr(sdk, "DeepSeekHarness", forbidden_launch)
+    with OwnedDshOperationHost() as host:
+        with pytest.raises(OSError, match="fixture artifact write failed"):
+            host.run(
+                request(),
+                runtime_root=store.root.parent.parent,
+                registry_path=service.registry_path,
+                project=service.registry_path.parent.parent,
+                dsh_home=tmp_path / "sdk-home",
+                operation_module=module,
+                provider="deepseek-official",
+                model="fixture-model",
+                reasoning_effort="high",
+                max_tokens=128,
+                runtime_bin=(Path(os.environ["DSH_NATIVE_OPERATION_RUNTIME"])
+                    if os.environ.get("DSH_NATIVE_OPERATION_RUNTIME") else None),
+            )
+        assert writes == (["operation.mjs"] if failed_artifact == "operation.mjs"
+                          else ["operation.mjs", "operation.patch.yml"])
+        assert host.harness is None
+        assert not host.connection.path.exists()
+        assert not host.connection.path.parent.exists()
+        with pytest.raises(ValueError, match="closed"):
+            host.run(request())
+        assert load_codex_cli_session(
+            store.root.parent.parent, lineage=_lineage(request()), host="dsh"
+        ) is None