fix(platform-ext): wait for the instance flock before reporting a stop

packages/nemo_platform_ext/src/nemo_platform_ext/local/process.py

@@ -60,6 +60,10 @@
 _SIGTERM_POLL_INTERVAL = 0.25
 _SIGKILL_WAIT_TIMEOUT = 5.0
 _DEFAULT_STOP_TIMEOUT = 30.0
+# How long ``stop_instance`` waits for the flock to be released after the last process it
+# signaled has exited.  The kernel drops the lock as part of closing the fd during process
+# teardown, so this only has to cover the tail of that teardown, not a whole shutdown.
+_LOCK_RELEASE_TIMEOUT = 5.0
 
 
 def _pause(seconds: float) -> None:
@@ -569,9 +573,14 @@ def _snapshot_children(pid: int) -> list[psutil.Process]:
 def _sweep_orphans(children: list[psutil.Process], timeout: float = 5.0) -> list[int]:
     """Terminate any still-alive processes from a prior snapshot.
 
-    Sends SIGTERM, waits up to *timeout*, then SIGKILL survivors.
+    Sends SIGTERM, waits up to *timeout*, then SIGKILL survivors and waits again.
     Returns PIDs that were signaled.  Handles ``NoSuchProcess`` gracefully
     since children may have already exited during graceful shutdown.
+
+    The second wait matters: these children inherited the instance flock fd from the
+    parent, and the kernel only releases the lock once the last holder's fd is closed.
+    Returning while a SIGKILLed child is still being torn down leaves the lock held, so
+    an ``is_instance_alive`` probe immediately after a "successful" stop reports True.
     """
     alive_children = [c for c in children if c.is_running()]
     if not alive_children:
@@ -586,12 +595,19 @@ def _sweep_orphans(children: list[psutil.Process], timeout: float = 5.0) -> list
             pass  # Already exited or not owned by us — skip.
 
     _, still_alive = psutil.wait_procs(alive_children, timeout=timeout)
+    kill_sent: list[psutil.Process] = []
     for child in still_alive:
         try:
             child.kill()
+            kill_sent.append(child)
         except (psutil.NoSuchProcess, psutil.AccessDenied):
             pass  # Raced with exit or not owned — nothing to do.
 
+    if kill_sent:
+        _, unreaped = psutil.wait_procs(kill_sent, timeout=timeout)
+        if unreaped:
+            logger.warning("Orphaned child %s still alive after SIGKILL", [c.pid for c in unreaped])
+
     if killed:
         logger.info(
             "Swept %d orphaned child %s: %s", len(killed), "process" if len(killed) == 1 else "processes", killed
@@ -673,10 +689,32 @@ def stop_instance(
 
     swept = _sweep_orphans(children) if children else []
 
+    # A process exiting and its flock being released are not the same instant: the kernel
+    # drops the lock while closing fds during teardown.  Callers treat a successful stop as
+    # "the scope is free now" (and immediately probe with ``is_instance_alive``), so hold the
+    # post-condition here rather than making every caller poll for it.
+    if not _wait_for_lock_release(scope, base_dir=base_dir, timeout=_LOCK_RELEASE_TIMEOUT):
+        logger.warning(
+            "Instance '%s' flock still held %.0fs after pid %d exited; a child may have outlived the sweep",
+            scope,
+            _LOCK_RELEASE_TIMEOUT,
+            pid,
+        )
+
     remove_descriptor(scope, base_dir=base_dir)
     return StopResult(stopped_pids=[pid], swept_children=swept)
 
 
+def _wait_for_lock_release(scope: str, *, base_dir: Path | None, timeout: float) -> bool:
+    """Poll until the instance flock is free, or *timeout* elapses."""
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if not is_instance_alive(scope, base_dir=base_dir):
+            return True
+        _pause(_SIGTERM_POLL_INTERVAL)
+    return not is_instance_alive(scope, base_dir=base_dir)
+
+
 def _wait_for_pid_exit(pid: int, *, timeout: float) -> bool:
     deadline = time.monotonic() + timeout
     while time.monotonic() < deadline:

packages/nemo_platform_ext/tests/cli/commands/test_services_process.py

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import logging
 import os
 import signal
 import subprocess
@@ -845,6 +846,49 @@ def test_escalates_to_sigkill(self) -> None:
                 proc.kill()
                 proc.wait(timeout=5)
 
+    def test_warns_when_child_survives_sigkill(self, caplog: pytest.LogCaptureFixture) -> None:
+        """A child that outlives SIGKILL is reported rather than silently dropped.
+
+        Nothing survives SIGKILL on demand — the real trigger is a process wedged
+        in uninterruptible sleep, which a test cannot arrange — so `wait_procs` is
+        stubbed to keep reporting the child as alive. The terminate/kill calls are
+        still real; only the observation is simulated.
+        """
+        proc = subprocess.Popen(
+            [
+                sys.executable,
+                "-c",
+                "import signal, time; signal.signal(signal.SIGTERM, signal.SIG_IGN); time.sleep(60)",
+            ],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        try:
+            time.sleep(0.3)
+            ps_child = psutil.Process(proc.pid)
+            # The first wait drives the SIGTERM -> SIGKILL escalation; the second
+            # is the one whose non-empty result trips the warning.
+            with (
+                patch.object(
+                    process_module.psutil,
+                    "wait_procs",
+                    side_effect=[([], [ps_child]), ([], [ps_child])],
+                ) as wait_procs,
+                caplog.at_level(logging.WARNING, logger=process_module.logger.name),
+            ):
+                killed = _sweep_orphans([ps_child], timeout=0.1)
+
+            # Two waits, not one. The second is the post-SIGKILL wait this PR adds,
+            # so a single-wait implementation must not be able to satisfy this test.
+            assert wait_procs.call_count == 2
+            assert proc.pid in killed
+            assert f"Orphaned child [{proc.pid}] still alive after SIGKILL" in caplog.text
+            proc.wait(timeout=5)
+        finally:
+            if proc.poll() is None:
+                proc.kill()
+                proc.wait(timeout=5)
+
 
 # ---------------------------------------------------------------------------
 # stop_instance — child sweep integration

packages/nemo_platform_ext/tests/local/test_daemon_lifecycle.py

@@ -376,6 +376,149 @@ def test_stop_instance_sweeps_orphaned_children(tmp_path: Path) -> None:
                 pass
 
 
+def test_wait_for_lock_release_blocks_until_holder_exits(tmp_path: Path) -> None:
+    """``_wait_for_lock_release`` should return only once the flock is actually free."""
+    scope = "unit-lock-release"
+    state_dir = tmp_path / "state"
+    lock_path = state_dir / "instances" / scope / process.LOCK_FILENAME
+    lock_path.parent.mkdir(parents=True)
+    lock_path.touch()
+
+    # A holder that keeps the flock for ~1s, then exits and releases it.
+    holder = subprocess.Popen(
+        [
+            sys.executable,
+            "-c",
+            "import fcntl, sys, time; "
+            "fd = open(sys.argv[1], 'r+'); "
+            "fcntl.flock(fd, fcntl.LOCK_EX); "
+            "print('held', flush=True); "
+            "time.sleep(1.0)",
+            str(lock_path),
+        ],
+        stdout=subprocess.PIPE,
+        text=True,
+    )
+    try:
+        assert holder.stdout is not None
+        assert holder.stdout.readline().strip() == "held"
+        assert process.is_instance_alive(scope, base_dir=state_dir)
+
+        started = time.monotonic()
+        assert process._wait_for_lock_release(scope, base_dir=state_dir, timeout=10.0)
+        assert not process.is_instance_alive(scope, base_dir=state_dir)
+        # It waited rather than returning eagerly on a still-held lock.
+        assert time.monotonic() - started >= 0.5
+    finally:
+        holder.kill()
+        holder.wait(timeout=5)
+
+
+def test_wait_for_lock_release_times_out_while_held(tmp_path: Path) -> None:
+    """A lock held for the whole window should report failure, not success."""
+    scope = "unit-lock-held"
+    state_dir = tmp_path / "state"
+    lock_path = state_dir / "instances" / scope / process.LOCK_FILENAME
+    lock_path.parent.mkdir(parents=True)
+    lock_path.touch()
+
+    holder = subprocess.Popen(
+        [
+            sys.executable,
+            "-c",
+            "import fcntl, sys, time; "
+            "fd = open(sys.argv[1], 'r+'); "
+            "fcntl.flock(fd, fcntl.LOCK_EX); "
+            "print('held', flush=True); "
+            "time.sleep(300)",
+            str(lock_path),
+        ],
+        stdout=subprocess.PIPE,
+        text=True,
+    )
+    try:
+        assert holder.stdout is not None
+        assert holder.stdout.readline().strip() == "held"
+        assert not process._wait_for_lock_release(scope, base_dir=state_dir, timeout=0.5)
+    finally:
+        holder.kill()
+        holder.wait(timeout=5)
+
+
+@pytest.mark.integration
+def test_stop_instance_releases_lock_held_by_surviving_child(tmp_path: Path) -> None:
+    """Regression: ``stop_instance`` must not report success while the scope's
+    flock is still held.
+
+    A process exiting and its flock being released are not the same instant — the
+    kernel drops the lock while closing fds during teardown, and any process that
+    inherited the fd keeps it held until *it* is gone too.  ``stop_instance`` used
+    to return as soon as the descriptor PID was dead, so an immediate
+    ``is_instance_alive`` probe raced that teardown and intermittently saw True.
+
+    The holder here is deliberately outside the parent's process tree: that makes
+    the window wide and fixed instead of scheduler-dependent, and it models the
+    case the sweep can't reach (a lock-inheriting process spawned after the child
+    snapshot was taken).
+    """
+    scope = "integ-lock-release"
+    state_dir = tmp_path / "state"
+    lock_path = state_dir / "instances" / scope / process.LOCK_FILENAME
+    lock_path.parent.mkdir(parents=True)
+    lock_path.touch()
+
+    hold_seconds = 2.0
+    holder = subprocess.Popen(
+        [
+            sys.executable,
+            "-c",
+            "import fcntl, sys, time; "
+            "fd = open(sys.argv[1], 'r+'); "
+            "fcntl.flock(fd, fcntl.LOCK_EX); "
+            "print('held', flush=True); "
+            f"time.sleep({hold_seconds})",
+            str(lock_path),
+        ],
+        stdout=subprocess.PIPE,
+        text=True,
+    )
+    # The descriptor PID: exits promptly on SIGTERM, and does not hold the lock.
+    parent = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(300)"], start_new_session=True)
+    try:
+        assert holder.stdout is not None
+        assert holder.stdout.readline().strip() == "held"
+        assert process.is_instance_alive(scope, base_dir=state_dir)
+
+        process.write_descriptor(
+            process.InstanceDescriptor(
+                pid=parent.pid,
+                config=PlatformAppConfig(scope=scope, host="127.0.0.1", port=0, state_root=state_dir),
+                transport="tcp",
+                mode="daemon",
+                create_time=psutil.Process(parent.pid).create_time(),
+            ),
+            base_dir=state_dir,
+        )
+
+        started = time.monotonic()
+        result = process.stop_instance(scope, base_dir=state_dir, timeout=10, force=True)
+        elapsed = time.monotonic() - started
+        assert parent.pid in result.stopped_pids
+
+        # The post-condition callers rely on, checked with no grace period.
+        assert not process.is_instance_alive(scope, base_dir=state_dir)
+        assert process.read_descriptor(scope, base_dir=state_dir) is None
+        # It blocked on the lock rather than returning the moment the PID died.
+        assert elapsed >= hold_seconds / 2
+    finally:
+        for proc in (holder, parent):
+            try:
+                proc.kill()
+                proc.wait(timeout=3)
+            except (OSError, subprocess.TimeoutExpired):
+                pass
+
+
 @pytest.mark.integration
 def test_stop_instance_foreground_mode_requires_force(tmp_path: Path) -> None:
     """Stopping a foreground-mode instance without ``force=True`` should raise

plugins/nemo-deployments/tests/integration/backends/docker/test_docker_backend.py

@@ -93,7 +93,14 @@ def _docker_backend_with_observe_timeout(
     *,
     oneshot_observe_timeout_seconds: int,
 ) -> DockerDeploymentBackend:
-    return _build_docker_backend(oneshot_observe_timeout_seconds=oneshot_observe_timeout_seconds)
+    # `pull_images` is off so the caller can pre-pull and keep the registry
+    # round-trip out of any window it times. `create_deployment` pulls
+    # unconditionally, not just when the image is missing locally, so leaving
+    # this on would put ~2s of Docker Hub latency inside the measurement.
+    return _build_docker_backend(
+        oneshot_observe_timeout_seconds=oneshot_observe_timeout_seconds,
+        pull_images=False,
+    )
 
 
 def _always_http_config() -> DeploymentConfig:
@@ -158,7 +165,13 @@ async def test_never_deployment_succeeds(docker_backend: DockerDeploymentBackend
 @pytest.mark.asyncio
 async def test_never_deployment_outlives_observe_wait_then_succeeds() -> None:
     """Long Never jobs return STARTING on create and finish via read_status polling."""
-    job_sleep_seconds = 5
+    # The job has to outlive the observe wait by a wide enough margin that a
+    # create_deployment which blocked until exit is unmistakable. The margin is
+    # what makes the timing assertion below meaningful, and it has to clear the
+    # cost of container create/start on a contended CI docker daemon (~3s
+    # observed) -- this test shares an xdist loadgroup with the heavier
+    # test_reconcile_docker cases, all hitting the same daemon.
+    job_sleep_seconds = 20
     observe_timeout_seconds = 1
     docker_backend = _docker_backend_with_observe_timeout(
         oneshot_observe_timeout_seconds=observe_timeout_seconds,
@@ -169,8 +182,9 @@ async def test_never_deployment_outlives_observe_wait_then_succeeds() -> None:
     client = docker.from_env()
 
     try:
-        # Warm the image cache so the timed window below measures the observe wait
-        # rather than an uncached image pull.
+        # The backend is built with `pull_images=False`, so this pull is what puts
+        # the image on the host. Doing it here keeps it out of the timed window
+        # below, which is measuring the observe wait.
         await asyncio.to_thread(client.images.pull, ALPINE_IMAGE)
 
         started = time.monotonic()
@@ -185,9 +199,13 @@ async def test_never_deployment_outlives_observe_wait_then_succeeds() -> None:
 
         assert created.status == "STARTING"
         assert "after observe wait" in (created.status_message or "")
-        assert create_elapsed < observe_timeout_seconds + 2.0
+        # Deliberately loose. The STARTING assertions above are what pin "returned
+        # during the observe wait"; this one only has to catch the gross regression
+        # of blocking for the whole job, which would land at ~job_sleep_seconds.
+        # Keeping it well clear of CI jitter is worth more than a tight bound.
+        assert create_elapsed < observe_timeout_seconds + 8.0
 
-        deadline = time.monotonic() + 15.0
+        deadline = time.monotonic() + 45.0
         status = created
         while time.monotonic() < deadline:
             status = await docker_backend.read_status(workspace="itest", name="sleep-job")