truf-server/app/owned_process.py

import sys

sys.dont_write_bytecode = True
if not sys.dont_write_bytecode:
    raise RuntimeError("owned process host could not disable bytecode writes")

import ctypes
import errno
import json
import math
import os
import queue
import signal
import struct
import subprocess
import threading
import time


# Keep this module standard-library-only: it is re-executed as the isolated
# containment host before any application code is allowed to run.
_HOST_FLAG = "--owned-process-host"
_MAX_PACKET_SIZE = 16 * 1024 * 1024
_CREATE_BREAKAWAY_FROM_JOB = 0x01000000
_TEST_FAIL_JOB_ASSIGNMENT = "OWNED_PROCESS_TEST_FAIL_JOB_ASSIGNMENT"
_WINDOWS_INHERIT_LOCK = threading.Lock()
_JOB_OBJECT_LIMIT_JOB_MEMORY = 0x00000200
_JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE = 0x00002000
_JOB_OBJECT_CPU_RATE_CONTROL_INFORMATION = 15
_JOB_OBJECT_CPU_RATE_CONTROL_ENABLE = 0x1
_JOB_OBJECT_CPU_RATE_CONTROL_WEIGHT_BASED = 0x2
_PROCESS_MEMORY_PRIORITY = 0
_MAX_JOB_PROCESS_IDS = 4096
_MAX_REAP_BATCH = 64
_HOST_PRIVATE_ENV_KEYS = frozenset({
    "SCANNER_SUPERVISED",
    "TRUF_MANAGED_POSTGRES_DSN",
    "SCANNER_DB_URL",
    "DATABASE_URL",
    "SCANNER_DASHBOARD_DB_URL",
    "KEYCHECK_DB_URL",
    "PGPASSWORD",
    "PGUSER",
    "PGDATABASE",
    "PGHOST",
    "PGHOSTADDR",
    "PGPORT",
    "PGSERVICE",
    "PGSERVICEFILE",
    "PGPASSFILE",
    "PGOPTIONS",
    "PGSSLMODE",
    "PGSSLKEY",
    "PGSSLCERT",
    "PGSSLROOTCERT",
})


if os.name == "nt":
    from ctypes import wintypes

    class _IO_COUNTERS(ctypes.Structure):
        _fields_ = [
            ("ReadOperationCount", ctypes.c_ulonglong),
            ("WriteOperationCount", ctypes.c_ulonglong),
            ("OtherOperationCount", ctypes.c_ulonglong),
            ("ReadTransferCount", ctypes.c_ulonglong),
            ("WriteTransferCount", ctypes.c_ulonglong),
            ("OtherTransferCount", ctypes.c_ulonglong),
        ]

    class _JOBOBJECT_BASIC_LIMIT_INFORMATION(ctypes.Structure):
        _fields_ = [
            ("PerProcessUserTimeLimit", ctypes.c_longlong),
            ("PerJobUserTimeLimit", ctypes.c_longlong),
            ("LimitFlags", wintypes.DWORD),
            ("MinimumWorkingSetSize", ctypes.c_size_t),
            ("MaximumWorkingSetSize", ctypes.c_size_t),
            ("ActiveProcessLimit", wintypes.DWORD),
            ("Affinity", ctypes.c_size_t),
            ("PriorityClass", wintypes.DWORD),
            ("SchedulingClass", wintypes.DWORD),
        ]

    class _JOBOBJECT_EXTENDED_LIMIT_INFORMATION(ctypes.Structure):
        _fields_ = [
            ("BasicLimitInformation", _JOBOBJECT_BASIC_LIMIT_INFORMATION),
            ("IoInfo", _IO_COUNTERS),
            ("ProcessMemoryLimit", ctypes.c_size_t),
            ("JobMemoryLimit", ctypes.c_size_t),
            ("PeakProcessMemoryUsed", ctypes.c_size_t),
            ("PeakJobMemoryUsed", ctypes.c_size_t),
        ]

    class _JOBOBJECT_BASIC_ACCOUNTING_INFORMATION(ctypes.Structure):
        _fields_ = [
            ("TotalUserTime", ctypes.c_longlong),
            ("TotalKernelTime", ctypes.c_longlong),
            ("ThisPeriodTotalUserTime", ctypes.c_longlong),
            ("ThisPeriodTotalKernelTime", ctypes.c_longlong),
            ("TotalPageFaultCount", wintypes.DWORD),
            ("TotalProcesses", wintypes.DWORD),
            ("ActiveProcesses", wintypes.DWORD),
            ("TotalTerminatedProcesses", wintypes.DWORD),
        ]

    class _JOBOBJECT_BASIC_PROCESS_ID_LIST(ctypes.Structure):
        _fields_ = [
            ("NumberOfAssignedProcesses", wintypes.DWORD),
            ("NumberOfProcessIdsInList", wintypes.DWORD),
            ("ProcessIdList", ctypes.c_size_t * _MAX_JOB_PROCESS_IDS),
        ]

    class _JOBOBJECT_CPU_RATE_CONTROL_INFORMATION(ctypes.Structure):
        _fields_ = [
            ("ControlFlags", wintypes.DWORD),
            ("Weight", wintypes.DWORD),
        ]

    class _MEMORY_PRIORITY_INFORMATION(ctypes.Structure):
        _fields_ = [("MemoryPriority", wintypes.ULONG)]

    class _FILETIME(ctypes.Structure):
        _fields_ = [
            ("dwLowDateTime", wintypes.DWORD),
            ("dwHighDateTime", wintypes.DWORD),
        ]

    _P_BOOL = ctypes.POINTER(wintypes.BOOL)
    _P_DWORD = ctypes.POINTER(wintypes.DWORD)
    _P_FILETIME = ctypes.POINTER(_FILETIME)

    _KERNEL32 = ctypes.WinDLL("kernel32", use_last_error=True)
    _CREATE_JOB_OBJECT = _KERNEL32.CreateJobObjectW
    _CREATE_JOB_OBJECT.argtypes = [ctypes.c_void_p, wintypes.LPCWSTR]
    _CREATE_JOB_OBJECT.restype = wintypes.HANDLE
    _SET_INFORMATION_JOB_OBJECT = _KERNEL32.SetInformationJobObject
    _SET_INFORMATION_JOB_OBJECT.argtypes = [
        wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p, wintypes.DWORD,
    ]
    _SET_INFORMATION_JOB_OBJECT.restype = wintypes.BOOL
    _QUERY_INFORMATION_JOB_OBJECT = _KERNEL32.QueryInformationJobObject
    _QUERY_INFORMATION_JOB_OBJECT.argtypes = [
        wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p, wintypes.DWORD, _P_DWORD,
    ]
    _QUERY_INFORMATION_JOB_OBJECT.restype = wintypes.BOOL
    _ASSIGN_PROCESS_TO_JOB_OBJECT = _KERNEL32.AssignProcessToJobObject
    _ASSIGN_PROCESS_TO_JOB_OBJECT.argtypes = [wintypes.HANDLE, wintypes.HANDLE]
    _ASSIGN_PROCESS_TO_JOB_OBJECT.restype = wintypes.BOOL
    _IS_PROCESS_IN_JOB = _KERNEL32.IsProcessInJob
    _IS_PROCESS_IN_JOB.argtypes = [wintypes.HANDLE, wintypes.HANDLE, _P_BOOL]
    _IS_PROCESS_IN_JOB.restype = wintypes.BOOL
    _GET_CURRENT_PROCESS = _KERNEL32.GetCurrentProcess
    _GET_CURRENT_PROCESS.argtypes = []
    _GET_CURRENT_PROCESS.restype = wintypes.HANDLE
    _SET_HANDLE_INFORMATION = _KERNEL32.SetHandleInformation
    _SET_HANDLE_INFORMATION.argtypes = [wintypes.HANDLE, wintypes.DWORD, wintypes.DWORD]
    _SET_HANDLE_INFORMATION.restype = wintypes.BOOL
    _GET_HANDLE_INFORMATION = _KERNEL32.GetHandleInformation
    _GET_HANDLE_INFORMATION.argtypes = [wintypes.HANDLE, _P_DWORD]
    _GET_HANDLE_INFORMATION.restype = wintypes.BOOL
    _CLOSE_HANDLE = _KERNEL32.CloseHandle
    _CLOSE_HANDLE.argtypes = [wintypes.HANDLE]
    _CLOSE_HANDLE.restype = wintypes.BOOL
    _TERMINATE_JOB_OBJECT = _KERNEL32.TerminateJobObject
    _TERMINATE_JOB_OBJECT.argtypes = [wintypes.HANDLE, wintypes.UINT]
    _TERMINATE_JOB_OBJECT.restype = wintypes.BOOL
    _EXIT_PROCESS = _KERNEL32.ExitProcess
    _EXIT_PROCESS.argtypes = [wintypes.UINT]
    _EXIT_PROCESS.restype = None
    _GET_PROCESS_TIMES = _KERNEL32.GetProcessTimes
    _GET_PROCESS_TIMES.argtypes = [
        wintypes.HANDLE, _P_FILETIME, _P_FILETIME, _P_FILETIME, _P_FILETIME,
    ]
    _GET_PROCESS_TIMES.restype = wintypes.BOOL
    _QUERY_FULL_PROCESS_IMAGE_NAME = _KERNEL32.QueryFullProcessImageNameW
    _QUERY_FULL_PROCESS_IMAGE_NAME.argtypes = [
        wintypes.HANDLE, wintypes.DWORD, wintypes.LPWSTR, _P_DWORD,
    ]
    _QUERY_FULL_PROCESS_IMAGE_NAME.restype = wintypes.BOOL
    _SET_PROCESS_INFORMATION = _KERNEL32.SetProcessInformation
    _SET_PROCESS_INFORMATION.argtypes = [
        wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p, wintypes.DWORD,
    ]
    _SET_PROCESS_INFORMATION.restype = wintypes.BOOL
    _GET_PROCESS_INFORMATION = _KERNEL32.GetProcessInformation
    _GET_PROCESS_INFORMATION.argtypes = [
        wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p, wintypes.DWORD,
    ]
    _GET_PROCESS_INFORMATION.restype = wintypes.BOOL
else:
    _IO_COUNTERS = _JOBOBJECT_BASIC_LIMIT_INFORMATION = None
    _JOBOBJECT_EXTENDED_LIMIT_INFORMATION = None
    _JOBOBJECT_BASIC_ACCOUNTING_INFORMATION = None
    _JOBOBJECT_BASIC_PROCESS_ID_LIST = _FILETIME = None
    _JOBOBJECT_CPU_RATE_CONTROL_INFORMATION = _MEMORY_PRIORITY_INFORMATION = None
    _KERNEL32 = None


class OwnedProcessError(OSError):
    def __init__(self, stage, message, error_number=None, windows_error=None):
        super().__init__(error_number or 0, f"owned process {stage} failed: {message}")
        self.stage = stage
        self.winerror = windows_error


def _write_all(fd, data):
    view = memoryview(data)
    while view:
        written = os.write(fd, view)
        if written <= 0:
            raise BrokenPipeError("control pipe closed")
        view = view[written:]


def _read_exact(fd, size):
    chunks = []
    remaining = size
    while remaining:
        chunk = os.read(fd, remaining)
        if not chunk:
            raise EOFError("pipe closed")
        chunks.append(chunk)
        remaining -= len(chunk)
    return b"".join(chunks)


def _write_packet(fd, value):
    data = json.dumps(value, ensure_ascii=True, separators=(",", ":")).encode("utf-8")
    if len(data) > _MAX_PACKET_SIZE:
        raise ValueError("owned process control packet is too large")
    _write_all(fd, struct.pack("!I", len(data)) + data)


def _read_packet(fd):
    size = struct.unpack("!I", _read_exact(fd, 4))[0]
    if size > _MAX_PACKET_SIZE:
        raise ValueError("owned process control packet is too large")
    return json.loads(_read_exact(fd, size).decode("utf-8"))


def _serializable_command(args):
    if isinstance(args, (str, os.PathLike)):
        return os.fspath(args)
    if isinstance(args, bytes):
        raise TypeError("OwnedProcess does not accept bytes command arguments")
    values = []
    for value in args:
        if isinstance(value, bytes):
            raise TypeError("OwnedProcess does not accept bytes command arguments")
        values.append(os.fspath(value) if isinstance(value, os.PathLike) else str(value))
    return values


def _serializable_environment(env):
    source = os.environ if env is None else env
    result = {}
    for key, value in source.items():
        if isinstance(key, bytes) or isinstance(value, bytes):
            raise TypeError("OwnedProcess does not accept bytes environment entries")
        result[str(key)] = str(value)
    result.pop(_TEST_FAIL_JOB_ASSIGNMENT, None)
    return result


def _validate_job_memory_limit_bytes(value):
    if isinstance(value, bool) or not isinstance(value, int):
        raise ValueError("job_memory_limit_bytes must be a non-negative integer")
    if value < 0:
        raise ValueError("job_memory_limit_bytes must be a non-negative integer")
    if value > (1 << (ctypes.sizeof(ctypes.c_size_t) * 8)) - 1:
        raise ValueError("job_memory_limit_bytes exceeds the host address size")
    return value


def _validate_job_cpu_weight(value):
    if isinstance(value, bool) or not isinstance(value, int) or value < 0 or value > 9:
        raise ValueError("job_cpu_weight must be 0 or an integer from 1 through 9")
    return value


def _validate_process_memory_priority(value):
    if isinstance(value, bool) or not isinstance(value, int) or value < 0 or value > 5:
        raise ValueError("process_memory_priority must be 0 or an integer from 1 through 5")
    return value


def _host_environment():
    """The containment host never needs application authority credentials."""
    env = os.environ.copy()
    for key in list(env):
        normalized = str(key).upper()
        if (
            normalized.startswith("TRUF_SUPERVISOR_")
            or normalized.startswith("TRUF_POSTGRES_")
            or normalized in _HOST_PRIVATE_ENV_KEYS
        ):
            env.pop(key, None)
    return env


def _require_isolated_host():
    if not (
        sys.flags.isolated
        and sys.flags.no_site
        and sys.flags.dont_write_bytecode
    ):
        raise RuntimeError("owned process host requires -I -S -B")


def _status_reader(fd_owner, result_queue):
    try:
        fd = fd_owner.get_nowait()
    except queue.Empty:
        return
    try:
        result_queue.put((True, _read_packet(fd)))
    except BaseException as exc:
        result_queue.put((False, exc))
    finally:
        try:
            os.close(fd)
        except OSError:
            pass


def _validated_identity(value, pid):
    prefix = "windows-filetime:" if os.name == "nt" else "proc-start-ticks:"
    if not isinstance(value, dict) or type(pid) is not int or pid <= 0:
        raise OwnedProcessError("identity", "missing process identity")
    creation = value.get("creation_time")
    timestamp = value.get("creation_time_unix")
    executable = value.get("executable")
    try:
        valid_timestamp = type(timestamp) in (int, float) and math.isfinite(timestamp) and timestamp > 0
    except OverflowError:
        valid_timestamp = False
    if (
        type(value.get("pid")) is not int or value["pid"] != pid
        or not isinstance(creation, str) or not creation.startswith(prefix)
        or not creation[len(prefix):].isascii() or not creation[len(prefix):].isdigit()
        or not valid_timestamp
        or not isinstance(executable, str) or not os.path.isabs(executable) or "\0" in executable
    ):
        raise OwnedProcessError("identity", "incomplete or mismatched process identity")
    return dict(value)


class OwnedProcess:
    """Popen-like process whose payload tree is owned by a host containment boundary."""

    def __init__(
        self,
        args,
        bufsize=-1,
        executable=None,
        stdin=None,
        stdout=None,
        stderr=None,
        preexec_fn=None,
        close_fds=True,
        shell=False,
        cwd=None,
        env=None,
        universal_newlines=None,
        startupinfo=None,
        creationflags=0,
        restore_signals=True,
        start_new_session=False,
        pass_fds=(),
        *,
        user=None,
        group=None,
        extra_groups=None,
        encoding=None,
        errors=None,
        text=None,
        umask=-1,
        pipesize=-1,
        process_group=None,
        job_memory_limit_bytes=0,
        job_cpu_weight=0,
        process_memory_priority=0,
        _startup_timeout=15,
    ):
        if preexec_fn is not None:
            raise ValueError("OwnedProcess does not support preexec_fn")
        if not close_fds:
            raise ValueError("OwnedProcess requires close_fds=True")
        if startupinfo is not None:
            raise ValueError("OwnedProcess does not support caller startupinfo")
        if start_new_session:
            raise ValueError("OwnedProcess owns the payload session")
        if pass_fds:
            raise ValueError("OwnedProcess does not pass arbitrary file descriptors")
        if user is not None or group is not None or extra_groups is not None or umask != -1:
            raise ValueError("OwnedProcess does not support credential or umask changes")
        if process_group not in (None, -1):
            raise ValueError("OwnedProcess owns the payload process group")
        if int(creationflags or 0) & _CREATE_BREAKAWAY_FROM_JOB:
            raise ValueError("CREATE_BREAKAWAY_FROM_JOB is forbidden for owned processes")
        job_memory_limit_bytes = _validate_job_memory_limit_bytes(job_memory_limit_bytes)
        job_cpu_weight = _validate_job_cpu_weight(job_cpu_weight)
        process_memory_priority = _validate_process_memory_priority(process_memory_priority)

        self.args = args
        self._owner_pid = os.getpid()
        self._host_process = None
        self._control_fd = None
        self._control_lock = threading.Lock()
        self._payload_pid = None
        self._payload_identity = None
        self._host_identity = None
        self._job_membership_verified = False
        self._job_accounting = {}
        self._job_members = []
        self._status_fd_owner = queue.SimpleQueue()
        self._configuration_sent = False

        config = {
            "args": _serializable_command(args),
            "bufsize": int(bufsize),
            "executable": os.fspath(executable) if executable is not None else None,
            "shell": bool(shell),
            "cwd": os.fspath(cwd) if cwd is not None else None,
            "env": _serializable_environment(env),
            "creationflags": int(creationflags or 0),
            "restore_signals": bool(restore_signals),
            "job_memory_limit_bytes": job_memory_limit_bytes,
            "job_cpu_weight": job_cpu_weight,
            "process_memory_priority": process_memory_priority,
        }

        # Windows venv executables are redirecting launchers. Retain the actual
        # stdlib-only observer, not a launcher with a different process identity.
        host_python = sys._base_executable if os.name == "nt" else sys.executable
        control_read, control_write = os.pipe()
        status_read, status_write = os.pipe()
        self._status_fd_owner.put(status_read)
        self._control_fd = control_write
        host_command = [
            host_python,
            "-I",
            "-S",
            "-B",
            os.path.abspath(__file__),
            _HOST_FLAG,
        ]
        popen_options = {
            "stdin": stdin,
            "stdout": stdout,
            "stderr": stderr,
            "bufsize": bufsize,
            "universal_newlines": universal_newlines,
            "encoding": encoding,
            "errors": errors,
            "text": text,
            "close_fds": True,
            "pipesize": pipesize,
            "env": _host_environment(),
        }

        try:
            if os.name == "nt":
                import msvcrt

                control_handle = msvcrt.get_osfhandle(control_read)
                status_handle = msvcrt.get_osfhandle(status_write)
                host_command.extend([str(control_handle), str(status_handle)])
                startup = subprocess.STARTUPINFO()
                startup.lpAttributeList = {"handle_list": [control_handle, status_handle]}
                popen_options["startupinfo"] = startup
                popen_options["creationflags"] = self._host_creationflags(config["creationflags"])
                with _WINDOWS_INHERIT_LOCK:
                    os.set_inheritable(control_read, True)
                    os.set_inheritable(status_write, True)
                    try:
                        self._host_process = subprocess.Popen(host_command, **popen_options)
                    finally:
                        os.set_inheritable(control_read, False)
                        os.set_inheritable(status_write, False)
            else:
                host_command.extend([str(control_read), str(status_write)])
                popen_options["pass_fds"] = (control_read, status_write)
                # An inner observer must survive the outer payload-group stop.
                popen_options["start_new_session"] = True
                self._host_process = subprocess.Popen(host_command, **popen_options)
        except BaseException:
            self._close_fd(control_read)
            self._close_fd(control_write)
            self._cancel_status_reader()
            self._close_fd(status_write)
            self._control_fd = None
            raise

        self._close_fd(control_read)
        self._close_fd(status_write)
        try:
            _write_packet(self._control_fd, config)
            self._configuration_sent = True
            status = self._wait_for_startup(float(_startup_timeout))
            if not isinstance(status, dict) or status.get("ok") is not True:
                detail = status if isinstance(status, dict) else {}
                raise OwnedProcessError(
                    str(detail.get("stage") or "startup"),
                    str(detail.get("message") or "host exited without a valid handshake"),
                    detail.get("errno"),
                    detail.get("winerror"),
                )
            if status.get("job_membership_verified") is not True:
                raise OwnedProcessError("job_membership", "host did not prove payload containment")
            self._payload_pid = status.get("pid")
            if self._payload_pid == self._host_process.pid:
                raise OwnedProcessError("identity", "payload cannot be its own observer")
            self._payload_identity = _validated_identity(status.get("payload_identity"), self._payload_pid)
            self._host_identity = _validated_identity(status.get("host_identity"), self._host_process.pid)
            members = status.get("job_members")
            if not isinstance(members, list) or len(members) > _MAX_JOB_PROCESS_IDS:
                raise OwnedProcessError("identity", "invalid containment member list")
            self._job_members = [
                _validated_identity(item, item.get("pid") if isinstance(item, dict) else None)
                for item in members
            ]
            if self._payload_identity not in self._job_members or self._host_identity not in self._job_members:
                raise OwnedProcessError("identity", "retained identities missing from containment members")
            accounting = status.get("job_accounting")
            if not isinstance(accounting, dict) or (
                os.name != "nt" and accounting.get("child_subreaper_verified") is not True
            ):
                raise OwnedProcessError("job_membership", "missing containment accounting or subreaper proof")
            self._job_accounting = dict(accounting)
            self._job_membership_verified = True
        except BaseException:
            # An unclaimed read end can keep the host blocked on a full pipe.
            self._cancel_status_reader()
            self._request_stop()
            self._reap_failed_start()
            raise
        finally:
            self._cancel_status_reader()

    @staticmethod
    def _close_fd(fd):
        if fd is None:
            return
        try:
            os.close(fd)
        except OSError:
            pass

    @staticmethod
    def _host_creationflags(payload_flags):
        # Keep host visibility and scheduling aligned with the payload without
        # copying payload-only process-group flags.
        allowed = (
            getattr(subprocess, "CREATE_NO_WINDOW", 0x08000000)
            | getattr(subprocess, "IDLE_PRIORITY_CLASS", 0x00000040)
            | getattr(subprocess, "BELOW_NORMAL_PRIORITY_CLASS", 0x00004000)
            | getattr(subprocess, "NORMAL_PRIORITY_CLASS", 0x00000020)
            | getattr(subprocess, "ABOVE_NORMAL_PRIORITY_CLASS", 0x00008000)
            | getattr(subprocess, "HIGH_PRIORITY_CLASS", 0x00000080)
            | getattr(subprocess, "REALTIME_PRIORITY_CLASS", 0x00000100)
        )
        return int(payload_flags) & allowed

    def _cancel_status_reader(self):
        try:
            fd = self._status_fd_owner.get_nowait()
        except queue.Empty:
            return
        self._close_fd(fd)

    def _wait_for_startup(self, timeout):
        result_queue = queue.Queue(maxsize=1)
        # Either the reader claims the FD or failed construction cancels it.
        # Thread.start() interruption cannot make both sides own the same FD.
        thread = threading.Thread(target=_status_reader, args=(self._status_fd_owner, result_queue), daemon=True)
        thread.start()
        try:
            ok, result = result_queue.get(timeout=max(0.1, timeout))
        except queue.Empty as exc:
            raise subprocess.TimeoutExpired(self.args, timeout) from exc
        if not ok:
            if self._host_process.poll() is not None:
                return {
                    "ok": False,
                    "stage": "startup",
                    "message": f"host exited with code {self._host_process.returncode}",
                }
            raise result
        return result

    def _reap_failed_start(self):
        if self._host_process is None:
            return
        if os.name != "nt":
            # Keep the observer retained even if startup cleanup is interrupted.
            # Its exit, not a timeout or repeated Ctrl+C, acknowledges teardown.
            while True:
                try:
                    self._host_process.wait()
                    return
                except BaseException:
                    try:
                        time.sleep(0.05)
                    except BaseException:
                        pass
        try:
            self._host_process.wait(timeout=2)
            return
        except subprocess.TimeoutExpired:
            pass
        self._host_process.terminate()
        try:
            self._host_process.wait(timeout=2)
        except subprocess.TimeoutExpired:
            self._host_process.kill()
            self._host_process.wait()

    def _request_stop(self):
        if os.getpid() != self._owner_pid:
            # A forked proxy owns only its local FD, not the parent's job or lock.
            fd, self._control_fd = self._control_fd, None
            self._close_fd(fd)
            return
        with self._control_lock:
            fd = self._control_fd
            self._control_fd = None
        if fd is not None:
            try:
                if self._configuration_sent:
                    # EOF alone is insufficient when a fork inherited a writer.
                    if os.name != "nt":
                        os.set_blocking(fd, False)
                    os.write(fd, b"\0")
            except OSError:
                pass
            finally:
                self._close_fd(fd)

    @property
    def pid(self):
        return self._payload_pid

    @property
    def host_pid(self):
        return self._host_process.pid

    @property
    def payload_identity(self):
        return dict(self._payload_identity or {})

    @property
    def host_identity(self):
        return dict(self._host_identity or {})

    @property
    def job_membership_verified(self):
        return bool(self._job_membership_verified)

    def ownership_snapshot(self):
        return {
            "host_identity": self.host_identity,
            "payload_identity": self.payload_identity,
            "job_membership_verified": self.job_membership_verified,
            "job_accounting": dict(self._job_accounting),
            "active_members": [dict(item) for item in self._job_members],
        }

    @property
    def stdin(self):
        return self._host_process.stdin

    @property
    def stdout(self):
        return self._host_process.stdout

    @property
    def stderr(self):
        return self._host_process.stderr

    @property
    def returncode(self):
        return self._host_process.returncode

    def poll(self):
        code = self._host_process.poll()
        if code is not None:
            self._request_stop()
        return code

    def wait(self, timeout=None):
        try:
            try:
                return self._host_process.wait(timeout=timeout)
            except subprocess.TimeoutExpired as exc:
                raise subprocess.TimeoutExpired(self.args, timeout, output=exc.output, stderr=exc.stderr) from None
        finally:
            if self._host_process.returncode is not None:
                self._request_stop()

    def communicate(self, input=None, timeout=None):
        try:
            try:
                return self._host_process.communicate(input=input, timeout=timeout)
            except subprocess.TimeoutExpired as exc:
                raise subprocess.TimeoutExpired(self.args, timeout, output=exc.output, stderr=exc.stderr) from None
        finally:
            if self._host_process.returncode is not None:
                self._request_stop()

    def terminate(self):
        self._request_stop()

    def kill(self):
        self._request_stop()
        if os.name == "nt" and self._host_process.poll() is None:
            # Terminating only the retained host closes its sole Job handle.
            self._host_process.kill()

    def send_signal(self, sig):
        terminating = {signal.SIGTERM}
        if hasattr(signal, "SIGKILL"):
            terminating.add(signal.SIGKILL)
        if hasattr(signal, "CTRL_BREAK_EVENT"):
            terminating.add(signal.CTRL_BREAK_EVENT)
        if hasattr(signal, "CTRL_C_EVENT"):
            terminating.add(signal.CTRL_C_EVENT)
        if sig not in terminating:
            raise ValueError("OwnedProcess only supports containment-scoped stop signals")
        if hasattr(signal, "SIGKILL") and sig == signal.SIGKILL:
            self.kill()
        else:
            self.terminate()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, value, traceback):
        if self.stdout:
            self.stdout.close()
        if self.stderr:
            self.stderr.close()
        try:
            if self.stdin:
                self.stdin.close()
        finally:
            self.wait()

    def __del__(self):
        try:
            self._request_stop()
        except BaseException:
            pass


def run_owned(*popenargs, input=None, capture_output=False, timeout=None, check=False, **kwargs):
    if input is not None:
        if kwargs.get("stdin") is not None:
            raise ValueError("stdin and input arguments may not both be used")
        kwargs["stdin"] = subprocess.PIPE
    if capture_output:
        if kwargs.get("stdout") is not None or kwargs.get("stderr") is not None:
            raise ValueError("stdout/stderr and capture_output may not both be used")
        kwargs["stdout"] = subprocess.PIPE
        kwargs["stderr"] = subprocess.PIPE

    with OwnedProcess(*popenargs, **kwargs) as process:
        try:
            stdout, stderr = process.communicate(input, timeout=timeout)
        except subprocess.TimeoutExpired as exc:
            process.kill()
            exc.stdout, exc.stderr = process.communicate()
            raise
        except BaseException:
            process.kill()
            process.wait()
            raise
        code = process.poll()
        if check and code:
            raise subprocess.CalledProcessError(code, popenargs[0], output=stdout, stderr=stderr)
    return subprocess.CompletedProcess(popenargs[0], code, stdout, stderr)


def _open_inherited_fd(value, write=False):
    if os.name != "nt":
        fd = int(value)
        os.set_inheritable(fd, False)
        return fd

    import msvcrt

    flags = os.O_WRONLY if write else os.O_RDONLY
    flags |= getattr(os, "O_BINARY", 0)
    fd = msvcrt.open_osfhandle(int(value), flags)
    os.set_inheritable(fd, False)
    return fd


def _windows_job_limit_information(job_memory_limit_bytes=0):
    job_memory_limit_bytes = _validate_job_memory_limit_bytes(job_memory_limit_bytes)
    limits = _JOBOBJECT_EXTENDED_LIMIT_INFORMATION()
    limits.BasicLimitInformation.LimitFlags = _JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
    if job_memory_limit_bytes:
        limits.BasicLimitInformation.LimitFlags |= _JOB_OBJECT_LIMIT_JOB_MEMORY
        limits.JobMemoryLimit = job_memory_limit_bytes
    return limits


def _windows_job(job_memory_limit_bytes=0, job_cpu_weight=0):
    job = _CREATE_JOB_OBJECT(None, None)
    if not job:
        raise ctypes.WinError(ctypes.get_last_error())
    try:
        handle_flag_inherit = 0x00000001
        if not _SET_HANDLE_INFORMATION(job, handle_flag_inherit, 0):
            raise ctypes.WinError(ctypes.get_last_error())
        handle_flags = wintypes.DWORD()
        if not _GET_HANDLE_INFORMATION(job, ctypes.byref(handle_flags)):
            raise ctypes.WinError(ctypes.get_last_error())
        if handle_flags.value & handle_flag_inherit:
            raise OSError(errno.EPERM, "Job handle remained inheritable")

        limits = _windows_job_limit_information(job_memory_limit_bytes)
        if not _SET_INFORMATION_JOB_OBJECT(job, 9, ctypes.byref(limits), ctypes.sizeof(limits)):
            raise ctypes.WinError(ctypes.get_last_error())
        if job_cpu_weight:
            cpu_policy = _JOBOBJECT_CPU_RATE_CONTROL_INFORMATION()
            cpu_policy.ControlFlags = (
                _JOB_OBJECT_CPU_RATE_CONTROL_ENABLE
                | _JOB_OBJECT_CPU_RATE_CONTROL_WEIGHT_BASED
            )
            cpu_policy.Weight = int(job_cpu_weight)
            if not _SET_INFORMATION_JOB_OBJECT(
                job, _JOB_OBJECT_CPU_RATE_CONTROL_INFORMATION,
                ctypes.byref(cpu_policy), ctypes.sizeof(cpu_policy),
            ):
                raise ctypes.WinError(ctypes.get_last_error())
        if os.getenv(_TEST_FAIL_JOB_ASSIGNMENT) == "1":
            raise OSError(errno.EPERM, "deliberate Job assignment failure")
        current_process = _GET_CURRENT_PROCESS()
        if not _ASSIGN_PROCESS_TO_JOB_OBJECT(job, current_process):
            raise ctypes.WinError(ctypes.get_last_error())
        in_job = wintypes.BOOL()
        if not _IS_PROCESS_IN_JOB(current_process, job, ctypes.byref(in_job)):
            raise ctypes.WinError(ctypes.get_last_error())
        if not in_job.value:
            raise OSError(errno.EPERM, "host Job membership verification failed")
        return _KERNEL32, job
    except BaseException:
        _CLOSE_HANDLE(job)
        raise


def _windows_process_identity(handle, pid):
    creation = _FILETIME()
    ignored_exit = _FILETIME()
    ignored_kernel = _FILETIME()
    ignored_user = _FILETIME()
    if not _GET_PROCESS_TIMES(
        handle, ctypes.byref(creation), ctypes.byref(ignored_exit),
        ctypes.byref(ignored_kernel), ctypes.byref(ignored_user),
    ):
        raise ctypes.WinError(ctypes.get_last_error())
    filetime = (int(creation.dwHighDateTime) << 32) | int(creation.dwLowDateTime)
    path = ctypes.create_unicode_buffer(32768)
    length = wintypes.DWORD(len(path))
    if not _QUERY_FULL_PROCESS_IMAGE_NAME(handle, 0, path, ctypes.byref(length)):
        raise ctypes.WinError(ctypes.get_last_error())
    return {
        "pid": int(pid),
        "creation_time": f"windows-filetime:{filetime}",
        "creation_time_unix": (filetime - 116444736000000000) / 10000000.0,
        "executable": os.path.normcase(os.path.realpath(os.path.abspath(path.value))),
    }


def _windows_process_memory_priority(handle):
    policy = _MEMORY_PRIORITY_INFORMATION()
    if not _GET_PROCESS_INFORMATION(
        handle, _PROCESS_MEMORY_PRIORITY, ctypes.byref(policy), ctypes.sizeof(policy),
    ):
        raise ctypes.WinError(ctypes.get_last_error())
    return int(policy.MemoryPriority)


def _windows_set_process_memory_priority(handle, priority):
    policy = _MEMORY_PRIORITY_INFORMATION()
    policy.MemoryPriority = int(priority)
    if not _SET_PROCESS_INFORMATION(
        handle, _PROCESS_MEMORY_PRIORITY, ctypes.byref(policy), ctypes.sizeof(policy),
    ):
        raise ctypes.WinError(ctypes.get_last_error())
    if _windows_process_memory_priority(handle) != int(priority):
        raise OSError(errno.EPERM, "process memory-priority verification failed")


def _windows_verify_payload_job(
    job, payload, job_memory_limit_bytes=0, job_cpu_weight=0,
    process_memory_priority=0,
):
    payload_handle = wintypes.HANDLE(int(payload._handle))
    in_job = wintypes.BOOL()
    if not _IS_PROCESS_IN_JOB(payload_handle, job, ctypes.byref(in_job)):
        raise ctypes.WinError(ctypes.get_last_error())
    if not in_job.value:
        raise OSError(errno.EPERM, "payload is not in the exact retained Job")

    process_ids = _JOBOBJECT_BASIC_PROCESS_ID_LIST()
    returned = wintypes.DWORD()
    if not _QUERY_INFORMATION_JOB_OBJECT(
        job, 3, ctypes.byref(process_ids), ctypes.sizeof(process_ids), ctypes.byref(returned),
    ):
        raise ctypes.WinError(ctypes.get_last_error())
    count = int(process_ids.NumberOfProcessIdsInList)
    assigned = int(process_ids.NumberOfAssignedProcesses)
    if count > _MAX_JOB_PROCESS_IDS or assigned > _MAX_JOB_PROCESS_IDS:
        raise OSError(errno.EOVERFLOW, "Job member list exceeds its configured bound")
    members = [int(process_ids.ProcessIdList[index]) for index in range(count)]
    if int(payload.pid) not in members:
        raise OSError(errno.EPERM, "payload PID is absent from the exact Job member list")

    accounting = _JOBOBJECT_BASIC_ACCOUNTING_INFORMATION()
    if not _QUERY_INFORMATION_JOB_OBJECT(
        job, 1, ctypes.byref(accounting), ctypes.sizeof(accounting), ctypes.byref(returned),
    ):
        raise ctypes.WinError(ctypes.get_last_error())
    limits = _JOBOBJECT_EXTENDED_LIMIT_INFORMATION()
    if not _QUERY_INFORMATION_JOB_OBJECT(
        job, 9, ctypes.byref(limits), ctypes.sizeof(limits), ctypes.byref(returned),
    ):
        raise ctypes.WinError(ctypes.get_last_error())
    limit_flags = int(limits.BasicLimitInformation.LimitFlags)
    if not limit_flags & _JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE:
        raise OSError(errno.EPERM, "Job kill-on-close limit is missing")
    if job_memory_limit_bytes:
        if not limit_flags & _JOB_OBJECT_LIMIT_JOB_MEMORY:
            raise OSError(errno.EPERM, "Job memory limit is missing")
        if int(limits.JobMemoryLimit) != int(job_memory_limit_bytes):
            raise OSError(errno.EPERM, "Job memory limit verification failed")

    if job_cpu_weight:
        cpu_policy = _JOBOBJECT_CPU_RATE_CONTROL_INFORMATION()
        if not _QUERY_INFORMATION_JOB_OBJECT(
            job, _JOB_OBJECT_CPU_RATE_CONTROL_INFORMATION,
            ctypes.byref(cpu_policy), ctypes.sizeof(cpu_policy), ctypes.byref(returned),
        ):
            raise ctypes.WinError(ctypes.get_last_error())
        expected_flags = (
            _JOB_OBJECT_CPU_RATE_CONTROL_ENABLE
            | _JOB_OBJECT_CPU_RATE_CONTROL_WEIGHT_BASED
        )
        if int(cpu_policy.ControlFlags) != expected_flags or int(cpu_policy.Weight) != int(job_cpu_weight):
            raise OSError(errno.EPERM, "Job CPU-weight verification failed")

    payload_memory_priority = _windows_process_memory_priority(payload_handle)
    if process_memory_priority and payload_memory_priority != int(process_memory_priority):
        raise OSError(errno.EPERM, "payload memory-priority verification failed")
    payload_identity = _windows_process_identity(payload_handle, payload.pid)
    host_identity = _windows_process_identity(_GET_CURRENT_PROCESS(), os.getpid())
    member_identities = [host_identity]
    if payload_identity["pid"] != host_identity["pid"]:
        member_identities.append(payload_identity)
    return {
        "payload_identity": payload_identity,
        "host_identity": host_identity,
        "job_members": member_identities,
        "job_accounting": {
            "assigned_processes": assigned,
            "listed_processes": count,
            "active_processes": int(accounting.ActiveProcesses),
            "total_processes": int(accounting.TotalProcesses),
            "terminated_processes": int(accounting.TotalTerminatedProcesses),
            "peak_job_memory_bytes": int(limits.PeakJobMemoryUsed),
            "job_memory_limit_bytes": int(limits.JobMemoryLimit),
            "job_limit_flags": limit_flags,
            "job_cpu_weight": int(job_cpu_weight),
            "process_memory_priority": payload_memory_priority,
        },
    }


def _host_error(stage, exc):
    return {
        "ok": False,
        "stage": stage,
        "message": str(exc) or type(exc).__name__,
        "errno": getattr(exc, "errno", None),
        "winerror": getattr(exc, "winerror", None),
    }


def _host_stdio(fd):
    try:
        os.fstat(fd)
        return fd
    except OSError:
        return None


def _terminate_posix_group(process_group):
    try:
        os.killpg(process_group, signal.SIGKILL)
    except ProcessLookupError:
        pass


def _linux_process_identity(pid):
    try:
        with open(f"/proc/{pid}/stat", "rb") as handle:
            head, separator, tail = handle.read().rpartition(b") ")
        fields = tail.split()
        if not separator or int(head.partition(b" (")[0]) != pid:
            raise ValueError("invalid proc process ID")
        start_ticks = int(fields[19])
        if start_ticks < 0:
            raise ValueError("invalid proc start time")
        executable = os.readlink(f"/proc/{pid}/exe")
        if not os.path.isabs(executable):
            raise ValueError("nonabsolute proc executable")
        executable = os.path.normcase(os.path.realpath(os.path.abspath(executable)))
        clock_ticks = int(os.sysconf("SC_CLK_TCK"))
        boot_time = None
        with open("/proc/stat", "rb") as handle:
            for line in handle:
                if line.startswith(b"btime "):
                    boot_time = int(line.split()[1])
                    break
        if not boot_time or boot_time < 0 or clock_ticks <= 0:
            raise ValueError("proc boot time or clock rate unavailable")
        identity = {
            "pid": pid,
            "creation_time": f"proc-start-ticks:{start_ticks}",
            "creation_time_unix": boot_time + start_ticks / clock_ticks,
            "executable": executable,
        }
        return _validated_identity(identity, pid)
    except (OSError, ValueError, IndexError, OverflowError) as exc:
        raise OwnedProcessError("identity", f"unable to inspect retained Linux process {pid}") from exc


def _linux_enable_subreaper():
    if sys.platform != "linux" or not all(
        hasattr(os, name) for name in (
            "waitid", "WNOWAIT", "WEXITED", "WNOHANG", "P_ALL", "CLD_EXITED", "CLD_KILLED", "CLD_DUMPED",
        )
    ):
        raise OSError(errno.ENOSYS, "owned POSIX processes require Linux subreapers and waitid")
    prctl = ctypes.CDLL(None, use_errno=True).prctl
    prctl.argtypes = [ctypes.c_int] + [ctypes.c_ulong] * 4
    prctl.restype = ctypes.c_int
    enabled = ctypes.c_int()
    if prctl(36, 1, 0, 0, 0) != 0 or prctl(37, ctypes.addressof(enabled), 0, 0, 0) != 0:
        raise OSError(ctypes.get_errno(), "unable to establish child subreaper")
    if enabled.value != 1:
        raise OSError(errno.EPERM, "child subreaper verification failed")
    # An inherited SIGCHLD ignore disposition would auto-reap the group leader.
    signal.signal(signal.SIGCHLD, signal.SIG_DFL)
    if os.getpgid(0) != os.getpid() or os.getsid(0) != os.getpid():
        raise OSError(errno.EPERM, "Linux observer is not in its own session")


def _linux_payload_returncode(payload):
    # Drain adoptees while the root is alive, but bound each pass so sustained
    # child exits cannot starve control requests. Never reap the root here.
    for _ in range(_MAX_REAP_BATCH):
        status = os.waitid(os.P_ALL, 0, os.WEXITED | os.WNOHANG | os.WNOWAIT)
        if status is None:
            return None
        if status.si_code == os.CLD_EXITED:
            returncode = status.si_status
        elif status.si_code in (os.CLD_KILLED, os.CLD_DUMPED):
            returncode = -status.si_status
        else:
            raise OSError(errno.ECHILD, "unexpected retained child wait status")
        if status.si_pid == payload.pid:
            return returncode
        if status.si_pid <= 0:
            raise OSError(errno.ECHILD, "invalid retained child wait process")
        os.waitpid(status.si_pid, 0)
    return None


def _linux_direct_children():
    try:
        with open(
            f"/proc/{os.getpid()}/task/{os.getpid()}/children",
            "r", encoding="ascii",
        ) as handle:
            values = handle.read(1024 * 1024).split()
    except OSError as exc:
        raise OwnedProcessError("descendant_enumeration", "unable to enumerate owned Linux descendants") from exc
    if len(values) > _MAX_JOB_PROCESS_IDS:
        raise OwnedProcessError("descendant_enumeration", "owned Linux descendant count exceeds its bound")
    children = []
    for value in values:
        try:
            pid = int(value)
        except ValueError as exc:
            raise OwnedProcessError("descendant_enumeration", "invalid owned Linux descendant identity") from exc
        if pid <= 0 or pid == os.getpid():
            raise OwnedProcessError("descendant_enumeration", "invalid owned Linux descendant process ID")
        children.append(pid)
    return children


def _linux_kill_exact_descendant(pid):
    try:
        identity = _linux_process_identity(pid)
    except OwnedProcessError:
        return
    pidfd = None
    try:
        if hasattr(os, "pidfd_open"):
            try:
                pidfd = os.pidfd_open(pid, 0)
            except OSError:
                pidfd = None
        sender = getattr(signal, "pidfd_send_signal", None)
        if pidfd is not None and sender is not None:
            sender(pidfd, signal.SIGKILL, None, 0)
        else:
            current = _linux_process_identity(pid)
            if (
                current["creation_time"] != identity["creation_time"]
                or current["executable"] != identity["executable"]
            ):
                return
            os.kill(pid, signal.SIGKILL)
        try:
            process_group = os.getpgid(pid)
        except ProcessLookupError:
            return
        if process_group > 0 and process_group != os.getpgrp():
            _terminate_posix_group(process_group)
    except ProcessLookupError:
        pass
    finally:
        if pidfd is not None:
            os.close(pidfd)


def _linux_reap_adopted_batch():
    reaped = 0
    for _ in range(_MAX_REAP_BATCH):
        try:
            pid, _status = os.waitpid(-1, getattr(os, "WNOHANG", 1))
        except InterruptedError:
            continue
        except ChildProcessError:
            return reaped
        if pid == 0:
            return reaped
        reaped += 1
    return reaped


def _linux_stop_and_reap(payload):
    if payload.returncode is None:
        # Keep the leader unreaped until after killpg, pinning its numeric PGID.
        _terminate_posix_group(payload.pid)
        payload.wait()
    while True:
        _linux_reap_adopted_batch()
        children = _linux_direct_children()
        if not children:
            # Recheck after a scheduling point so a just-killed intermediate
            # cannot publish an adopted child after teardown was acknowledged.
            time.sleep(0.01)
            _linux_reap_adopted_batch()
            if not _linux_direct_children():
                try:
                    os.waitpid(-1, getattr(os, "WNOHANG", 1))
                except ChildProcessError:
                    return
                continue
        for pid in children:
            _linux_kill_exact_descendant(pid)
        # Never exit while an escaped-session descendant remains. Each pass is
        # bounded; an unkillable process keeps this observer alive and prevents
        # the owner from mistaking host exit for successful containment teardown.
        time.sleep(0.01)


def _exit_like_payload(returncode):
    if os.name == "nt":
        # Process teardown closes the host's sole Job handle after preserving
        # the payload's full DWORD exit code, killing any remaining members.
        _EXIT_PROCESS(int(returncode) & 0xFFFFFFFF)
        raise AssertionError("ExitProcess returned")
    if returncode < 0:
        signum = -int(returncode)
        if signum != signal.SIGKILL:
            signal.signal(signum, signal.SIG_DFL)
        signal.pthread_sigmask(signal.SIG_UNBLOCK, {signum})
        os.kill(os.getpid(), signum)
    os._exit(int(returncode) & 0xFF)


def _host_main(control_value, status_value):
    control_fd = _open_inherited_fd(control_value)
    status_fd = _open_inherited_fd(status_value, write=True)
    status_sent = False
    stage = "configuration"
    payload = None
    job_api = None
    job = None
    stop_event = threading.Event()
    signal_stop_requested = False

    def request_stop(signum=None, frame=None):
        nonlocal signal_stop_requested
        signal_stop_requested = True

    if os.name != "nt":
        signal.signal(signal.SIGTERM, request_stop)
        signal.signal(signal.SIGINT, request_stop)

    try:
        _require_isolated_host()
        config = _read_packet(control_fd)
        if int(config.get("creationflags") or 0) & _CREATE_BREAKAWAY_FROM_JOB:
            raise ValueError("CREATE_BREAKAWAY_FROM_JOB is forbidden for owned processes")
        job_memory_limit_bytes = _validate_job_memory_limit_bytes(config.get("job_memory_limit_bytes", 0))
        job_cpu_weight = _validate_job_cpu_weight(config.get("job_cpu_weight", 0))
        process_memory_priority = _validate_process_memory_priority(
            config.get("process_memory_priority", 0)
        )

        if os.name == "nt":
            stage = "job_assignment"
            if process_memory_priority:
                _windows_set_process_memory_priority(
                    _GET_CURRENT_PROCESS(), process_memory_priority,
                )
            job_api, job = _windows_job(job_memory_limit_bytes, job_cpu_weight)
        else:
            stage = "containment"
            _linux_enable_subreaper()

        if signal_stop_requested or stop_event.is_set():
            raise InterruptedError("parent stopped before payload launch")

        stage = "payload_launch"
        payload_options = {
            "stdin": _host_stdio(0),
            "stdout": _host_stdio(1),
            "stderr": _host_stdio(2),
            "bufsize": int(config.get("bufsize", -1)),
            "executable": config.get("executable"),
            "shell": bool(config.get("shell")),
            "cwd": config.get("cwd"),
            "env": config.get("env"),
            "restore_signals": bool(config.get("restore_signals", True)),
            "close_fds": True,
        }
        if os.name == "nt":
            payload_options["creationflags"] = int(config.get("creationflags") or 0)
        else:
            payload_options["start_new_session"] = True
        payload = subprocess.Popen(config["args"], **payload_options)
        if os.name == "nt":
            stage = "job_membership"
            membership = _windows_verify_payload_job(
                job, payload, job_memory_limit_bytes, job_cpu_weight,
                process_memory_priority,
            )
            status = {
                "ok": True,
                "pid": payload.pid,
                "job_membership_verified": True,
                **membership,
            }
        else:
            stage = "identity"
            payload_identity = _linux_process_identity(payload.pid)
            host_identity = _linux_process_identity(os.getpid())
            if os.getpgid(payload.pid) != payload.pid or os.getsid(payload.pid) != payload.pid:
                raise OSError(errno.EPERM, "payload is not in its owned Linux session")
            status = {
                "ok": True,
                "pid": payload.pid,
                "job_membership_verified": True,
                "payload_identity": payload_identity,
                "host_identity": host_identity,
                "job_members": [host_identity, payload_identity],
                "job_accounting": {"child_subreaper_verified": True, "listed_processes": 2},
            }
        _write_packet(status_fd, status)
        status_sent = True
        os.close(status_fd)
        status_fd = None

        def watch_parent():
            try:
                os.read(control_fd, 1)
            except OSError:
                pass
            stop_event.set()

        threading.Thread(target=watch_parent, daemon=True).start()
        while True:
            returncode = payload.poll() if os.name == "nt" else _linux_payload_returncode(payload)
            if returncode is not None:
                if os.name != "nt":
                    _linux_stop_and_reap(payload)
                return _exit_like_payload(returncode)
            if signal_stop_requested or stop_event.wait(0.05):
                if os.name == "nt":
                    job_api.TerminateJobObject(job, 1)
                    job_api.ExitProcess(1)
                else:
                    _linux_stop_and_reap(payload)
                    os._exit(1)
    except BaseException as exc:
        if not status_sent and status_fd is not None:
            try:
                _write_packet(status_fd, _host_error(stage, exc))
            except BaseException:
                pass
        if payload is not None and os.name != "nt":
            while True:
                try:
                    _linux_stop_and_reap(payload)
                    break
                except OSError:
                    # Cleanup failure must leave the observer alive, not let a
                    # caller mistake its exit for confirmed containment teardown.
                    time.sleep(0.05)
        if os.name == "nt" and job is not None:
            job_api.ExitProcess(127)
        return 127
    finally:
        if status_fd is not None:
            try:
                os.close(status_fd)
            except OSError:
                pass
        try:
            os.close(control_fd)
        except OSError:
            pass


if __name__ == "__main__" and len(sys.argv) == 4 and sys.argv[1] == _HOST_FLAG:
    sys.exit(_host_main(sys.argv[2], sys.argv[3]))