fix(updater): harden upgrade restart and rollback

flocks/cli/service_manager.py

@@ -26,7 +26,7 @@
 import httpx
 
 from flocks.browser.admin import stop_all_daemons as stop_all_browser_daemons
-from flocks.cli.service_config import ServiceConfig, loopback_host
+from flocks.cli.service_config import ServiceConfig, loopback_host, service_config_payload
 from flocks.cli.service_control import (
     read_logs,
     read_supervisor_status,
@@ -760,6 +760,8 @@ def _resolve_upgrade_runtime(console, *, frontend_port: int, attempt_recover: bo
     error = result.get("error")
     if action == "recovered":
         _console_print(console, "[flocks] 已恢复未完成升级，正式 WebUI 将继续接管端口。")
+    elif action == "failure_preserved":
+        _console_print(console, "[flocks] 已清理升级临时页，并保留回滚失败状态。")
     elif action != "noop":
         _console_print(console, "[flocks] 已清理升级临时页残留。")
 
@@ -1130,6 +1132,11 @@ def _backend_command_and_env(root: Path, config: ServiceConfig) -> tuple[list[st
     env = os.environ.copy()
     env["_FLOCKS_WEBUI_HOST"] = config.frontend_host
     env["_FLOCKS_WEBUI_PORT"] = str(config.frontend_port)
+    env["_FLOCKS_SERVICE_CONFIG"] = json.dumps(
+        service_config_payload(config),
+        ensure_ascii=True,
+        sort_keys=True,
+    )
     env["PYTHONUNBUFFERED"] = "1"
     env.setdefault("FLOCKS_CONSOLE_BASE_URL", DEFAULT_FLOCKS_CONSOLE_BASE_URL)
     return command, env
@@ -1371,6 +1378,8 @@ def _wait_for_supervisor_ready(
     timeout: float = SUPERVISOR_START_TIMEOUT_SECONDS,
 ) -> dict[str, Any]:
     """Wait for the supervisor control API and managed services to become ready."""
+    from flocks.cli.service_process import tcp_port_accepts_connections
+
     deadline = time.monotonic() + timeout
     last_payload: dict[str, Any] | None = None
     while time.monotonic() < deadline:
@@ -1381,15 +1390,20 @@ def _wait_for_supervisor_ready(
             last_payload = status.raw
             backend_state = status.backend.state
             webui_state = status.webui.state
-            if backend_state == "healthy" and webui_state in {"healthy", "static"}:
+            if (
+                backend_state == "healthy"
+                and webui_state in {"healthy", "static"}
+                and status.backend.port is not None
+                and tcp_port_accepts_connections(status.backend.host, status.backend.port)
+            ):
                 return status.raw
             if backend_state == "degraded" or webui_state == "degraded":
                 return status.raw
         except Exception:
             pass
         time.sleep(0.5)
     if last_payload is not None:
-        return last_payload
+        raise ServiceError("Flocks daemon 启动超时：后端 TCP 端口未就绪，请检查日志。")
     raise ServiceError("Flocks daemon 启动超时，请检查日志。")
 
 
@@ -1426,8 +1440,6 @@ def _start_supervisor_process(config: ServiceConfig, paths: RuntimePaths, consol
     """Spawn the detached service supervisor daemon."""
     root = ensure_install_layout()
     log_path = supervisor_log_path(paths)
-    if not supervisor_uses_tcp_control():
-        supervisor_socket_path(paths).unlink(missing_ok=True)
     command = resolve_flocks_cli_command(root) + [
         "service-daemon",
         "--server-host",
@@ -1449,6 +1461,11 @@ def _start_supervisor_process(config: ServiceConfig, paths: RuntimePaths, consol
         command.append("--skip-webui-build")
     env = os.environ.copy()
     env["PYTHONUNBUFFERED"] = "1"
+    env["_FLOCKS_SERVICE_CONFIG"] = json.dumps(
+        service_config_payload(config),
+        ensure_ascii=True,
+        sort_keys=True,
+    )
     return _spawn_process(command, cwd=root, log_path=log_path, env=env)
 
 
@@ -1513,6 +1530,7 @@ def _stop_all_unlocked(console, *, paths: RuntimePaths) -> None:
     cleanup_config = ServiceConfig()
     legacy_config = _legacy_runtime_config(paths, cleanup_config)
     stop_status = None
+    daemon_pid = None
     if not supervisor_is_running(paths):
         console.print("[flocks] Flocks daemon 未运行。")
         cleanup_legacy_runtime_processes(paths, console)
@@ -1521,6 +1539,7 @@ def _stop_all_unlocked(console, *, paths: RuntimePaths) -> None:
         return
     try:
         stop_status = read_supervisor_status(paths=paths, timeout=1.0)
+        daemon_pid = stop_status.daemon.pid
         cleanup_config = stop_status.config
         legacy_config = _legacy_runtime_config(paths, cleanup_config)
     except Exception:
@@ -1532,7 +1551,9 @@ def _stop_all_unlocked(console, *, paths: RuntimePaths) -> None:
 
     deadline = time.monotonic() + 20.0
     while time.monotonic() < deadline:
-        if not supervisor_is_running(paths):
+        control_stopped = not supervisor_is_running(paths)
+        daemon_stopped = not pid_is_running(daemon_pid)
+        if control_stopped and daemon_stopped:
             cleanup_legacy_runtime_processes(paths, console)
             cleanup_orphan_service_ports(cleanup_config, console, extra_configs=[legacy_config])
             stop_all_browser_daemons()
@@ -1558,8 +1579,9 @@ def _start_all_without_stop(config: ServiceConfig, console) -> None:
     cleanup_orphan_service_ports(config, console)
     _ensure_webui_dist(ensure_install_layout(), config, console)
     process = _start_supervisor_process(config, paths, console)
-    console.print("[flocks] Flocks daemon 已启动。")
+    console.print("[flocks] Flocks daemon 进程已启动，正在等待服务就绪...")
     payload = _wait_for_supervisor_ready(paths, process=process)
+    console.print("[flocks] Flocks daemon 已启动。")
     _print_status_payload(payload, console, include_daemon_step=False)
     if not _startup_payload_is_ready(payload):
         raise ServiceError(_startup_failure_message(payload))

flocks/cli/service_supervisor.py

@@ -119,6 +119,7 @@ def __init__(
         self._shutdown_requested = threading.Event()
         self._server: ThreadingHTTPServer | None = None
         self._server_thread: threading.Thread | None = None
+        self._control_socket_identity: tuple[int, int] | None = None
         self._backend_paused = False
         self._webui_paused = False
         self.backend = ManagedService(
@@ -190,6 +191,8 @@ def _start_control_server(self) -> None:
             socket_path.unlink(missing_ok=True)
             assert _UnixControlServer is not None
             server = _UnixControlServer(str(socket_path), handler)
+            stat_result = socket_path.stat()
+            self._control_socket_identity = (stat_result.st_dev, stat_result.st_ino)
         self._server = server
         self._server_thread = threading.Thread(target=server.serve_forever, name="flocks-supervisor-control", daemon=True)
         self._server_thread.start()
@@ -201,8 +204,17 @@ def _stop_control_server(self) -> None:
             self._server.server_close()
         if self._server_thread is not None:
             self._server_thread.join(timeout=5.0)
-        if not supervisor_uses_tcp_control():
-            supervisor_socket_path(self.paths).unlink(missing_ok=True)
+        if not supervisor_uses_tcp_control() and self._control_socket_identity is not None:
+            socket_path = supervisor_socket_path(self.paths)
+            try:
+                stat_result = socket_path.stat()
+            except FileNotFoundError:
+                pass
+            else:
+                current_identity = (stat_result.st_dev, stat_result.st_ino)
+                if current_identity == self._control_socket_identity:
+                    socket_path.unlink(missing_ok=True)
+            self._control_socket_identity = None
 
     def _handler_class(self):
         daemon = self

flocks/server/app.py

@@ -472,17 +472,20 @@ async def _delayed_trigger_runtime_start() -> None:
     except Exception as e:
         log.warning("workflow.trigger_runtime.start_failed", {"error": str(e)})
 
-    try:
-        from flocks.updater.updater import recover_upgrade_state
+    if os.getenv("_FLOCKS_SERVICE_CONFIG"):
+        log.info("updater.recovery.skipped", {"reason": "managed_service"})
+    else:
+        try:
+            from flocks.updater.updater import recover_upgrade_state
 
-        await _run_startup_phase(
-            log,
-            "updater.recover_upgrade_state",
-            lambda: asyncio.to_thread(recover_upgrade_state),
-        )
-        log.info("updater.recovery.checked")
-    except Exception as e:
-        log.warning("updater.recovery.failed", {"error": str(e)})
+            await _run_startup_phase(
+                log,
+                "updater.recover_upgrade_state",
+                lambda: asyncio.to_thread(recover_upgrade_state),
+            )
+            log.info("updater.recovery.checked")
+        except Exception as e:
+            log.warning("updater.recovery.failed", {"error": str(e)})
 
     blocking_startup_ms = int((time.perf_counter() - startup_started_at) * 1000)
     log.info("server.startup.ready", {

flocks/updater/restart_handoff.py

@@ -11,6 +11,7 @@
 
 import argparse
 import asyncio
+import json
 import shutil
 import subprocess
 import time
@@ -83,6 +84,12 @@ def _stop_supervisor_before_restart(
     if not service_control.supervisor_is_running(paths):
         return True
 
+    daemon_pid = None
+    try:
+        daemon_pid = service_control.read_supervisor_status(paths=paths, timeout=1.0).daemon.pid
+    except Exception:
+        pass
+
     try:
         service_control.request_stop(paths=paths, timeout=timeout_seconds)
     except Exception as exc:
@@ -91,10 +98,12 @@ def _stop_supervisor_before_restart(
 
     deadline = time.monotonic() + timeout_seconds
     while time.monotonic() < deadline:
-        if not service_control.supervisor_is_running(paths):
+        control_stopped = not service_control.supervisor_is_running(paths)
+        daemon_stopped = not service_manager.pid_is_running(daemon_pid)
+        if control_stopped and daemon_stopped:
             return True
         time.sleep(poll_interval_seconds)
-    return not service_control.supervisor_is_running(paths)
+    return not service_control.supervisor_is_running(paths) and not service_manager.pid_is_running(daemon_pid)
 
 
 def _parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
@@ -118,6 +127,7 @@ def _parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
     parser.add_argument("--bundle-sha256")
     parser.add_argument("--cleanup-dir")
     parser.add_argument("--prepare-handover", action="store_true")
+    parser.add_argument("--service-config-json")
     parser.add_argument("restart_argv", nargs=argparse.REMAINDER)
     args = parser.parse_args(argv)
     if args.restart_argv and args.restart_argv[0] == "--":
@@ -177,10 +187,35 @@ def _rollback_failed_upgrade(args: argparse.Namespace, error: str) -> None:
 
 
 def _prepare_upgrade_handover(args: argparse.Namespace) -> bool:
+    from flocks.cli.service_config import ServiceConfig, service_config_from_payload
     from flocks.updater import updater
 
     try:
-        updater._prepare_upgrade_handover(args.version)
+        config_payload = {
+            "backend_host": args.backend_host,
+            "backend_port": args.backend_port,
+            "frontend_host": args.frontend_host,
+            "frontend_port": args.frontend_port,
+        }
+        if args.service_config_json:
+            payload = json.loads(args.service_config_json)
+            if not isinstance(payload, dict):
+                raise ValueError("service config snapshot must be a JSON object")
+            config_payload.update(payload)
+        else:
+            legacy_host = _argv_option(args.restart_argv, "--server-host")
+            legacy_port = _argv_option(args.restart_argv, "--server-port")
+            if legacy_host:
+                config_payload["legacy_backend_host"] = legacy_host
+            if legacy_port:
+                config_payload["legacy_backend_port"] = int(legacy_port)
+        config = service_config_from_payload(
+            config_payload,
+            default=ServiceConfig(),
+            no_browser=True,
+            skip_frontend_build=True,
+        )
+        updater._prepare_upgrade_handover(args.version, config=config)
     except Exception as exc:
         _record_handoff_log(f"prepare_handover_failed error={exc}")
         return False
@@ -210,6 +245,17 @@ def _cli_subcommand(argv: Sequence[str]) -> str | None:
     return None
 
 
+def _argv_option(argv: Sequence[str], option: str) -> str | None:
+    """Return a CLI option value from either ``--name value`` or ``--name=value``."""
+    prefix = f"{option}="
+    for index, value in enumerate(argv):
+        if value.startswith(prefix):
+            return value[len(prefix) :]
+        if value == option and index + 1 < len(argv):
+            return argv[index + 1]
+    return None
+
+
 def _restart_argv_for_current_runtime(args: argparse.Namespace, restart_argv: Sequence[str]) -> list[str]:
     if _cli_subcommand(restart_argv) != "serve":
         return list(restart_argv)
@@ -237,6 +283,7 @@ def _restart_argv_for_current_runtime(args: argparse.Namespace, restart_argv: Se
 def run(argv: Sequence[str] | None = None) -> int:
     args = _parse_args(argv)
     restart_argv = _restart_argv_for_current_runtime(args, args.restart_argv)
+    args.restart_argv = restart_argv
     if not restart_argv:
         _record_handoff_log("missing_restart_argv")
         return 2