Add crash-safe Codex native process teardown (#1252)

* Add crash-safe Codex native process registry Co-authored-by: omnigent <noreply@omnigent.ai> * Guard Codex crash reap with owner liveness Co-authored-by: omnigent <noreply@omnigent.ai> --------- Co-authored-by: omnigent <noreply@omnigent.ai>
2026-06-25 22:23:44 +07:00
parent e560384a3d
commit 80955e278a
3 changed files with 725 additions and 10 deletions
@@ -11,6 +11,7 @@ import re
 import shlex
 import sys
 import tempfile
+import uuid
 from collections.abc import AsyncIterator
 from dataclasses import dataclass
 from pathlib import Path
@@ -22,6 +23,14 @@ if TYPE_CHECKING:
    from omnigent.onboarding.provider_config import ProviderEntry

 from omnigent.codex_native_bridge import write_policy_hook_config
+from omnigent.codex_native_process_registry import (
+    CodexNativeProcessOwnerLock,
+    acquire_codex_native_process_owner_lock,
+    codex_native_session_tag_cmdline_arg,
+    reconcile_codex_native_process_registry,
+    register_codex_native_process,
+    unregister_codex_native_process,
+)
 from omnigent.inner import _proc
 from omnigent.inner.codex_executor import (
    _clean_codex_env,
@@ -503,6 +512,8 @@ class CodexNativeAppServer:
    policy_hook_disabled_reason: str | None = None
    policy_notice_pending: bool = False
    pinned_model: str | None = None
+    process_registry_tag: str | None = None
+    process_owner_lock: CodexNativeProcessOwnerLock | None = None

    async def start(self) -> None:
        """
@@ -557,9 +568,15 @@ class CodexNativeAppServer:
                    ap_server_url=self.ap_server_url,
                    ap_auth_headers=self.ap_auth_headers or {},
                )
+        reconcile_codex_native_process_registry()
        resolved_listen = self.listen_url or f"unix://{self.socket_path}"
+        self.process_registry_tag = f"codex-native-{uuid.uuid4().hex}"
+        tagged_argv0 = (
+            f"{Path(self.codex_path).name} "
+            f"{codex_native_session_tag_cmdline_arg(self.process_registry_tag)}"
+        )
        argv = [
-            self.codex_path,
+            tagged_argv0,
            "app-server",
            "--listen",
            resolved_listen,
@@ -567,15 +584,30 @@ class CodexNativeAppServer:
        for override in self.config_overrides:
            argv.extend(["-c", override])
        proc_env = {**self.env, "CODEX_HOME": str(self.codex_home)}
-        self.proc = await asyncio.create_subprocess_exec(
-            *argv,
-            stdin=asyncio.subprocess.DEVNULL,
-            stdout=asyncio.subprocess.DEVNULL,
-            stderr=asyncio.subprocess.PIPE,
-            env=proc_env,
-            cwd=str(self.cwd),
-            **_proc.spawn_kwargs(),
-        )
+        self.process_owner_lock = acquire_codex_native_process_owner_lock()
+        try:
+            self.proc = await asyncio.create_subprocess_exec(
+                *argv,
+                stdin=asyncio.subprocess.DEVNULL,
+                stdout=asyncio.subprocess.DEVNULL,
+                stderr=asyncio.subprocess.PIPE,
+                env=proc_env,
+                cwd=str(self.cwd),
+                executable=self.codex_path,
+                **_proc.spawn_kwargs(),
+            )
+        except BaseException:
+            if self.process_owner_lock is not None:
+                self.process_owner_lock.close()
+                self.process_owner_lock = None
+            raise
+        if self.process_owner_lock is not None:
+            register_codex_native_process(
+                pid=self.proc.pid,
+                pgid=_process_group_id(self.proc),
+                session_tag=self.process_registry_tag,
+                owner_lock_path=self.process_owner_lock.path,
+            )
        self.recent_stderr = []
        self.stderr_task = asyncio.create_task(
            self._stderr_loop(),
@@ -712,12 +744,18 @@ class CodexNativeAppServer:
            except asyncio.TimeoutError:
                _kill_process_tree(self.proc)
                await self.proc.wait()
+        if self.process_registry_tag is not None:
+            unregister_codex_native_process(self.process_registry_tag)
+        if self.process_owner_lock is not None:
+            self.process_owner_lock.close()
        if self.stderr_task is not None:
            self.stderr_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await self.stderr_task
        self.proc = None
        self.stderr_task = None
+        self.process_registry_tag = None
+        self.process_owner_lock = None

    async def _wait_until_ready(self) -> None:
        """
@@ -1575,6 +1613,19 @@ def _terminate_process_tree(process: asyncio.subprocess.Process) -> None:
    _proc.terminate_tree(process)


+def _process_group_id(process: asyncio.subprocess.Process) -> int:
+    """
+    Return the child process group id used for crash-safe reaping.
+
+    :param process: Subprocess handle.
+    :returns: Process group id, falling back to pid on non-POSIX hosts.
+    """
+    if os.name == "posix":
+        with contextlib.suppress(ProcessLookupError, PermissionError, OSError):
+            return os.getpgid(process.pid)
+    return process.pid
+
+
 def _kill_process_tree(process: asyncio.subprocess.Process) -> None:
    """
    Send SIGKILL to a subprocess process group when possible.
@@ -0,0 +1,417 @@
+"""Crash-safe process registry for native Codex app-server children."""
+
+from __future__ import annotations
+
+import contextlib
+import json
+import logging
+import os
+import signal
+import subprocess
+import uuid
+from collections.abc import Generator
+from dataclasses import asdict, dataclass
+from pathlib import Path
+
+from omnigent.codex_native_state import _codex_native_state_root
+
+try:
+    import fcntl
+except ImportError:  # pragma: no cover - Windows has no flock.
+    fcntl = None  # type: ignore[assignment]
+
+_logger = logging.getLogger(__name__)
+_REGISTRY_FILE = "process-registry.json"
+_OWNER_LOCK_DIR = "process-owners"
+_TAG_ARG_PREFIX = "omnigent_crash_teardown_tag="
+
+
+@dataclass(frozen=True)
+class CodexNativeProcessEntry:
+    """
+    One crash-reapable native Codex subprocess registry entry.
+
+    :param pid: Child process id.
+    :param pgid: Child process group id.
+    :param tmux_session_name: Optional tmux session name owned by the child.
+    :param session_tag: Unique tag also embedded in the child command line.
+    :param owner_lock_path: Lock file held by the parent while it owns
+        the child. If the lock is still held during reconciliation, the
+        child is a live sibling and must not be reaped.
+    """
+
+    pid: int
+    pgid: int
+    tmux_session_name: str | None
+    session_tag: str
+    owner_lock_path: str | None = None
+
+
+@dataclass
+class CodexNativeProcessOwnerLock:
+    """
+    Kernel-backed liveness handle for a native Codex launcher process.
+
+    :param path: Path to the owner lock file.
+    :param fd: Open file descriptor holding an exclusive flock.
+    """
+
+    path: Path
+    fd: int
+
+    def close(self) -> None:
+        """
+        Release the owner lock.
+
+        :returns: None.
+        """
+        with contextlib.suppress(OSError):
+            fcntl.flock(self.fd, fcntl.LOCK_UN)
+        with contextlib.suppress(OSError):
+            os.close(self.fd)
+        with contextlib.suppress(OSError):
+            self.path.unlink()
+
+
+def codex_native_process_registry_path() -> Path:
+    """
+    Return the stable on-disk registry file path.
+
+    :returns: Registry path under the existing codex-native state root.
+    """
+    return _codex_native_state_root() / _REGISTRY_FILE
+
+
+def acquire_codex_native_process_owner_lock() -> CodexNativeProcessOwnerLock | None:
+    """
+    Acquire a per-launcher owner lock for crash-safe reconciliation.
+
+    The lock is intentionally held by an open file descriptor for the
+    launcher process lifetime. If the launcher crashes, the OS releases
+    the flock, making its children eligible for the next reconciliation
+    sweep. A healthy concurrent launcher still holds the lock, so its
+    child entries are skipped even when their child PID/tag are live.
+
+    :returns: Held owner lock, or ``None`` if it could not be created.
+    """
+    if fcntl is None:
+        return None
+    root = _codex_native_state_root() / _OWNER_LOCK_DIR
+    try:
+        root.mkdir(mode=0o700, parents=True, exist_ok=True)
+        path = root / f"{uuid.uuid4().hex}.lock"
+        fd = os.open(path, os.O_CREAT | os.O_RDWR | getattr(os, "O_CLOEXEC", 0), 0o600)
+        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except OSError:
+        _logger.warning("codex-native process owner lock create failed", exc_info=True)
+        with contextlib.suppress(NameError, OSError):
+            os.close(fd)
+        return None
+    return CodexNativeProcessOwnerLock(path=path, fd=fd)
+
+
+def codex_native_session_tag_cmdline_arg(session_tag: str) -> str:
+    """
+    Return an inert command-line marker carrying the crash-reap tag.
+
+    :param session_tag: Unique per-process tag.
+    :returns: Command-line marker value.
+    :raises ValueError: If *session_tag* is empty.
+    """
+    if not session_tag:
+        raise ValueError("session_tag must be non-empty")
+    return f"{_TAG_ARG_PREFIX}{session_tag}"
+
+
+def register_codex_native_process(
+    *,
+    pid: int,
+    pgid: int,
+    session_tag: str,
+    owner_lock_path: Path | str | None,
+    tmux_session_name: str | None = None,
+    registry_path: Path | None = None,
+) -> None:
+    """
+    Add or replace one native Codex process registry entry.
+
+    :param pid: Child process id.
+    :param pgid: Child process group id.
+    :param session_tag: Unique tag embedded in the child command line.
+    :param owner_lock_path: Lock file held by the launcher process.
+    :param tmux_session_name: Optional tmux session name owned by the child.
+    :param registry_path: Test override for the registry file path.
+    :returns: None.
+    """
+    if pid <= 0 or pgid <= 0 or not session_tag:
+        return
+    entry = CodexNativeProcessEntry(
+        pid=pid,
+        pgid=pgid,
+        tmux_session_name=tmux_session_name,
+        session_tag=session_tag,
+        owner_lock_path=str(owner_lock_path) if owner_lock_path is not None else None,
+    )
+    path = registry_path or codex_native_process_registry_path()
+    with _registry_lock(path):
+        entries = [
+            existing for existing in _read_registry(path) if existing.session_tag != session_tag
+        ]
+        entries.append(entry)
+        _write_registry(path, entries)
+
+
+def unregister_codex_native_process(
+    session_tag: str,
+    *,
+    registry_path: Path | None = None,
+) -> None:
+    """
+    Remove one native Codex process registry entry.
+
+    :param session_tag: Unique per-process registry tag.
+    :param registry_path: Test override for the registry file path.
+    :returns: None.
+    """
+    if not session_tag:
+        return
+    path = registry_path or codex_native_process_registry_path()
+    with _registry_lock(path):
+        entries = [entry for entry in _read_registry(path) if entry.session_tag != session_tag]
+        _write_registry(path, entries)
+
+
+def reconcile_codex_native_process_registry(*, registry_path: Path | None = None) -> None:
+    """
+    Reap crash-leftover native Codex children recorded by prior runs.
+
+    PID reuse is guarded by requiring the live process command line to
+    still contain the entry's unique session tag before killing anything.
+
+    :param registry_path: Test override for the registry file path.
+    :returns: None.
+    """
+    path = registry_path or codex_native_process_registry_path()
+    with _registry_lock(path):
+        survivors: list[CodexNativeProcessEntry] = []
+        for entry in _read_registry(path):
+            if _owner_lock_held(entry.owner_lock_path):
+                survivors.append(entry)
+                continue
+            if not _pid_alive(entry.pid):
+                continue
+            if not _process_cmdline_has_tag(entry.pid, entry.session_tag):
+                continue
+            if not _terminate_process_group(entry.pgid):
+                survivors.append(entry)
+                continue
+            _reap_tmux_session(entry.tmux_session_name)
+        _write_registry(path, survivors)
+
+
+@contextlib.contextmanager
+def _registry_lock(path: Path) -> Generator[None, None, None]:
+    """
+    Serialize the read-modify-write on the shared registry file.
+
+    The registry is a single host-global file mutated by every concurrent
+    launcher, so an unlocked read-modify-write can drop an entry written by
+    another launcher between its read and its write — leaving an orphan that
+    crash reconciliation can never reap. An exclusive flock on a sibling lock
+    file makes the whole sequence atomic across processes. Degrades to a no-op
+    when locking is unavailable (Windows, or a lock-file failure).
+
+    :param path: Registry file path being mutated.
+    :returns: Context manager guarding the mutation.
+    """
+    if fcntl is None:
+        yield
+        return
+    fd = None
+    try:
+        path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+        fd = os.open(str(path) + ".lock", os.O_CREAT | os.O_RDWR, 0o600)
+        fcntl.flock(fd, fcntl.LOCK_EX)
+    except OSError:
+        _logger.warning("codex-native process registry lock failed", exc_info=True)
+        if fd is not None:
+            with contextlib.suppress(OSError):
+                os.close(fd)
+            fd = None
+    try:
+        yield
+    finally:
+        if fd is not None:
+            with contextlib.suppress(OSError):
+                fcntl.flock(fd, fcntl.LOCK_UN)
+            with contextlib.suppress(OSError):
+                os.close(fd)
+
+
+def _read_registry(path: Path) -> list[CodexNativeProcessEntry]:
+    try:
+        raw = path.read_text(encoding="utf-8")
+    except FileNotFoundError:
+        return []
+    except OSError:
+        _logger.warning("codex-native process registry read failed", exc_info=True)
+        return []
+    try:
+        payload = json.loads(raw)
+    except json.JSONDecodeError:
+        _logger.warning("codex-native process registry JSON is malformed; ignoring")
+        return []
+    if not isinstance(payload, list):
+        return []
+    entries: list[CodexNativeProcessEntry] = []
+    for item in payload:
+        entry = _entry_from_json(item)
+        if entry is not None:
+            entries.append(entry)
+    return entries
+
+
+def _write_registry(path: Path, entries: list[CodexNativeProcessEntry]) -> None:
+    try:
+        path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+        payload = [asdict(entry) for entry in entries]
+        tmp = path.with_suffix(".json.tmp")
+        tmp.write_text(json.dumps(payload, separators=(",", ":")) + "\n", encoding="utf-8")
+        os.replace(tmp, path)
+    except OSError:
+        _logger.warning("codex-native process registry write failed", exc_info=True)
+
+
+def _entry_from_json(item: object) -> CodexNativeProcessEntry | None:
+    if not isinstance(item, dict):
+        return None
+    pid = item.get("pid")
+    pgid = item.get("pgid")
+    session_tag = item.get("session_tag")
+    tmux_session_name = item.get("tmux_session_name")
+    owner_lock_path = item.get("owner_lock_path")
+    if not isinstance(pid, int) or pid <= 0:
+        return None
+    if not isinstance(pgid, int) or pgid <= 0:
+        return None
+    if not isinstance(session_tag, str) or not session_tag:
+        return None
+    if tmux_session_name is not None and not isinstance(tmux_session_name, str):
+        tmux_session_name = None
+    if owner_lock_path is not None and not isinstance(owner_lock_path, str):
+        owner_lock_path = None
+    return CodexNativeProcessEntry(
+        pid=pid,
+        pgid=pgid,
+        tmux_session_name=tmux_session_name,
+        session_tag=session_tag,
+        owner_lock_path=owner_lock_path,
+    )
+
+
+def _owner_lock_held(owner_lock_path: str | None) -> bool:
+    if not owner_lock_path:
+        return False
+    if fcntl is None:
+        return False
+    try:
+        fd = os.open(owner_lock_path, os.O_RDWR)
+    except FileNotFoundError:
+        return False
+    except OSError:
+        return True
+    try:
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            return True
+        except OSError:
+            return True
+        return False
+    finally:
+        with contextlib.suppress(OSError):
+            fcntl.flock(fd, fcntl.LOCK_UN)
+        with contextlib.suppress(OSError):
+            os.close(fd)
+
+
+def _pid_alive(pid: int) -> bool:
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True
+    except OSError:
+        return False
+    return True
+
+
+def _process_cmdline_has_tag(pid: int, session_tag: str) -> bool:
+    needle = codex_native_session_tag_cmdline_arg(session_tag)
+    cmdline = _process_cmdline(pid)
+    return needle in cmdline
+
+
+def _process_cmdline(pid: int) -> str:
+    proc_cmdline = Path("/proc") / str(pid) / "cmdline"
+    with contextlib.suppress(OSError):
+        raw = proc_cmdline.read_bytes()
+        if raw:
+            return raw.replace(b"\x00", b" ").decode("utf-8", errors="replace")
+    try:
+        proc = subprocess.run(
+            ["ps", "-p", str(pid), "-ww", "-o", "command="],
+            check=False,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.DEVNULL,
+            text=True,
+            timeout=2.0,
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        return ""
+    return proc.stdout.strip() if proc.returncode == 0 else ""
+
+
+def _terminate_process_group(pgid: int) -> bool:
+    if os.name == "posix":
+        try:
+            os.killpg(pgid, signal.SIGTERM)
+        except ProcessLookupError:
+            return True
+        except (PermissionError, OSError):
+            return False
+        return True
+    return False
+
+
+def _reap_tmux_session(tmux_session_name: str | None) -> None:
+    if not tmux_session_name:
+        return
+    if _tmux_session_exists(tmux_session_name):
+        _kill_tmux_session(tmux_session_name)
+
+
+def _tmux_session_exists(tmux_session_name: str) -> bool:
+    try:
+        proc = subprocess.run(
+            ["tmux", "has-session", "-t", tmux_session_name],
+            check=False,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            timeout=2.0,
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        return False
+    return proc.returncode == 0
+
+
+def _kill_tmux_session(tmux_session_name: str) -> None:
+    with contextlib.suppress(OSError, subprocess.TimeoutExpired):
+        subprocess.run(
+            ["tmux", "kill-session", "-t", tmux_session_name],
+            check=False,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            timeout=2.0,
+        )
@@ -0,0 +1,247 @@
+"""Tests for crash-safe native Codex process registry reconciliation."""
+
+from __future__ import annotations
+
+import json
+import signal
+from pathlib import Path
+
+import pytest
+
+from omnigent import codex_native_process_registry as registry
+
+fcntl = pytest.importorskip("fcntl")
+
+
+def _registry_payload(path: Path) -> list[dict[str, object]]:
+    """
+    Return the raw JSON registry payload.
+
+    :param path: Registry file path.
+    :returns: Parsed registry entries.
+    """
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def test_registry_add_remove_round_trip(tmp_path: Path) -> None:
+    """Registry writes and removes a tagged codex child entry."""
+    path = tmp_path / "registry.json"
+
+    registry.register_codex_native_process(
+        pid=123,
+        pgid=456,
+        tmux_session_name="omnigent-codex-123",
+        session_tag="tag-123",
+        owner_lock_path=tmp_path / "owner.lock",
+        registry_path=path,
+    )
+
+    assert _registry_payload(path) == [
+        {
+            "pid": 123,
+            "pgid": 456,
+            "tmux_session_name": "omnigent-codex-123",
+            "session_tag": "tag-123",
+            "owner_lock_path": str(tmp_path / "owner.lock"),
+        }
+    ]
+
+    registry.unregister_codex_native_process("tag-123", registry_path=path)
+
+    assert _registry_payload(path) == []
+
+
+def test_reconciliation_reaps_alive_tagged_process(tmp_path: Path, monkeypatch) -> None:
+    """A live process with the matching cmdline tag is reaped by process group."""
+    path = tmp_path / "registry.json"
+    registry.register_codex_native_process(
+        pid=123,
+        pgid=456,
+        session_tag="tag-123",
+        owner_lock_path=None,
+        registry_path=path,
+    )
+    killed: list[tuple[int, signal.Signals]] = []
+    monkeypatch.setattr(registry, "_pid_alive", lambda pid: pid == 123)
+    monkeypatch.setattr(
+        registry,
+        "_process_cmdline",
+        lambda _pid: "codex omnigent_crash_teardown_tag=tag-123 app-server",
+    )
+    monkeypatch.setattr(registry.os, "killpg", lambda pgid, sig: killed.append((pgid, sig)))
+
+    registry.reconcile_codex_native_process_registry(registry_path=path)
+
+    assert killed == [(456, signal.SIGTERM)]
+    assert _registry_payload(path) == []
+
+
+def test_reconciliation_skips_pid_reuse_without_matching_tag(tmp_path: Path, monkeypatch) -> None:
+    """A reused PID is never killed when the cmdline lacks the session tag."""
+    path = tmp_path / "registry.json"
+    registry.register_codex_native_process(
+        pid=123,
+        pgid=456,
+        session_tag="tag-123",
+        owner_lock_path=None,
+        registry_path=path,
+    )
+    killed: list[tuple[int, signal.Signals]] = []
+    monkeypatch.setattr(registry, "_pid_alive", lambda pid: pid == 123)
+    monkeypatch.setattr(registry, "_process_cmdline", lambda _pid: "python unrelated.py")
+    monkeypatch.setattr(registry.os, "killpg", lambda pgid, sig: killed.append((pgid, sig)))
+
+    registry.reconcile_codex_native_process_registry(registry_path=path)
+
+    assert killed == []
+    assert _registry_payload(path) == []
+
+
+def test_reconciliation_skips_live_sibling_when_owner_lock_is_held(
+    tmp_path: Path, monkeypatch
+) -> None:
+    """A healthy sibling child is not reaped while its launcher owns the lock."""
+    path = tmp_path / "registry.json"
+    owner_lock = tmp_path / "owner.lock"
+    registry.register_codex_native_process(
+        pid=123,
+        pgid=456,
+        session_tag="tag-123",
+        owner_lock_path=owner_lock,
+        registry_path=path,
+    )
+    killed: list[tuple[int, signal.Signals]] = []
+    monkeypatch.setattr(registry, "_owner_lock_held", lambda value: value == str(owner_lock))
+    monkeypatch.setattr(registry, "_pid_alive", lambda pid: pid == 123)
+    monkeypatch.setattr(
+        registry,
+        "_process_cmdline",
+        lambda _pid: "codex omnigent_crash_teardown_tag=tag-123 app-server",
+    )
+    monkeypatch.setattr(registry.os, "killpg", lambda pgid, sig: killed.append((pgid, sig)))
+
+    registry.reconcile_codex_native_process_registry(registry_path=path)
+
+    assert killed == []
+    assert _registry_payload(path) == [
+        {
+            "pid": 123,
+            "pgid": 456,
+            "tmux_session_name": None,
+            "session_tag": "tag-123",
+            "owner_lock_path": str(owner_lock),
+        }
+    ]
+
+
+def test_reconciliation_reaps_when_owner_lock_is_not_held(tmp_path: Path, monkeypatch) -> None:
+    """A tagged child is reaped after its owning launcher lock is gone."""
+    path = tmp_path / "registry.json"
+    owner_lock = tmp_path / "owner.lock"
+    registry.register_codex_native_process(
+        pid=123,
+        pgid=456,
+        session_tag="tag-123",
+        owner_lock_path=owner_lock,
+        registry_path=path,
+    )
+    killed: list[tuple[int, signal.Signals]] = []
+    monkeypatch.setattr(registry, "_owner_lock_held", lambda _value: False)
+    monkeypatch.setattr(registry, "_pid_alive", lambda pid: pid == 123)
+    monkeypatch.setattr(
+        registry,
+        "_process_cmdline",
+        lambda _pid: "codex omnigent_crash_teardown_tag=tag-123 app-server",
+    )
+    monkeypatch.setattr(registry.os, "killpg", lambda pgid, sig: killed.append((pgid, sig)))
+
+    registry.reconcile_codex_native_process_registry(registry_path=path)
+
+    assert killed == [(456, signal.SIGTERM)]
+    assert _registry_payload(path) == []
+
+
+def test_reconciliation_drops_dead_pids(tmp_path: Path, monkeypatch) -> None:
+    """Dead process entries are discarded without kill attempts."""
+    path = tmp_path / "registry.json"
+    registry.register_codex_native_process(
+        pid=123,
+        pgid=456,
+        session_tag="tag-123",
+        owner_lock_path=None,
+        registry_path=path,
+    )
+    killed: list[tuple[int, signal.Signals]] = []
+    monkeypatch.setattr(registry, "_pid_alive", lambda _pid: False)
+    monkeypatch.setattr(registry.os, "killpg", lambda pgid, sig: killed.append((pgid, sig)))
+
+    registry.reconcile_codex_native_process_registry(registry_path=path)
+
+    assert killed == []
+    assert _registry_payload(path) == []
+
+
+def test_tmux_session_reaped_only_when_recorded_name_exists(tmp_path: Path, monkeypatch) -> None:
+    """Matching tagged process reaps only the recorded existing tmux session."""
+    path = tmp_path / "registry.json"
+    registry.register_codex_native_process(
+        pid=123,
+        pgid=456,
+        tmux_session_name="omnigent-codex-live",
+        session_tag="tag-live",
+        owner_lock_path=None,
+        registry_path=path,
+    )
+    registry.register_codex_native_process(
+        pid=124,
+        pgid=457,
+        tmux_session_name="omnigent-codex-missing",
+        session_tag="tag-missing",
+        owner_lock_path=None,
+        registry_path=path,
+    )
+    killed_tmux: list[str] = []
+    monkeypatch.setattr(registry, "_pid_alive", lambda pid: pid in {123, 124})
+    monkeypatch.setattr(
+        registry,
+        "_process_cmdline",
+        lambda pid: (
+            "codex "
+            f"omnigent_crash_teardown_tag=tag-{'live' if pid == 123 else 'missing'} "
+            "app-server"
+        ),
+    )
+    monkeypatch.setattr(registry.os, "killpg", lambda _pgid, _sig: None)
+    monkeypatch.setattr(
+        registry,
+        "_tmux_session_exists",
+        lambda name: name == "omnigent-codex-live",
+    )
+    monkeypatch.setattr(registry, "_kill_tmux_session", lambda name: killed_tmux.append(name))
+
+    registry.reconcile_codex_native_process_registry(registry_path=path)
+
+    assert killed_tmux == ["omnigent-codex-live"]
+    assert _registry_payload(path) == []
+
+
+def test_owner_lock_liveness_round_trip(tmp_path: Path, monkeypatch) -> None:
+    """A held owner lock reads as held; releasing it makes the entry reapable."""
+    monkeypatch.setattr(registry, "_codex_native_state_root", lambda: tmp_path)
+    lock = registry.acquire_codex_native_process_owner_lock()
+    assert lock is not None
+    assert registry._owner_lock_held(str(lock.path)) is True
+    lock.close()
+    assert registry._owner_lock_held(str(lock.path)) is False
+
+
+def test_registry_lock_serializes_read_modify_write(tmp_path: Path) -> None:
+    """The registry lock is exclusive across the read-modify-write window."""
+    path = tmp_path / "registry.json"
+    with registry._registry_lock(path):
+        fd = registry.os.open(str(path) + ".lock", registry.os.O_RDWR)
+        try:
+            with pytest.raises(BlockingIOError):
+                fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        finally:
+            registry.os.close(fd)