ci: prevent orphaned DeepSeek V4 engines on test interruption

.github/workflows/ci.yml

@@ -188,7 +188,7 @@ jobs:
           run_cmd="$run_cmd PTO2_RING_TASK_WINDOW=$PTO2_RING_TASK_WINDOW"
           run_cmd="$run_cmd PTO2_RING_HEAP=$PTO2_RING_HEAP"
           run_cmd="$run_cmd python -m pytest tests/test_qwen3_accuracy.py -q -s"
-          task-submit --device auto --timeout 1200 --max-time 1800 --run "$run_cmd"
+          task-submit --device auto --timeout 3600 --max-time 1800 --run "$run_cmd"
 
       - name: Run Qwen3 serving guard (prefix cache, chunked prefill, multi-batch)
         env:
@@ -210,7 +210,7 @@ jobs:
           run_cmd="$run_cmd PTO2_RING_TASK_WINDOW=$PTO2_RING_TASK_WINDOW"
           run_cmd="$run_cmd PTO2_RING_HEAP=$PTO2_RING_HEAP"
           run_cmd="$run_cmd python -m pytest tests/test_qwen3_serving.py -q -s"
-          task-submit --device auto --timeout 1200 --max-time 1800 --run "$run_cmd"
+          task-submit --device auto --timeout 3600 --max-time 1800 --run "$run_cmd"
 
       - name: Run DeepSeek V4 HTTP generation accuracy guard
         env:
@@ -242,22 +242,45 @@ jobs:
           run_cmd="$run_cmd SERVING_WORKER_STEP_TIMEOUT=$SERVING_WORKER_STEP_TIMEOUT"
           run_cmd="$run_cmd python -m pytest tests/test_deepseek_v4_accuracy.py -q -s"
           task-submit --device auto --device-num 8 --ignore-whitelist \
-            --timeout 1200 --max-time 1800 \
+            --timeout 3600 --max-time 1800 \
             --run "$run_cmd"
 
       - name: Kill orphaned task-submit tasks
         if: always()
         working-directory: .
         run: |
           marker="pypto-serving-ci-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
-          task-submit --list 2>/dev/null \
-            | grep -F "CI_TASK_MARKER=$marker" \
-            | grep -oE 'task_[0-9_]+' \
-            | sort -u \
-            | while read -r task_id; do
-                echo "Killing orphaned task $task_id"
-                task-submit --kill "$task_id" || true
-              done || true
+          target="CI_TASK_MARKER=$marker"
+          echo "cleaning up task-submit tasks matching $target"
+
+          # Match with `--find`, NOT `--list`: --list is a human-facing view that
+          # truncates the command column to 77 chars, so a grep for the full
+          # marker can silently miss and leave orphaned tasks holding devices.
+          # --find matches against the untruncated COMMAND and prints bare
+          # task-ids. It needs task-submit >= 2026-07-14 on the runner.
+          if ! ids=$(task-submit --find "$target"); then
+            echo "::error::task-submit --find failed — is task-submit up to date on this runner?"
+            exit 1
+          fi
+
+          if [ -z "$ids" ]; then
+            echo "  no orphaned tasks"
+            exit 0
+          fi
+
+          # A cleanup step that prints nothing is indistinguishable from one that
+          # is quietly broken — report what we killed.
+          n=0
+          while read -r task_id; do
+            [ -n "$task_id" ] || continue
+            echo "  killing $task_id"
+            if task-submit --kill "$task_id"; then
+              n=$((n + 1))
+            else
+              echo "::warning::failed to kill $task_id"
+            fi
+          done <<< "$ids"
+          echo "killed $n orphaned task(s)"
 
   platform-build:
     runs-on: ubuntu-latest

tests/test_deepseek_v4_accuracy.py

@@ -11,6 +11,8 @@
 
 from __future__ import annotations
 
+import contextlib
+import ctypes
 import io
 import json
 import os
@@ -38,6 +40,112 @@
 OVERALL_TIMEOUT_SECONDS = 1650
 HEARTBEAT_SECONDS = 30
 
+# How long to let the server group settle after SIGTERM before escalating to SIGKILL.
+#
+# On the normal path (test passed or failed on its own) we can afford to be patient.
+#
+# On the interrupted path we cannot. task-submit's watchdog sends SIGTERM to our
+# process group and escalates to SIGKILL 3 seconds later; the --kill path allows
+# 5 (KILL_GRACE). Once that SIGKILL lands, pytest is gone and nobody is left to
+# reap the server — which is exactly how ~800 GiB of orphaned engines accumulated.
+# So when we are being terminated, the whole teardown has to fit inside ~3s.
+NORMAL_TERM_GRACE_SECONDS = 20.0
+SIGNAL_TERM_GRACE_SECONDS = 1.0
+
+# prctl(2) option number. Not exposed by the signal/os modules, so spell it out.
+_PR_SET_PDEATHSIG = 1
+
+# Load libc once, at import time. _set_pdeathsig() runs as a preexec_fn — after
+# fork but before exec — where calling ctypes.CDLL() would dlopen in the child and
+# is unsafe if any other thread in the parent held the loader lock across the fork.
+try:
+    _libc = ctypes.CDLL("libc.so.6", use_errno=True)
+except (OSError, AttributeError):
+    _libc = None
+
+_TERMINATION_SIGNALS = tuple(
+    sig
+    for sig in (
+        getattr(signal, "SIGTERM", None),
+        getattr(signal, "SIGINT", None),
+        getattr(signal, "SIGHUP", None),
+    )
+    if sig is not None
+)
+
+
+class TerminatedBySignal(KeyboardInterrupt):
+    """Raised in the main thread when the CI runner asks this process to stop.
+
+    Subclasses KeyboardInterrupt so pytest treats it as an interrupt and tears the
+    session down, rather than recording it as an ordinary test failure and moving
+    on to the next test.
+    """
+
+    def __init__(self, signum: int) -> None:
+        super().__init__(f"terminated by signal {signum}")
+        self.signum = signum
+
+
+@contextlib.contextmanager
+def _raise_on_termination():
+    """Turn SIGTERM/SIGINT/SIGHUP into an exception so that `finally` blocks run.
+
+    CPython leaves SIGTERM at SIG_DFL, so a plain `kill` tears the interpreter down
+    where it stands: no stack unwind, no `finally`, no `atexit`. The server we
+    started with start_new_session=True is in its own session and process group, so
+    it survives the signal that killed us and is inherited by init — still holding
+    its NPU cards and several hundred GiB of shared memory. Raising from the handler
+    is what gives the teardown below a chance to run at all.
+    """
+    if threading.current_thread() is not threading.main_thread():
+        # signal.signal() is main-thread-only. Under pytest we always are on it;
+        # degrade to a no-op rather than exploding if that ever stops being true.
+        yield
+        return
+
+    def handler(signum: int, _frame) -> None:
+        # Deafen ourselves before unwinding. The runner sends SIGTERM and then
+        # escalates, and a second delivery landing inside the teardown would abort
+        # it halfway — leaving behind exactly the orphan we are trying to prevent.
+        # SIGKILL still gets through; PR_SET_PDEATHSIG below is the answer to that.
+        for sig in _TERMINATION_SIGNALS:
+            with contextlib.suppress(OSError, ValueError):
+                signal.signal(sig, signal.SIG_IGN)
+        raise TerminatedBySignal(signum)
+
+    previous: dict[int, object] = {}
+    for sig in _TERMINATION_SIGNALS:
+        try:
+            previous[sig] = signal.signal(sig, handler)
+        except (OSError, ValueError):
+            continue
+    try:
+        yield
+    finally:
+        for sig, old_handler in previous.items():
+            with contextlib.suppress(OSError, ValueError):
+                signal.signal(sig, old_handler)
+
+
+def _set_pdeathsig() -> None:
+    """Child-side preexec hook: ask the kernel to SIGKILL us if our parent dies.
+
+    Backstop for the one signal we cannot catch. If pytest is SIGKILLed, no handler
+    runs and no teardown happens — but PR_SET_PDEATHSIG is enforced by the kernel,
+    so the server still dies. It only covers the process we fork directly, so the
+    server's own workers can in principle linger; treat this as a safety net under
+    _raise_on_termination(), not as a replacement for it.
+    """
+    if _libc is None:
+        raise OSError("libc.so.6 not loaded")
+    if _libc.prctl(_PR_SET_PDEATHSIG, signal.SIGKILL, 0, 0, 0) != 0:
+        raise OSError(ctypes.get_errno(), "prctl(PR_SET_PDEATHSIG) failed")
+    # Close the fork/prctl race: if the parent already died in that window, the
+    # death signal fired before we asked for it and will never be re-sent.
+    if os.getppid() == 1:
+        os._exit(1)
+
 
 def _task_devices() -> tuple[int, ...]:
     raw_devices = os.environ.get("TASK_DEVICE", "")
@@ -177,7 +285,9 @@ def send_request() -> None:
     raise TimeoutError("DeepSeek completion exceeded the end-to-end timeout")
 
 
-def _stop_process_group(process: subprocess.Popen) -> None:
+def _stop_process_group(
+    process: subprocess.Popen, term_grace: float = NORMAL_TERM_GRACE_SECONDS
+) -> None:
     try:
         os.killpg(process.pid, signal.SIGTERM)
     except ProcessLookupError:
@@ -187,7 +297,7 @@ def _stop_process_group(process: subprocess.Popen) -> None:
         return
 
     try:
-        process.wait(timeout=20)
+        process.wait(timeout=term_grace)
     except subprocess.TimeoutExpired:
         try:
             os.killpg(process.pid, signal.SIGKILL)
@@ -205,8 +315,9 @@ def _stop_process_group(process: subprocess.Popen) -> None:
         return
 
     # The server parent may exit before a worker child. Give the process group a
-    # short grace period, then kill any remaining descendants.
-    shutdown_deadline = time.monotonic() + 2
+    # short grace period, then kill any remaining descendants. Bounded by term_grace
+    # so the interrupted path still fits inside the runner's SIGKILL window.
+    shutdown_deadline = time.monotonic() + min(2.0, term_grace)
     while time.monotonic() < shutdown_deadline:
         try:
             os.killpg(process.pid, 0)
@@ -244,31 +355,50 @@ def test_deepseek_v4_http_completion_matches_expected_text(tmp_path: Path) -> No
     log_path = tmp_path / "deepseek-v4-server.log"
     deadline = time.monotonic() + OVERALL_TIMEOUT_SECONDS
 
-    try:
-        with log_path.open("w", encoding="utf-8") as server_log:
-            process = subprocess.Popen(
-                _server_command(model_dir, devices, port),
-                cwd=ROOT,
-                stdout=server_log,
-                stderr=subprocess.STDOUT,
-                start_new_session=True,
-                text=True,
-            )
-            try:
-                _wait_for_health(process, port, deadline)
-                response = _request_completion(process, port, deadline)
-                print(f"DeepSeek completion response: {response}", flush=True)
-
-                assert response.get("model") == MODEL_ID
-                choices = response.get("choices")
-                assert isinstance(choices, list) and len(choices) == 1
-                assert choices[0].get("text") == EXPECTED_TEXT
-                assert choices[0].get("finish_reason") == "length"
-            finally:
-                _stop_process_group(process)
-    except BaseException:
-        _print_server_log(log_path)
-        raise
+    with _raise_on_termination():
+        try:
+            with log_path.open("w", encoding="utf-8") as server_log:
+                # `process` is bound before the try so that a signal arriving between
+                # fork and assignment still reaches the teardown. Popen's own internals
+                # are covered by PR_SET_PDEATHSIG.
+                process = None
+                interrupted = False
+                try:
+                    process = subprocess.Popen(
+                        _server_command(model_dir, devices, port),
+                        cwd=ROOT,
+                        stdout=server_log,
+                        stderr=subprocess.STDOUT,
+                        start_new_session=True,
+                        preexec_fn=_set_pdeathsig,
+                        text=True,
+                    )
+                    _wait_for_health(process, port, deadline)
+                    response = _request_completion(process, port, deadline)
+                    print(f"DeepSeek completion response: {response}", flush=True)
+
+                    assert response.get("model") == MODEL_ID
+                    choices = response.get("choices")
+                    assert isinstance(choices, list) and len(choices) == 1
+                    assert choices[0].get("text") == EXPECTED_TEXT
+                    assert choices[0].get("finish_reason") == "length"
+                except TerminatedBySignal:
+                    interrupted = True
+                    raise
+                finally:
+                    if process is not None:
+                        # Being torn down by the runner means we have ~3s before SIGKILL,
+                        # so escalate to SIGKILL fast. On the normal path, be patient and
+                        # let the engine shut down cleanly.
+                        _stop_process_group(
+                            process,
+                            term_grace=(
+                                SIGNAL_TERM_GRACE_SECONDS if interrupted else NORMAL_TERM_GRACE_SECONDS
+                            ),
+                        )
+        except BaseException:
+            _print_server_log(log_path)
+            raise
 
 
 def test_completion_http_error_includes_response_body(monkeypatch) -> None:
@@ -309,6 +439,74 @@ def wait(timeout):
     assert "still alive after SIGKILL" in capsys.readouterr().out
 
 
+def test_stop_process_group_honours_term_grace(monkeypatch) -> None:
+    """The interrupted path must escalate to SIGKILL well inside the runner's window."""
+    waits: list[float] = []
+    signals: list[int] = []
+
+    class StuckProcess:
+        pid = 123
+
+        @staticmethod
+        def wait(timeout):
+            waits.append(timeout)
+            raise subprocess.TimeoutExpired("server", timeout)
+
+    monkeypatch.setattr(os, "killpg", lambda _pid, sig: signals.append(sig))
+
+    _stop_process_group(StuckProcess(), term_grace=SIGNAL_TERM_GRACE_SECONDS)
+
+    assert waits[0] == SIGNAL_TERM_GRACE_SECONDS
+    assert signals[:2] == [signal.SIGTERM, signal.SIGKILL]
+
+
+def test_sigterm_unwinds_the_stack_so_teardown_runs() -> None:
+    """The whole point: a bare SIGTERM must not skip `finally`.
+
+    Without _raise_on_termination() this test does not fail — it takes the pytest
+    process down with it, which is precisely the production bug.
+    """
+    torn_down: list[str] = []
+
+    with pytest.raises(TerminatedBySignal) as excinfo:
+        with _raise_on_termination():
+            try:
+                os.kill(os.getpid(), signal.SIGTERM)
+                # The handler runs at the next bytecode boundary; give it one.
+                time.sleep(1)
+            finally:
+                torn_down.append("server stopped")
+
+    assert excinfo.value.signum == signal.SIGTERM
+    assert torn_down == ["server stopped"]
+
+
+def test_repeated_sigterm_cannot_abort_teardown() -> None:
+    """The runner escalates. A second SIGTERM must not interrupt cleanup in progress."""
+    torn_down: list[str] = []
+
+    with pytest.raises(TerminatedBySignal):
+        with _raise_on_termination():
+            try:
+                os.kill(os.getpid(), signal.SIGTERM)
+                time.sleep(1)
+            finally:
+                # Simulates the escalation landing mid-teardown. It must be ignored,
+                # not turned into a second exception that abandons the cleanup.
+                os.kill(os.getpid(), signal.SIGTERM)
+                time.sleep(0.2)
+                torn_down.append("server stopped")
+
+    assert torn_down == ["server stopped"]
+
+
+def test_signal_handlers_are_restored_on_exit() -> None:
+    original = signal.getsignal(signal.SIGTERM)
+    with _raise_on_termination():
+        assert signal.getsignal(signal.SIGTERM) is not original
+    assert signal.getsignal(signal.SIGTERM) is original
+
+
 def test_print_server_log_reads_only_tail(tmp_path, capsys) -> None:
     log_path = tmp_path / "server.log"
     log_path.write_bytes(b"excluded-prefix\n" + b"x" * 60000 + b"\nincluded-tail\n")