gpu monitor，better monitor ui

deployment/manual_dispatch_release.py

@@ -1063,7 +1063,17 @@ def _finalize_remote_staged_dir(
                 + sh_quote(stage_dir_s)
                 + " "
                 + sh_quote(dst_dir_s)
-                + " && rm -rf \"$backup\""
+                + " && "
+                + "if [ -n \"${backup:-}\" ] && [ -e \"$backup\" -o -L \"$backup\" ]; then "
+                + "rm -rf \"$backup\" || "
+                + "{ "
+                + "echo "
+                + sh_quote(
+                    "[manual_dispatch_release] warning: failed to remove old staged backup; keep it for later cleanup: "
+                )
+                + " \"$backup\" 1>&2; "
+                + "}; "
+                + "fi"
             )
         ),
     )

deployment/tests/test_gen_bare_deploy_bash.py

@@ -64,6 +64,7 @@ def _build_checks(selected_test_id: Optional[str]) -> List[Tuple[str, Callable[[
         ("bootstrap_start_reuses_already_present_selection", test_bootstrap_start_reuses_already_present_selection),
         ("bare_start_fails_when_child_exits_within_startup_window", test_bare_start_fails_when_child_exits_within_startup_window),
         ("pid_ready_check_requires_full_stable_window_after_first_child_observation", test_pid_ready_check_requires_full_stable_window_after_first_child_observation),
+        ("pid_ready_check_ignores_nested_selection_supervisor_children", test_pid_ready_check_ignores_nested_selection_supervisor_children),
         ("atomic_group_start_does_not_auto_stop_on_failure", test_atomic_group_start_does_not_auto_stop_on_failure),
         ("atomic_group_preserves_nested_heredoc_terminator", test_atomic_group_preserves_nested_heredoc_terminator),
         ("atomic_group_stop_script_is_shell_valid", test_atomic_group_stop_script_is_shell_valid),
@@ -812,6 +813,142 @@ def _shutdown(_signum, _frame):
         print("PASS: test_pid_ready_check_requires_full_stable_window_after_first_child_observation")
 
 
+def test_pid_ready_check_ignores_nested_selection_supervisor_children() -> None:
+    proc_lifecycle = _load_python_module(
+        module_name="test_proc_lifecycle_codegen_nested_supervisor_runtime",
+        path=DEPLOYMENT_DIR / "utils" / "proc_lifecycle_codegen.py",
+    )
+    helpers = proc_lifecycle.render_bash_proc_lifecycle_funcs_pid_tree(
+        timeouts=proc_lifecycle.StopTimeouts(term_seconds=60, kill_seconds=10, supersede_seconds=30)
+    )
+    with tempfile.TemporaryDirectory(prefix="test_proc_lifecycle_nested_supervisor_") as td:
+        tmpdir = Path(td)
+        shell_script = tmpdir / "probe.sh"
+        root_script = tmpdir / "root_supervisor.py"
+        child_script = tmpdir / "real_child.py"
+        nested_supervisor_script = tmpdir / "selection_supervisor.py"
+
+        child_script.write_text(
+            textwrap.dedent(
+                """
+                #!/usr/bin/env python3
+                import signal
+                import time
+
+                def _shutdown(_signum, _frame):
+                    raise SystemExit(0)
+
+                signal.signal(signal.SIGTERM, _shutdown)
+                signal.signal(signal.SIGINT, _shutdown)
+
+                while True:
+                    time.sleep(0.2)
+                """
+            ).strip()
+            + "\n",
+            encoding="utf-8",
+        )
+        nested_supervisor_script.write_text(
+            textwrap.dedent(
+                """
+                #!/usr/bin/env python3
+                import signal
+                import time
+
+                def _shutdown(_signum, _frame):
+                    raise SystemExit(0)
+
+                signal.signal(signal.SIGTERM, _shutdown)
+                signal.signal(signal.SIGINT, _shutdown)
+
+                while True:
+                    time.sleep(0.2)
+                """
+            ).strip()
+            + "\n",
+            encoding="utf-8",
+        )
+        root_script.write_text(
+            textwrap.dedent(
+                f"""
+                #!/usr/bin/env python3
+                import signal
+                import subprocess
+                import sys
+                import time
+                from pathlib import Path
+
+                procs = []
+
+                def _shutdown(_signum, _frame):
+                    for proc in procs:
+                        if proc.poll() is None:
+                            proc.terminate()
+                    deadline = time.time() + 5
+                    for proc in procs:
+                        if proc.poll() is None:
+                            try:
+                                proc.wait(timeout=max(0.0, deadline - time.time()))
+                            except subprocess.TimeoutExpired:
+                                proc.kill()
+                    raise SystemExit(0)
+
+                signal.signal(signal.SIGTERM, _shutdown)
+                signal.signal(signal.SIGINT, _shutdown)
+
+                procs.append(subprocess.Popen([sys.executable, str(Path({str(child_script)!r}))]))
+                procs.append(subprocess.Popen([sys.executable, str(Path({str(nested_supervisor_script)!r}))]))
+                while True:
+                    for proc in procs:
+                        if proc.poll() is not None:
+                            raise SystemExit(proc.returncode or 0)
+                    time.sleep(0.2)
+                """
+            ).strip()
+            + "\n",
+            encoding="utf-8",
+        )
+
+        shell_script.write_text(
+            textwrap.dedent(
+                f"""\
+                #!/usr/bin/env bash
+                set -euo pipefail
+                {helpers}
+                python3 {shlex.quote(str(root_script))} &
+                root_pid="$!"
+                startup_deadline_seconds=6
+                if ! wait_service_probably_ready_pid_tree "svc_plain" "$root_pid" 4 "$startup_deadline_seconds" "[test]"; then
+                  wait_rc="$?"
+                  kill "$root_pid" >/dev/null 2>&1 || true
+                  wait "$root_pid" >/dev/null 2>&1 || true
+                  exit "$wait_rc"
+                fi
+                kill "$root_pid" >/dev/null 2>&1 || true
+                wait "$root_pid" >/dev/null 2>&1 || true
+                exit 0
+                """
+            ),
+            encoding="utf-8",
+        )
+        shell_script.chmod(0o755)
+
+        result = subprocess.run(
+            ["bash", str(shell_script)],
+            check=False,
+            capture_output=True,
+            text=True,
+            cwd=str(DEPLOYMENT_DIR.parent),
+            timeout=20,
+        )
+        assert result.returncode == 0, (
+            f"expected startup gate success rc={result.returncode} stdout={result.stdout!r} stderr={result.stderr!r}"
+        )
+        assert "multiple direct child pids" not in result.stdout, result.stdout
+        assert "probable-ready: ok" in result.stdout, result.stdout
+        print("PASS: test_pid_ready_check_ignores_nested_selection_supervisor_children")
+
+
 def test_atomic_group_preserves_nested_heredoc_terminator() -> None:
     with tempfile.TemporaryDirectory(prefix="test_gen_bare_deploy_bash_atomic_heredoc_") as td:
         tmpdir = Path(td)

deployment/tests/test_manual_dispatch_release_test_rsc_contract.py

@@ -30,6 +30,33 @@ def _load_module():
 
 
 class TestManualDispatchReleaseTestRscContract(unittest.TestCase):
+    def test_finalize_remote_staged_dir_keeps_backup_when_cleanup_fails(self) -> None:
+        captured: list[tuple[str | None, str]] = []
+
+        def _fake_check_call_bash_with_optional_password(*, password: str | None, cmd: str) -> None:
+            captured.append((password, cmd))
+
+        with mock.patch.object(
+            _DISPATCH,
+            "_check_call_bash_with_optional_password",
+            side_effect=_fake_check_call_bash_with_optional_password,
+        ):
+            _DISPATCH._finalize_remote_staged_dir(
+                stage_dir_s="/remote/.fluxon_release.stage.abc123",
+                dst_dir_s="/remote/fluxon_release",
+                ssh_user="root",
+                ip="203.0.113.7",
+                ssh_port=30245,
+                ssh_password=None,
+            )
+
+        self.assertEqual(len(captured), 1)
+        password, cmd = captured[0]
+        self.assertIsNone(password)
+        self.assertIn('mv \'"\'"\'/remote/.fluxon_release.stage.abc123\'"\'"\' \'"\'"\'/remote/fluxon_release\'"\'"\'', cmd)
+        self.assertIn('rm -rf "$backup" || {', cmd)
+        self.assertIn("[manual_dispatch_release] warning: failed to remove old staged backup; keep it for later cleanup:", cmd)
+
     def test_deploy_and_profiles_dispatches_test_rsc_tree(self) -> None:
         with tempfile.TemporaryDirectory() as td:
             release_dir = Path(td)