Merge pull request #1756 from debpalash/fix/windows-direct-job-owner-1734

fix(windows): remove the sidecar supervisor hop

CHANGELOG.md

@@ -28,6 +28,7 @@ the frozen-backend fallback mirror it for their toolchains.
 
 ### Fixed
 
+- Windows isolated engines now retain direct Job ownership without an extra Python supervisor process that can deadlock the child loader (#1734)
 - The setup splash now waits through the backend's full startup budget instead of reporting slow Windows CUDA initialization as stuck after two minutes (#1749)
 - Dubbing jobs can now reuse every source-language code produced by automatic ASR detection without a 400 error on the next upload (#1737)
 - Incomplete Sherpa-ONNX model snapshots now self-repair before recognizer startup instead of failing on a missing ONNX file (#1733)

backend/core/contained_subprocess.py

@@ -3,13 +3,14 @@
 The desktop owns the backend with an OS process group/Job.  Engine and
 installer operations also need an independently terminable subtree: killing
 only their direct child on a timeout leaves uv/git/model workers holding pipes
-and mutating files.  A small direct-child supervisor bridges both lifetimes.
+and mutating files.
 
-On POSIX the supervisor is the unreaped leader of a nested process group.  A
-control-pipe EOF (including kernel EOF when the backend dies) kills that group;
-the parent also drains the group before reaping its stable leader.  On Windows
-the supervisor assigns the operation, while suspended, to a nested
-kill-on-close Job.  The outer desktop Job still contains both levels.
+On POSIX a small supervisor is the unreaped leader of a nested process group.
+A control-pipe EOF (including kernel EOF when the backend dies) kills that
+group; the parent also drains the group before reaping its stable leader. On
+Windows the backend retains a nested kill-on-close Job directly and assigns
+the suspended operation before resuming it. The outer desktop Job remains the
+terminal fallback.
 
 Standalone/server launches use the same nested owner, preserving their
 independently terminable subtree without relying on ``taskkill`` or discovery.
@@ -263,42 +264,148 @@ class OwnedPopen:
                 pass
 
 
-def spawn_owned(argv: list[str], **kwargs: Any) -> "subprocess.Popen | OwnedPopen":
+class WindowsJobPopen:
+    """Popen-compatible handle whose child tree lives in a retained Job.
+
+    Windows Job handles already provide the stable ownership that POSIX needs
+    a supervisor process group for. Keeping the handle in the backend means an
+    abrupt backend exit closes it in the kernel and kills the whole operation
+    tree, without inserting a second Python process in the sidecar loader path
+    (#1734).
+    """
+
+    def __init__(self, proc: subprocess.Popen, job: Any, kernel32: Any) -> None:
+        self._proc = proc
+        self._job = job
+        self._kernel32 = kernel32
+        self._lock = threading.RLock()
+        self.stdin = proc.stdin
+        self.stdout = proc.stdout
+        self.stderr = proc.stderr
+
+    @property
+    def pid(self) -> int:
+        return self._proc.pid
+
+    @property
+    def args(self) -> Any:
+        return self._proc.args
+
+    @property
+    def returncode(self) -> Optional[int]:
+        return self._proc.returncode
+
+    def _close_job(self, *, terminate: bool) -> None:
+        job, self._job = self._job, None
+        if job is None:
+            return
+        try:
+            if terminate:
+                self._kernel32.TerminateJobObject(job, 1)
+        finally:
+            self._kernel32.CloseHandle(job)
+
+    def poll(self) -> Optional[int]:
+        with self._lock:
+            rc = self._proc.poll()
+            if rc is None:
+                return None
+            # A successful direct child may leave helpers behind. Match the
+            # supervisor contract by draining the retained Job before return.
+            self._close_job(terminate=True)
+            return rc
+
+    def wait(self, timeout: Optional[float] = None) -> int:
+        try:
+            rc = self._proc.wait(timeout=timeout)
+        except subprocess.TimeoutExpired:
+            raise
+        with self._lock:
+            self._close_job(terminate=True)
+        return rc
+
+    def terminate(self) -> None:
+        with self._lock:
+            self._close_job(terminate=True)
+
+    def kill(self) -> None:
+        self.terminate()
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._proc, name)
+
+    def __del__(self) -> None:
+        try:
+            self._close_job(terminate=True)
+        except Exception:
+            pass  # interpreter shutdown; closing the OS handle is best-effort
+
+
+def _spawn_windows_owned(argv: list[str], kwargs: dict[str, Any]) -> WindowsJobPopen:
+    """Start *argv* suspended, assign its tree to a Job, then resume it."""
+    import ctypes
+
+    job, kernel32, wintypes = _windows_job()
+    child: Optional[subprocess.Popen] = None
+    popen_kwargs = dict(kwargs)
+    supplied_env = popen_kwargs.get("env")
+    operation_env = dict(os.environ if supplied_env is None else supplied_env)
+    operation_env.pop(_DRAIN_FD_ENV, None)
+    operation_env.pop(_DESKTOP_MARKER, None)
+    popen_kwargs["env"] = operation_env
+    supplied_flags = int(popen_kwargs.pop("creationflags", 0))
+    popen_kwargs["creationflags"] = supplied_flags | 0x08000000 | 0x00000004
+    try:
+        child = subprocess.Popen(argv, **popen_kwargs)
+        assign = kernel32.AssignProcessToJobObject
+        assign.argtypes = (wintypes.HANDLE, wintypes.HANDLE)
+        assign.restype = wintypes.BOOL
+        if not assign(job, wintypes.HANDLE(child._handle)):
+            raise OSError(ctypes.get_last_error(), "AssignProcessToJobObject")
+        _resume_windows_process(kernel32, wintypes, child.pid)
+        return WindowsJobPopen(child, job, kernel32)
+    except BaseException:
+        kernel32.TerminateJobObject(job, 1)
+        if child is not None:
+            try:
+                child.kill()
+            except OSError:
+                pass  # the suspended child may already have exited
+            try:
+                child.wait(timeout=5)
+            except (OSError, subprocess.TimeoutExpired):
+                pass  # Job termination remains the authoritative cleanup
+        kernel32.CloseHandle(job)
+        raise
+
+
+def spawn_owned(
+    argv: list[str], **kwargs: Any
+) -> "subprocess.Popen | OwnedPopen | WindowsJobPopen":
     """Spawn an operation with a stable, independently terminable owner."""
 
-    drain_fd = backend_drain_fd(required=True) if os.name == "posix" else None
+    if os.name == "nt":
+        return _spawn_windows_owned(argv, kwargs)
+
+    drain_fd = backend_drain_fd(required=True)
     control_read, control_write = os.pipe()
     result_read, result_write = os.pipe()
-    control_token = control_read
-    result_token = result_write
-    if os.name == "nt":
-        import msvcrt
-
-        control_token = msvcrt.get_osfhandle(control_read)
-        result_token = msvcrt.get_osfhandle(result_write)
     wrapper_argv = _supervisor_argv(
-        control_token,
-        result_token,
+        control_read,
+        result_write,
         argv,
     )
     wrapper_kwargs = dict(kwargs)
-    if os.name == "posix":
-        wrapper_kwargs["start_new_session"] = True
-        pass_fds = [control_read, result_write]
-        if drain_fd is not None:
-            pass_fds.append(drain_fd)
-            if wrapper_kwargs.get("env") is not None:
-                wrapper_env = dict(wrapper_kwargs["env"])
-                wrapper_env[_DESKTOP_MARKER] = "1"
-                wrapper_env[_DRAIN_FD_ENV] = str(drain_fd)
-                wrapper_kwargs["env"] = wrapper_env
-        wrapper_kwargs["pass_fds"] = tuple(pass_fds)
-    else:
-        # Python's Windows fd inheritance requires inheritable CRT handles.
-        # All unrelated descriptors are non-inheritable by default (PEP 446).
-        os.set_handle_inheritable(control_token, True)
-        os.set_handle_inheritable(result_token, True)
-        wrapper_kwargs["close_fds"] = False
+    wrapper_kwargs["start_new_session"] = True
+    pass_fds = [control_read, result_write]
+    if drain_fd is not None:
+        pass_fds.append(drain_fd)
+        if wrapper_kwargs.get("env") is not None:
+            wrapper_env = dict(wrapper_kwargs["env"])
+            wrapper_env[_DESKTOP_MARKER] = "1"
+            wrapper_env[_DRAIN_FD_ENV] = str(drain_fd)
+            wrapper_kwargs["env"] = wrapper_env
+    wrapper_kwargs["pass_fds"] = tuple(pass_fds)
     try:
         proc = subprocess.Popen(wrapper_argv, **wrapper_kwargs)
     except BaseException:

backend/services/sidecar_install.py

@@ -60,7 +60,7 @@ from pathlib import Path
 from typing import Callable, Optional
 
 from core.config import DATA_DIR
-from core.contained_subprocess import OwnedPopen, spawn_owned
+from core.contained_subprocess import OwnedPopen, WindowsJobPopen, spawn_owned
 
 logger = logging.getLogger("omnivoice.sidecar_install")
 
@@ -1057,7 +1057,8 @@ def _run_logged(job: dict, argv: list[str], *, timeout: float,
     would hang past the timeout waiting for pipe EOF.
     """
     # ``spawn_owned`` creates the local timeout group/Job before the operation
-    # starts and links it to backend death through its control pipe.
+    # starts. POSIX links it to backend death through a control pipe; Windows
+    # retains a kill-on-close Job handle in this backend process.
     popen_kwargs = _install_containment_kwargs()
     try:
         proc = spawn_owned(
@@ -1096,14 +1097,14 @@ def _run_logged(job: dict, argv: list[str], *, timeout: float,
 
 def _kill_tree(proc: "subprocess.Popen") -> None:
     """Kill an operation through its stable nested group/Job owner."""
-    if isinstance(proc, OwnedPopen):
+    if isinstance(proc, (OwnedPopen, WindowsJobPopen)):
         # The retained supervisor/process-group or nested Job is the stable
         # per-operation owner.  Do not fall back to a direct PID kill.
         proc.kill()
         try:
             proc.wait(timeout=5)
         except subprocess.TimeoutExpired:
-            pass
+            return
         return
     # A test double or a legacy caller without the nested owner can only be
     # stopped through its stable direct-process handle.

backend/services/subprocess_backend.py

@@ -32,9 +32,9 @@ Threat-model summary (see Plan 02-01 frontmatter):
               AUTH-05 installed (``HFTokenRedactor``) on the root logger.
     T-02-04 — compromised sidecar emitting unexpected ops: parent allowlist
               ``PARENT_INBOUND_OPS`` rejects everything else.
-    T-02-05 — nested containment: a retained supervisor process group/Job owns
-              each engine operation and is linked to backend death by a control
-              pipe, while still permitting independent timeout teardown.
+    T-02-05 — nested containment: a retained POSIX supervisor process group or
+              Windows Job owns each engine operation, while still permitting
+              independent timeout teardown and cleanup on backend death.
 """
 from __future__ import annotations
 

backend/tests/test_contained_subprocess.py

@@ -257,3 +257,90 @@ def test_windows_assignment_failure_kills_suspended_unowned_child(monkeypatch):
     names = [event[0] for event in events]
     assert names.index("assign") < names.index("terminate") < names.index("kill")
     assert names.index("kill") < names.index("wait") < names.index("write")
+
+
+def test_windows_direct_job_owner_assigns_before_resume(monkeypatch):
+    """Windows skips the extra Python wrapper but retains pre-start Job ownership."""
+    events = []
+    job = 99
+
+    kernel = type("Kernel", (), {})()
+    kernel.AssignProcessToJobObject = _Call(
+        lambda assigned_job, process: events.append(("assign", assigned_job, process)) or True
+    )
+    kernel.TerminateJobObject = _Call(
+        lambda assigned_job, code: events.append(("terminate", assigned_job, code)) or True
+    )
+    kernel.CloseHandle = _Call(
+        lambda handle: events.append(("close", getattr(handle, "value", handle))) or True
+    )
+    monkeypatch.setattr(owned, "_windows_job", lambda: (job, kernel, wintypes))
+    monkeypatch.setattr(
+        owned,
+        "_resume_windows_process",
+        lambda _kernel, _types, pid: events.append(("resume", pid)),
+    )
+
+    class Child:
+        _handle = 77
+        pid = 123
+        args = ["operation.exe"]
+        stdin = None
+        stdout = object()
+        stderr = object()
+        returncode = None
+
+        def poll(self):
+            return self.returncode
+
+        def wait(self, timeout=None):
+            events.append(("wait", timeout))
+            return self.returncode
+
+        def kill(self):
+            events.append(("kill",))
+
+    child = Child()
+
+    def fake_popen(argv, **kwargs):
+        events.append(("spawn", argv, kwargs))
+        return child
+
+    monkeypatch.setattr(owned.subprocess, "Popen", fake_popen)
+    proc = owned._spawn_windows_owned(
+        ["operation.exe"],
+        {
+            "env": {
+                "KEEP": "yes",
+                "OMNIVOICE_DESKTOP_CONTAINED": "1",
+                "OMNIVOICE_DESKTOP_DRAIN_FD": "42",
+            },
+            "creationflags": 0x00000200,
+        },
+    )
+
+    names = [event[0] for event in events]
+    assert names[:3] == ["spawn", "assign", "resume"]
+    spawn_argv, spawn_kwargs = events[0][1:]
+    assert spawn_argv == ["operation.exe"]
+    assert spawn_kwargs["creationflags"] == 0x08000204
+    assert spawn_kwargs["env"] == {"KEEP": "yes"}
+    assert proc.stdout is child.stdout
+
+    child.returncode = 0
+    assert proc.poll() == 0
+    assert [event[0] for event in events][-2:] == ["terminate", "close"]
+
+
+def test_spawn_owned_selects_direct_windows_job_path(monkeypatch):
+    sentinel = object()
+    calls = []
+    monkeypatch.setattr(owned.os, "name", "nt")
+    monkeypatch.setattr(
+        owned,
+        "_spawn_windows_owned",
+        lambda argv, kwargs: calls.append((argv, kwargs)) or sentinel,
+    )
+
+    assert owned.spawn_owned(["sidecar.exe"], text=True) is sentinel
+    assert calls == [(["sidecar.exe"], {"text": True})]

tests/test_sidecar_install.py

@@ -373,6 +373,30 @@ def test_desktop_windows_timeout_never_taskkills_a_reusable_pid(monkeypatch):
     assert calls == {"handle_kill": True}
 
 
+def test_windows_job_timeout_waits_for_terminated_tree():
+    events = []
+
+    def timed_out_wait(timeout=None):
+        events.append(("wait", timeout))
+        raise subprocess.TimeoutExpired("operation.exe", timeout)
+
+    child = SimpleNamespace(
+        stdin=None,
+        stdout=None,
+        stderr=None,
+        wait=timed_out_wait,
+    )
+    kernel = SimpleNamespace(
+        TerminateJobObject=lambda job, code: events.append(("terminate", job, code)),
+        CloseHandle=lambda job: events.append(("close", job)),
+    )
+    proc = si.WindowsJobPopen(child, 99, kernel)
+
+    si._kill_tree(proc)
+
+    assert events == [("terminate", 99, 1), ("close", 99), ("wait", 5)]
+
+
 def test_desktop_installer_timeout_kills_nested_helper_before_it_can_mutate(
     monkeypatch, tmp_path
 ):