diff --git a/extensions/python/startup_migration/AGENTS.md b/extensions/python/startup_migration/AGENTS.md index a4c9e202a..fc75b8a8b 100644 --- a/extensions/python/startup_migration/AGENTS.md +++ b/extensions/python/startup_migration/AGENTS.md @@ -13,6 +13,7 @@ - Migrations must be safe to run repeatedly. - Preserve user data and create backups or reversible paths when changing durable state. - Keep long-running work bounded and observable. +- `_10_self_update_manager.py` may replace `/exe/self_update_manager.py` from the repository copy when the installed runtime updater is stale; it must validate required safety markers and keep a backup before replacement. ## Work Guidance diff --git a/extensions/python/startup_migration/_10_self_update_manager.py b/extensions/python/startup_migration/_10_self_update_manager.py new file mode 100644 index 000000000..8a188d6d0 --- /dev/null +++ b/extensions/python/startup_migration/_10_self_update_manager.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +import os +import shutil +import stat +from pathlib import Path +from typing import Any + +from helpers.extension import Extension +from helpers.print_style import PrintStyle + + +SELF_UPDATE_MANAGER_PATH = Path( + os.environ.get("A0_SELF_UPDATE_MANAGER_PATH", "/exe/self_update_manager.py") +) +SELF_UPDATE_MANAGER_SOURCE_PATH = Path( + os.environ.get( + "A0_SELF_UPDATE_MANAGER_SOURCE_PATH", + "/a0/docker/run/fs/exe/self_update_manager.py", + ) +) +BACKUP_SUFFIX = ".startup-migration-backup" +REQUIRED_RUNTIME_MARKERS = ( + "def should_include_usr_backup_entry(", + "Skipping non-regular usr backup entry", + "def clean_transient_desktop_agent_state(", + "clean_transient_desktop_agent_state(REPO_DIR, logger)", +) + + +class SelfUpdateManagerRuntimeSync(Extension): + def execute(self, **kwargs): + result = ensure_self_update_manager_runtime_current() + if result.get("updated"): + PrintStyle.info("Self-update manager runtime synchronized:", result["target"]) + elif result.get("warning"): + PrintStyle.warning("Self-update manager runtime sync skipped:", result["warning"]) + + +def ensure_self_update_manager_runtime_current( + *, + target_path: Path | str | None = None, + source_path: Path | str | None = None, +) -> dict[str, Any]: + target = Path(target_path) if target_path is not None else SELF_UPDATE_MANAGER_PATH + source = Path(source_path) if source_path is not None else SELF_UPDATE_MANAGER_SOURCE_PATH + + target_text, target_warning = _read_regular_text(target, role="runtime self-update manager") + if target_text is None: + return {"ok": True, "updated": False, "reason": target_warning} + + source_text, source_warning = _read_regular_text(source, role="source self-update manager") + if source_text is None: + return {"ok": False, "updated": False, "warning": source_warning} + + missing_source_markers = _missing_required_markers(source_text) + if missing_source_markers: + return { + "ok": False, + "updated": False, + "warning": ( + "source self-update manager is missing required safety markers: " + + ", ".join(missing_source_markers) + ), + } + + if not _missing_required_markers(target_text): + return {"ok": True, "updated": False, "reason": "already-current"} + + try: + backup = _replace_runtime_manager(target, source_text) + except OSError as exc: + return { + "ok": False, + "updated": False, + "warning": f"could not update {target}: {exc}", + } + + return { + "ok": True, + "updated": True, + "target": str(target), + "backup": str(backup), + } + + +def _missing_required_markers(text: str) -> list[str]: + return [marker for marker in REQUIRED_RUNTIME_MARKERS if marker not in text] + + +def _read_regular_text(path: Path, *, role: str) -> tuple[str | None, str]: + try: + path_stat = path.lstat() + except FileNotFoundError: + return None, f"{role} not found: {path}" + except OSError as exc: + return None, f"{role} could not be inspected: {path}: {exc}" + + if not stat.S_ISREG(path_stat.st_mode): + return None, f"{role} is not a regular file: {path}" + + try: + return path.read_text(encoding="utf-8"), "" + except OSError as exc: + return None, f"{role} could not be read: {path}: {exc}" + + +def _replace_runtime_manager(target: Path, source_text: str) -> Path: + target_stat = target.stat() + backup = _ensure_backup(target) + temp_path = target.with_name(f".{target.name}.{os.getpid()}.tmp") + try: + temp_path.write_text(source_text, encoding="utf-8") + os.chmod(temp_path, stat.S_IMODE(target_stat.st_mode)) + os.replace(temp_path, target) + finally: + temp_path.unlink(missing_ok=True) + return backup + + +def _ensure_backup(target: Path) -> Path: + backup = target.with_name(f"{target.name}{BACKUP_SUFFIX}") + if not backup.exists(): + shutil.copy2(target, backup) + return backup diff --git a/tests/test_self_update_runtime_sync.py b/tests/test_self_update_runtime_sync.py new file mode 100644 index 000000000..63cea78e8 --- /dev/null +++ b/tests/test_self_update_runtime_sync.py @@ -0,0 +1,141 @@ +from __future__ import annotations + +import stat +import sys +from pathlib import Path + + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from extensions.python.startup_migration import _10_self_update_manager as migration + + +SAFE_SOURCE = """#!/usr/bin/env python3 +from pathlib import Path + + +REPO_DIR = Path("/a0") + + +def should_include_usr_backup_entry(source_file, logger): + logger.log("Skipping non-regular usr backup entry") + return False + + +def clean_transient_desktop_agent_state(repo_dir, logger): + return None + + +def docker_run_ui(): + clean_transient_desktop_agent_state(REPO_DIR, logger) +""" + + +def test_self_update_runtime_sync_replaces_stale_manager(tmp_path): + source = tmp_path / "source_self_update_manager.py" + target = tmp_path / "self_update_manager.py" + stale = "# old updater without non-regular usr backup guards\n" + source.write_text(SAFE_SOURCE, encoding="utf-8") + target.write_text(stale, encoding="utf-8") + target.chmod(0o600) + + result = migration.ensure_self_update_manager_runtime_current( + target_path=target, + source_path=source, + ) + + assert result["ok"] is True + assert result["updated"] is True + assert target.read_text(encoding="utf-8") == SAFE_SOURCE + assert (target.stat().st_mode & 0o777) == 0o600 + backup = target.with_name(f"{target.name}{migration.BACKUP_SUFFIX}") + assert backup.read_text(encoding="utf-8") == stale + + +def test_self_update_runtime_sync_accepts_repository_manager_source(tmp_path): + source = PROJECT_ROOT / "docker" / "run" / "fs" / "exe" / "self_update_manager.py" + target = tmp_path / "self_update_manager.py" + stale = "# old updater without non-regular usr backup guards\n" + target.write_text(stale, encoding="utf-8") + + result = migration.ensure_self_update_manager_runtime_current( + target_path=target, + source_path=source, + ) + + assert result["ok"] is True + assert result["updated"] is True + assert target.read_text(encoding="utf-8") == source.read_text(encoding="utf-8") + + +def test_self_update_runtime_sync_skips_current_manager(tmp_path): + source = tmp_path / "source_self_update_manager.py" + target = tmp_path / "self_update_manager.py" + source.write_text(SAFE_SOURCE, encoding="utf-8") + target.write_text(SAFE_SOURCE, encoding="utf-8") + + result = migration.ensure_self_update_manager_runtime_current( + target_path=target, + source_path=source, + ) + + assert result == {"ok": True, "updated": False, "reason": "already-current"} + backup = target.with_name(f"{target.name}{migration.BACKUP_SUFFIX}") + assert not backup.exists() + + +def test_self_update_runtime_sync_refuses_source_without_required_markers(tmp_path): + source = tmp_path / "source_self_update_manager.py" + target = tmp_path / "self_update_manager.py" + stale = "# old updater without non-regular usr backup guards\n" + source.write_text("def create_usr_backup():\n pass\n", encoding="utf-8") + target.write_text(stale, encoding="utf-8") + + result = migration.ensure_self_update_manager_runtime_current( + target_path=target, + source_path=source, + ) + + assert result["ok"] is False + assert result["updated"] is False + assert "missing required safety markers" in result["warning"] + assert target.read_text(encoding="utf-8") == stale + backup = target.with_name(f"{target.name}{migration.BACKUP_SUFFIX}") + assert not backup.exists() + + +def test_self_update_runtime_sync_missing_target_is_quiet(tmp_path): + source = tmp_path / "source_self_update_manager.py" + target = tmp_path / "missing_self_update_manager.py" + source.write_text(SAFE_SOURCE, encoding="utf-8") + + result = migration.ensure_self_update_manager_runtime_current( + target_path=target, + source_path=source, + ) + + assert result["ok"] is True + assert result["updated"] is False + assert "not found" in result["reason"] + + +def test_self_update_runtime_sync_skips_non_regular_target(tmp_path): + source = tmp_path / "source_self_update_manager.py" + target = tmp_path / "self_update_manager.py" + link_target = tmp_path / "linked_self_update_manager.py" + source.write_text(SAFE_SOURCE, encoding="utf-8") + link_target.write_text("# linked updater\n", encoding="utf-8") + target.symlink_to(link_target) + + result = migration.ensure_self_update_manager_runtime_current( + target_path=target, + source_path=source, + ) + + assert result["ok"] is True + assert result["updated"] is False + assert "not a regular file" in result["reason"] + assert stat.S_ISLNK(target.lstat().st_mode) + assert link_target.read_text(encoding="utf-8") == "# linked updater\n"