Implement guest tool installation and management system

- Added `guest_tools.py` for standalone tool installer functionality.
- Introduced `ToolJobStore` in `tool_store.py` to manage per-VM tool job records.
- Created `ToolService` in `tools.py` for background installation workers using the guest execution channel.
- Enhanced `LifecycleService` in `lifecycle.py` to support tool installation queuing and preparation.
- Updated `guest_exec.py` to handle connection shutdown gracefully.
- Added validation and normalization functions for tool specifications.
- Implemented logging and error handling for tool installation processes.
- Ensured compatibility with Ubuntu 22.04 and ARM64/x86_64 architectures.

tests/test_tool_catalog.py

@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+import hashlib
+import io
+import os
+import shlex
+import tarfile
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+from uvm import guest_tools
+from uvm.errors import ValidationError
+from uvm.tool_catalog import CATALOG, ToolSelection, all_selections, normalize_tools, recipe
+from uvm.tools import guest_command
+
+
+class ToolCatalogTests(unittest.TestCase):
+    def test_requested_catalog_and_every_guest_payload_fit_the_agent_protocol(self):
+        self.assertEqual({tool["id"] for tool in CATALOG},
+                         {"git", "nodejs", "python", "jdk", "wget", "curl", "htop", "php", "conda", "micromamba"})
+        for arch in ("x86_64", "aarch64"):
+            recipes = [recipe(item, arch) for item in all_selections()]
+            command = guest_command({"action": "probe", "tools": recipes, "username": "worker", "architecture": arch})
+            self.assertLess(len(command.encode()), 65536)
+            argv = shlex.split(command)
+            self.assertEqual(argv[:3], ["/usr/bin/python3", "-I", "-c"])
+            compile(argv[3], "guest-tools", "exec")
+            for item in recipes:
+                if "url" in item:
+                    self.assertTrue(item["url"].startswith("https://"))
+                    self.assertNotIn("/latest", item["url"])
+                    self.assertRegex(item["sha256"], r"^[0-9a-f]{64}$")
+        self.assertEqual(len(all_selections()), 12)
+
+    def test_selection_validation_is_allowlisted_strict_and_deduplicated(self):
+        expected = (ToolSelection("git"), ToolSelection("python", "3.12"))
+        self.assertEqual(normalize_tools([{"id": "python", "version": "3.12"}, {"id": "git"}, {"id": "git"}]), expected)
+        for value in (None, "git", [], ["git"], [{"id": "git; id"}], [{"id": "python"}],
+                      [{"id": "python", "version": "3.14"}], [{"id": "git", "version": "latest"}],
+                      [{"id": "git", "command": "id"}], [{"id": ["git"]}], [{"id": "git"}] * 13):
+            with self.subTest(value=value), self.assertRaises(ValidationError):
+                normalize_tools(value)
+        self.assertEqual(normalize_tools([], allow_empty=True), ())
+
+    def test_versioned_python_commands_and_jdk_compiler_are_verified(self):
+        for version in ("3.11", "3.12", "3.13"):
+            item = recipe(ToolSelection("python", version), "aarch64")
+            self.assertEqual(set(item["links"]), {"python" + version})
+            self.assertIn("venv", item["checks"][0][-1])
+            self.assertEqual(item["checks"][1][:3], ["python" + version, "-m", "pip"])
+        jdk = recipe(ToolSelection("jdk"), "aarch64")
+        self.assertEqual(jdk["checks"], [["javac", "-version"], ["java", "-version"]])
+        self.assertIn("arm64", jdk["java_home"])
+
+
+class GuestToolRecipeTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = self.enterContext(tempfile.TemporaryDirectory())
+        self.root = Path(self.temporary)
+
+    def test_download_mismatch_removes_artifact_and_https_redirect_is_checked(self):
+        target = self.root / "download"
+        response = io.BytesIO(b"publisher payload")
+        response.geturl = lambda: "https://publisher.test/tool"
+        with patch.object(guest_tools.urllib.request, "urlopen", return_value=response):
+            with self.assertRaisesRegex(guest_tools.ToolError, "checksum"):
+                guest_tools.download("https://publisher.test/tool", target, "0" * 64)
+        self.assertFalse(target.exists())
+        response = io.BytesIO(b"publisher payload")
+        response.geturl = lambda: "https://publisher.test/tool"
+        with patch.object(guest_tools.urllib.request, "urlopen", return_value=response):
+            guest_tools.download("https://publisher.test/tool", target, hashlib.sha256(b"publisher payload").hexdigest())
+        self.assertEqual(target.read_bytes(), b"publisher payload")
+        with self.assertRaisesRegex(guest_tools.ToolError, "HTTPS"):
+            guest_tools.download("http://publisher.test/tool", target)
+        response = io.BytesIO(b"publisher payload")
+        response.geturl = lambda: "http://publisher.test/tool"
+        with patch.object(guest_tools.urllib.request, "urlopen", return_value=response):
+            with self.assertRaisesRegex(guest_tools.ToolError, "HTTPS"):
+                guest_tools.download("https://publisher.test/tool", target)
+
+    def test_archive_traversal_links_and_devices_cannot_escape_staging(self):
+        cases = [("../outside", tarfile.REGTYPE, ""), ("/outside", tarfile.REGTYPE, ""),
+                 ("link", tarfile.SYMTYPE, "../../outside"), ("link", tarfile.LNKTYPE, "../outside"),
+                 ("device", tarfile.CHRTYPE, "")]
+        for name, kind, link in cases:
+            with self.subTest(name=name, kind=kind):
+                archive = self.root / "archive.tar"
+                with tarfile.open(archive, "w") as stream:
+                    item = tarfile.TarInfo(name)
+                    item.type = kind
+                    item.linkname = link
+                    stream.addfile(item)
+                output = self.root / "output"
+                output.mkdir(exist_ok=True)
+                with self.assertRaises(guest_tools.ToolError):
+                    guest_tools.extract_archive(archive, output)
+                self.assertFalse((self.root / "outside").exists())
+
+    def test_internal_archive_symlinks_are_supported(self):
+        archive = self.root / "archive.tar"
+        with tarfile.open(archive, "w") as stream:
+            item = tarfile.TarInfo("python/bin/python3.12")
+            item.size = 4
+            item.mode = 0o755
+            stream.addfile(item, io.BytesIO(b"test"))
+            link = tarfile.TarInfo("python/bin/python3")
+            link.type = tarfile.SYMTYPE
+            link.linkname = "python3.12"
+            stream.addfile(link)
+        output = self.root / "output"
+        output.mkdir()
+        guest_tools.extract_archive(archive, output)
+        self.assertEqual((output / "python/bin/python3").read_bytes(), b"test")
+
+    def test_probe_uses_guest_account_and_rejects_wrong_runtime_or_missing_companion(self):
+        node = recipe(ToolSelection("nodejs"), "x86_64")
+        account = SimpleNamespace(pw_dir="/home/worker", pw_uid=1001, pw_gid=1002)
+        with (patch.object(guest_tools.pwd, "getpwnam", return_value=account),
+              patch.object(guest_tools.os, "getgrouplist", return_value=[1002, 1003]),
+              patch.object(guest_tools.shutil, "which", return_value="/usr/local/bin/node"),
+              patch.object(guest_tools.subprocess, "run") as run):
+            run.return_value = SimpleNamespace(returncode=0, stdout=b"v22.10.0\n")
+            result = guest_tools.probe(node, "worker")
+            self.assertFalse(result["installed"])
+            self.assertEqual(result["detected_version"], "22.10.0")
+            self.assertEqual(run.call_args.kwargs["user"], 1001)
+            self.assertEqual(run.call_args.kwargs["env"]["HOME"], "/home/worker")
+            run.side_effect = [SimpleNamespace(returncode=0, stdout=b"v24.21.0\n"),
+                               SimpleNamespace(returncode=1, stdout=b"broken npm")]
+            self.assertFalse(guest_tools.probe(node, "worker")["installed"])
+
+    def test_managed_files_preserve_edits_and_can_finish_interrupted_publication(self):
+        installer = guest_tools.Installer("root")
+        path = self.root / "profile"
+        # Isolated directory fixtures stand in for the root-owned guest hierarchy.
+        with (patch.object(guest_tools, "ROOT", self.root),
+              patch.object(guest_tools, "safe_directory"),
+              patch.object(guest_tools, "regular_file")):
+            installer.write_managed(path, b"initial")
+            installer.write_managed(path, b"updated")
+            self.assertEqual(path.read_bytes(), b"updated")
+            path.write_bytes(b"guest edit")
+            with self.assertRaisesRegex(guest_tools.ToolError, "customized"):
+                installer.write_managed(path, b"next")
+            self.assertEqual(path.read_bytes(), b"guest edit")
+            path.write_bytes(b"updated")
+            installer.remember(path, hashlib.sha256(b"next").hexdigest())
+            installer.write_managed(path, b"next")
+            self.assertEqual(path.read_bytes(), b"next")
+
+    def test_log_capture_retains_a_bounded_tail(self):
+        installer = guest_tools.Installer("root")
+        installer.append(b"a" * 20000)
+        installer.append(b"last failure")
+        self.assertEqual(len(installer.log), guest_tools.LOG_LIMIT)
+        self.assertTrue(installer.truncated)
+        self.assertTrue(installer.log.endswith(b"last failure"))

tests/test_tool_jobs.py

@@ -0,0 +1,151 @@
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import threading
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import replace
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+from tests.test_server import FakeLifecycle
+from uvm.config import Settings
+from uvm.errors import ConflictError, GuestExecutionError, NotFoundError, StateError
+from uvm.state import StateStore
+from uvm.tool_catalog import ToolSelection
+from uvm.tool_store import ToolJobStore
+from uvm.tools import ToolService
+
+
+class ToolJobTests(unittest.TestCase):
+    def setUp(self):
+        temporary = self.enterContext(tempfile.TemporaryDirectory())
+        self.settings = replace(Settings(), base=Path(temporary), tool_workers=1)
+        self.state = StateStore(self.settings)
+        self.vm = replace(FakeLifecycle().vm, runtime_owned=True, pid=123, process_start_time="1", boot_id="host-boot")
+        self.settings.vm_dir(self.vm.id).mkdir(parents=True, mode=0o700)
+        with self.state.transaction() as state:
+            state.vms[self.vm.id] = self.vm
+        self.store = ToolJobStore(self.settings, self.state)
+
+    def enqueue(self, selections=None, **options):
+        with self.state.operation_lock():
+            return self.store.enqueue_locked(self.vm.id, selections or [ToolSelection("git")], **options)
+
+    def test_duplicate_concurrent_submissions_have_one_durable_job(self):
+        with ThreadPoolExecutor(max_workers=4) as executor:
+            jobs = list(executor.map(lambda _: self.enqueue(request_id="request-1"), range(4)))
+        self.assertEqual(len({job["id"] for job in jobs}), 1)
+        self.assertEqual(len(self.store.read(self.vm.id)["jobs"]), 1)
+        path = self.settings.vm_dir(self.vm.id) / "tools.json"
+        self.assertEqual(path.stat().st_mode & 0o777, 0o600)
+        self.assertNotIn("guest-secret", path.read_text())
+        with self.assertRaises(ConflictError):
+            self.enqueue([ToolSelection("htop")])
+
+    def test_history_is_vm_scoped_and_deletion_is_never_undone(self):
+        job = self.enqueue()
+        with self.assertRaises(NotFoundError):
+            self.store.get_job(self.vm.name, job["id"])
+        with self.state.transaction() as state:
+            del state.vms[self.vm.id]
+        with self.assertRaises(NotFoundError):
+            self.store.update_job(self.vm.id, job["id"], lambda current, data: current.update(status="succeeded"))
+        self.assertNotIn(self.vm.id, self.state.load().vms)
+
+    def test_recovery_preserves_queued_work_and_marks_inflight_outcomes_unknown(self):
+        job = self.enqueue()
+        self.store.recover()
+        self.assertEqual(self.store.get_job(self.vm.id, job["id"])["status"], "queued")
+        self.store.claim()
+        self.store.update_job(self.vm.id, job["id"], lambda current, data: current["items"][0].update(status="installing"))
+        self.store.recover()
+        recovered = self.store.get_job(self.vm.id, job["id"])
+        self.assertEqual(recovered["status"], "interrupted")
+        self.assertTrue(recovered["outcome_unknown"])
+        self.assertEqual(recovered["items"][0]["status"], "interrupted")
+        self.assertIsNone(self.store.claim())
+
+    def test_symlinks_and_foreign_history_are_rejected(self):
+        outside = self.settings.base / "outside"
+        outside.write_text("private data")
+        path = self.settings.vm_dir(self.vm.id) / "tools.json"
+        path.symlink_to(outside)
+        with self.assertRaises((StateError, OSError)):
+            self.store.read(self.vm.id)
+        self.assertEqual(outside.read_text(), "private data")
+        path.unlink()
+        self.enqueue()
+        data = json.loads(path.read_text())
+        data["jobs"][0]["vm_id"] = "vm-other"
+        path.write_text(json.dumps(data))
+        with self.assertRaises(StateError):
+            self.store.read(self.vm.id)
+
+    def test_only_one_worker_can_own_this_registry_and_shutdown_releases_it(self):
+        first = ToolService(self.settings, self.state, Mock())
+        second = ToolService(self.settings, self.state, Mock())
+        self.addCleanup(first.shutdown)
+        self.addCleanup(second.shutdown)
+        first.start()
+        with self.assertRaises(ConflictError):
+            second.start()
+        first.shutdown()
+        second.start()
+
+    def make_service(self):
+        lifecycle = Mock()
+        lifecycle.prepare_tool_install.return_value = self.vm
+        service = ToolService(self.settings, self.state, lifecycle)
+        return service
+
+    @staticmethod
+    def observation(selection, installed=False):
+        return {**selection.to_dict(), "key": selection.key, "installed": installed,
+                "detected_version": "2.0.0" if installed else None, "usage": "tool --version", "error": None}
+
+    def test_worker_reports_partial_failure_and_retains_successful_tools(self):
+        selections = [ToolSelection("git"), ToolSelection("htop")]
+        job = self.enqueue(selections)
+        claimed = self.store.claim()
+        service = self.make_service()
+        results = [
+            {"ok": True, "busy": False, "boot_id": "guest-boot", "tools": [self.observation(item) for item in selections]},
+            {"ok": True, "status": "installed", "tool": self.observation(selections[0], True), "output": "git installed"},
+            {"ok": False, "error": "mirror unavailable", "output": "apt download failed"},
+        ]
+        with patch.object(service, "_execute", side_effect=results) as execute:
+            service._run_job(claimed)
+        result = self.store.get_job(self.vm.id, job["id"])
+        self.assertEqual(result["status"], "failed")
+        self.assertEqual([item["status"] for item in result["items"]], ["installed", "failed"])
+        self.assertEqual(result["items"][1]["output"], "apt download failed")
+        self.assertEqual(self.state.get(self.vm.id).status, "running")
+        self.assertEqual(execute.call_args.args[1]["boot_id"], "guest-boot")
+
+    def test_lost_result_is_never_replayed_and_remaining_tools_are_skipped(self):
+        selections = [ToolSelection("git"), ToolSelection("htop")]
+        job = self.enqueue(selections)
+        service = self.make_service()
+        with patch.object(service, "_execute", side_effect=[
+            {"ok": True, "busy": False, "boot_id": "boot", "tools": [self.observation(item) for item in selections]},
+            GuestExecutionError("connection lost", outcome_unknown=True),
+        ]) as execute:
+            service._run_job(self.store.claim())
+        result = self.store.get_job(self.vm.id, job["id"])
+        self.assertEqual(execute.call_count, 2)
+        self.assertEqual(result["status"], "interrupted")
+        self.assertTrue(result["outcome_unknown"])
+        self.assertEqual([item["status"] for item in result["items"]], ["interrupted", "skipped"])
+
+    def test_prior_guest_installation_blocks_a_retry_before_submission(self):
+        job = self.enqueue()
+        service = self.make_service()
+        with patch.object(service, "_execute", return_value={"ok": True, "busy": True}) as execute:
+            service._run_job(self.store.claim())
+        self.assertEqual(execute.call_count, 1)
+        result = self.store.get_job(self.vm.id, job["id"])
+        self.assertEqual(result["status"], "interrupted")
+        self.assertFalse(result["outcome_unknown"])

tests/test_tool_lifecycle.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from unittest.mock import Mock
+
+from tests.test_lifecycle import LifecycleTestCase
+from uvm.errors import ConflictError, NotFoundError, UvmError, ValidationError
+from uvm.tool_catalog import ToolSelection
+
+
+class ToolLifecycleTests(LifecycleTestCase):
+    def test_create_persists_tool_intent_before_boot(self):
+        def configure(config):
+            vm_id = next(iter(self.store.load().vms))
+            jobs = self.service._tool_store._read(vm_id)["jobs"]
+            self.assertEqual(jobs[0]["selections"], [{"id": "python", "version": "3.12"}])
+            self.assertEqual(jobs[0]["status"], "queued")
+        original = self.service._client_factory
+        def client(socket, timeout):
+            result = original(socket, timeout)
+            result.configure_and_start = configure
+            return result
+        self.service._client_factory = client
+        vm = self.create_vm(tools=(ToolSelection("python", "3.12"),))
+        self.assertEqual(vm.status, "running")
+        self.assertEqual(self.service._tool_store.read(vm.id)["jobs"][0]["source"], "create")
+
+    def test_invalid_tools_fail_before_reserving_a_vm_and_failed_boot_removes_intent(self):
+        with self.assertRaises(ValidationError):
+            self.create_vm(tools=({"id": "invalid"},))
+        self.assertEqual(self.store.load().vms, {})
+        self.host.failures["client.configure", None] = UvmError("failed boot")
+        with self.assertRaises(UvmError):
+            self.create_vm(tools=(ToolSelection("git"),))
+        self.assertEqual(list(self.settings.vms_dir.iterdir()), [])
+
+    def test_stopped_installation_starts_only_when_requested_and_aliases_are_rejected(self):
+        vm = self.create_vm(name="worker")
+        self.service.stop(vm.id)
+        for alias in (vm.name, vm.guest_ip):
+            with self.assertRaises(NotFoundError):
+                self.service.queue_tools(alias, [ToolSelection("git")], start_if_stopped=True)
+        with self.assertRaises(ConflictError):
+            self.service.queue_tools(vm.id, [ToolSelection("git")])
+        job = self.service.queue_tools(vm.id, [ToolSelection("git")], start_if_stopped=True)
+        self.assertEqual(self.store.get(vm.id).status, "stopped")
+        ready = self.service.prepare_tool_install(vm.id, start_if_stopped=job["start_if_stopped"])
+        self.assertEqual(ready.status, "running")
+        self.assertEqual(ready.guest_ip, vm.guest_ip)
+
+    def test_a_tool_step_cannot_reconnect_to_a_restarted_vm(self):
+        vm = self.create_vm()
+        identity = vm.pid, vm.process_start_time, vm.boot_id
+        self.service.restart(vm.id)
+        self.service._guest_connector = Mock()
+        with self.assertRaisesRegex(ConflictError, "restarted"):
+            self.service.execute(vm.id, "id", expected_identity=identity)
+        self.service._guest_connector.assert_not_called()

uvm/app.py

@@ -14,6 +14,7 @@ from .lifecycle import LifecycleService
 from .network import NetworkManager
 from .state import StateStore
 from .system import CommandRunner
+from .tools import ToolService
 
 
 @dataclass(slots=True)
@@ -24,6 +25,7 @@ class Application:
     state_store: StateStore
     installer: Installer
     lifecycle: LifecycleService
+    tools: ToolService | None = None
 
 
 def build_application(emit: Callable[[str], None] | None = print) -> Application:
@@ -41,4 +43,5 @@ def build_application(emit: Callable[[str], None] | None = print) -> Application
         state_store=state_store,
         installer=installer,
         lifecycle=lifecycle,
+        tools=ToolService(settings, state_store, lifecycle),
     )

uvm/config.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import os
+import math
 from dataclasses import dataclass, field
 from ipaddress import IPv4Address, IPv4Network
 from pathlib import Path
@@ -46,6 +47,9 @@ class Settings:
     api_timeout_s: float = 2.0
     api_socket_timeout_s: float = 5.0
     terminate_timeout_s: float = 5.0
+    tool_workers: int = 2
+    tool_ready_timeout_s: float = 120.0
+    tool_install_timeout_s: float = 1200.0
 
     def __post_init__(self) -> None:
         architecture = self.architecture.lower()
@@ -54,6 +58,13 @@ class Settings:
             raise ConfigurationError(f"unsupported host architecture: {self.architecture}")
         object.__setattr__(self, "architecture", architecture)
 
+        if type(self.tool_workers) is not int or not 1 <= self.tool_workers <= 8:
+            raise ConfigurationError("tool_workers must be an integer between 1 and 8")
+        for name in ("tool_ready_timeout_s", "tool_install_timeout_s"):
+            value = getattr(self, name)
+            if type(value) not in (int, float) or not math.isfinite(value) or not 0 < value <= 3600:
+                raise ConfigurationError(f"{name} must be finite, greater than zero, and at most 3600 seconds")
+
         # Resolve defaults at construction time, while preserving individual URL overrides.
         asset_base = f"{DEFAULT_GUEST_ASSETS_URL}/{architecture}"
         if not self.kernel_url:

uvm/domain.py

@@ -11,6 +11,7 @@ from typing import Any
 
 from .config import MAX_STORAGE_MIB, MIN_STORAGE_MIB
 from .errors import StateError, ValidationError
+from .tool_catalog import ToolSelection
 
 
 STATE_SCHEMA_VERSION = 3
@@ -55,6 +56,7 @@ class VmSpec:
     name: str | None = None
     ports: tuple[PortRule, ...] = (PortRule(22),)
     storage_mib: int = MIN_STORAGE_MIB
+    tools: tuple[ToolSelection, ...] = ()
 
 
 @dataclass(frozen=True, slots=True)

uvm/guest_exec.py

@@ -111,4 +111,10 @@ class GuestSession:
             ) from error
 
     def close(self) -> None:
+        # Wake a thread waiting for a long-running tool command on server shutdown.
+        # The guest still owns its bounded command and package-manager cleanup.
+        try:
+            self._connection.shutdown(socket.SHUT_RDWR)
+        except OSError:
+            pass
         self._connection.close()

uvm/guest_tools.py

@@ -0,0 +1,394 @@
+"""Standalone guest-only tool installer, submitted through the execution agent.
+
+Importing this module on the host has no side effects. It uses only Python 3.10
+stdlib; all package commands, downloads, and filesystem changes run in the guest.
+"""
+
+from __future__ import annotations
+
+import fcntl
+import hashlib
+import json
+import os
+import pwd
+import re
+import shlex
+import shutil
+import stat
+import subprocess
+import sys
+import tarfile
+import tempfile
+import time
+import urllib.request
+from pathlib import Path
+
+
+ROOT = Path("/opt/uvm-tools")
+LOCK = Path("/run/uvm-tools.lock")
+BIN_DIR = Path("/usr/local/bin")
+PROFILE_DIR = Path("/etc/profile.d")
+PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
+LOG_LIMIT = 16384
+PHP_KEY = "B8DC7E53946656EFBCE4C1DD71DAEAAB4AD4CAB6"
+
+
+class ToolError(Exception):
+    pass
+
+
+def atomic_write(path, data, mode=0o644):
+    descriptor, name = tempfile.mkstemp(prefix=".uvm-tools-", dir=path.parent)
+    try:
+        with os.fdopen(descriptor, "wb") as stream:
+            os.fchmod(stream.fileno(), mode)
+            stream.write(data)
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(name, path)
+        directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
+        try:
+            os.fsync(directory)
+        finally:
+            os.close(directory)
+    finally:
+        Path(name).unlink(missing_ok=True)
+
+
+def safe_directory(path):
+    for parent in reversed((path, *path.parents)):
+        if not parent.exists():
+            parent.mkdir(mode=0o755)
+        metadata = parent.lstat()
+        if not stat.S_ISDIR(metadata.st_mode) or metadata.st_uid != os.geteuid() or metadata.st_mode & 0o022:
+            raise ToolError(f"tool installation needs a root-owned directory: {parent}")
+
+
+def regular_file(path):
+    metadata = path.lstat()
+    if not stat.S_ISREG(metadata.st_mode) or metadata.st_uid != os.geteuid() or metadata.st_nlink != 1 or metadata.st_mode & 0o022:
+        raise ToolError(f"refusing to replace an unmanaged tool file: {path}")
+
+
+def download(url, destination, digest=None, *, max_bytes=1024 * 1024 * 1024):
+    if not url.startswith("https://"):
+        raise ToolError("tool downloads require HTTPS")
+    checksum = hashlib.sha256()
+    size = 0
+    with urllib.request.urlopen(url, timeout=30) as response, destination.open("wb") as output:
+        if not response.geturl().startswith("https://"):
+            raise ToolError("tool download redirected away from HTTPS")
+        while True:
+            chunk = response.read(65536)
+            if not chunk:
+                break
+            size += len(chunk)
+            if size > max_bytes:
+                raise ToolError("tool download exceeds its size limit")
+            checksum.update(chunk)
+            output.write(chunk)
+    if digest is not None and checksum.hexdigest() != digest:
+        destination.unlink(missing_ok=True)
+        raise ToolError("tool download checksum did not match the pinned release")
+
+
+def extract_archive(archive_path, destination):
+    """Bound extraction and check each path, including previously extracted links."""
+    root = destination.resolve()
+    with tarfile.open(archive_path) as archive:
+        members = archive.getmembers()
+        if len(members) > 50000 or sum(member.size for member in members) > 4 * 1024**3:
+            raise ToolError("tool archive exceeds its extraction limit")
+        for member in members:
+            name = Path(member.name)
+            target = destination / name
+            if name.is_absolute() or ".." in name.parts or not target.resolve().is_relative_to(root):
+                raise ToolError("unsafe path in tool archive")
+            if not (member.isfile() or member.isdir() or member.issym() or member.islnk()):
+                raise ToolError("unsupported file in tool archive")
+            if member.issym() or member.islnk():
+                link = (target.parent if member.issym() else destination) / member.linkname
+                if Path(member.linkname).is_absolute() or not link.resolve().is_relative_to(root):
+                    raise ToolError("unsafe link in tool archive")
+            member.mode &= 0o755
+            member.uid = member.gid = 0
+            member.uname = member.gname = "root"
+            options = {"filter": "data"} if hasattr(tarfile, "data_filter") else {}
+            archive.extract(member, destination, **options)
+
+
+def platform_info():
+    values = {}
+    for line in Path("/etc/os-release").read_text().splitlines():
+        if "=" in line:
+            key, value = line.split("=", 1)
+            values[key] = value.strip('"')
+    arch = os.uname().machine
+    return {"os": values.get("PRETTY_NAME", "Unknown OS"), "architecture": arch,
+            "supported": values.get("ID") == "ubuntu" and values.get("VERSION_ID") == "22.04"
+                         and arch in ("x86_64", "aarch64"),
+            "boot_id": Path("/proc/sys/kernel/random/boot_id").read_text().strip()}
+
+
+def guest_busy():
+    try:
+        descriptor = os.open(LOCK, os.O_RDONLY | os.O_NOFOLLOW)
+    except FileNotFoundError:
+        return False
+    try:
+        try:
+            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            return False
+        except BlockingIOError:
+            return True
+    finally:
+        os.close(descriptor)
+
+
+def probe(tool, username):
+    account = pwd.getpwnam(username)
+    environment = {"PATH": PATH, "HOME": account.pw_dir, "USER": username, "LOGNAME": username,
+                   "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "PYTHONDONTWRITEBYTECODE": "1"}
+    result = {"key": tool["key"], "id": tool["id"], "version": tool["version"], "installed": False,
+              "detected_version": None, "error": None, "usage": tool["usage"]}
+    for index, command in enumerate(tool["checks"]):
+        if shutil.which(command[0], path=PATH) is None:
+            return result
+        try:
+            process = subprocess.run(command, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
+                                     stderr=subprocess.STDOUT, timeout=3, env=environment, cwd=account.pw_dir,
+                                     user=account.pw_uid, group=account.pw_gid,
+                                     extra_groups=os.getgrouplist(username, account.pw_gid))
+            output = process.stdout[:4096].decode("utf-8", errors="replace")
+        except (OSError, subprocess.TimeoutExpired) as error:
+            result["error"] = f"could not verify {command[0]}: {error}"[:512]
+            return result
+        if process.returncode != 0:
+            result["error"] = f"{command[0]} verification failed: {output}"[:512]
+            return result
+        match = re.search(tool["match"], output)
+        if index == 0:
+            if not match:
+                result["error"] = f"could not determine the installed {tool['id']} version"
+                return result
+            result["detected_version"] = match.group(1)
+            if not result["detected_version"].startswith(tool["prefix_match"]):
+                return result
+        elif tool["id"] == "jdk" and (not match or not match.group(1).startswith("21.")):
+            return result
+    result["installed"] = True
+    return result
+
+
+class Installer:
+    def __init__(self, username):
+        self.username = username
+        self.log = bytearray()
+        self.truncated = False
+        self.lock = None
+        self.managed = {}
+
+    def append(self, data):
+        self.log.extend(data)
+        if len(self.log) > LOG_LIMIT:
+            del self.log[:-LOG_LIMIT]
+            self.truncated = True
+
+    def run(self, command):
+        self.append(("+ " + shlex.join(map(str, command)) + "\n").encode())
+        environment = {**os.environ, "PATH": PATH, "LC_ALL": "C.UTF-8", "DEBIAN_FRONTEND": "noninteractive"}
+        with subprocess.Popen(command, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
+                              stderr=subprocess.STDOUT, env=environment, pass_fds=(self.lock,)) as process:
+            while True:
+                chunk = process.stdout.read(8192)
+                if not chunk:
+                    break
+                self.append(chunk)
+            if process.wait() != 0:
+                raise ToolError(f"{command[0]} failed with exit code {process.returncode}; see the installation output")
+
+    def remember(self, path, value):
+        name = str(path)
+        self.managed[name] = list(dict.fromkeys([*self.managed.get(name, []), value]))[-2:]
+        atomic_write(ROOT / "managed.json", json.dumps(self.managed).encode(), 0o600)
+
+    def write_managed(self, path, data, mode=0o644):
+        safe_directory(path.parent)
+        digest = hashlib.sha256(data).hexdigest()
+        if path.exists() or path.is_symlink():
+            regular_file(path)
+            current = hashlib.sha256(path.read_bytes()).hexdigest()
+            if current == digest:
+                return
+            if current not in self.managed.get(str(path), []):
+                raise ToolError(f"customized file needs attention before installing: {path}")
+        # Record intent first so an interrupted publication can be retried.
+        self.remember(path, digest)
+        atomic_write(path, data, mode)
+
+    def link(self, name, target):
+        path = BIN_DIR / name
+        safe_directory(path.parent)
+        target = str(target)
+        if path.exists() or path.is_symlink():
+            if path.is_symlink() and os.readlink(path) == target:
+                return
+            if not path.is_symlink() or os.readlink(path) not in self.managed.get(str(path), []):
+                raise ToolError(f"customized executable needs attention before installing: {path}")
+        self.remember(path, target)
+        temporary = path.with_name(".uvm-" + name)
+        temporary.unlink(missing_ok=True)
+        temporary.symlink_to(target)
+        os.replace(temporary, path)
+
+    def apt(self, packages, *, refresh=False):
+        stamp = ROOT / "apt-updated"
+        if refresh or not stamp.exists() or time.time() - stamp.stat().st_mtime > 300:
+            self.run(["/usr/bin/apt-get", "-o", "DPkg::Lock::Timeout=300", "-o", "Acquire::Retries=2",
+                      "-o", "APT::Update::Error-Mode=any", "update"])
+            atomic_write(stamp, b"updated\n", 0o600)
+        pending = ROOT / "apt-pending"
+        if pending.exists():
+            # Complete configuration left by an interrupted tool installation.
+            self.run(["/usr/bin/dpkg", "--configure", "-a"])
+        atomic_write(pending, b"pending\n", 0o600)
+        self.run(["/usr/bin/apt-get", "-o", "DPkg::Lock::Timeout=300", "-o", "Acquire::Retries=2",
+                  "-o", "Dpkg::Options::=--force-confold", "-y", "--no-install-recommends", "install", *packages])
+        pending.unlink()
+
+    def php_repository(self, temporary):
+        self.apt(["ca-certificates", "gnupg"])
+        key = temporary / "php.asc"
+        download(f"https://keyserver.ubuntu.com/pks/lookup?op=get&search=0x{PHP_KEY}", key, max_bytes=65536)
+        result = subprocess.run(["gpg", "--batch", "--homedir", str(temporary), "--with-colons", "--show-keys", str(key)],
+                                capture_output=True, check=True)
+        fingerprints = [line.split(":")[9] for line in result.stdout.decode().splitlines() if line.startswith("fpr:")]
+        if not fingerprints or fingerprints[0] != PHP_KEY:
+            raise ToolError("PHP repository signing key did not match its pinned fingerprint")
+        binary_key = temporary / "php.gpg"
+        self.run(["gpg", "--batch", "--homedir", str(temporary), "--dearmor", "--output", str(binary_key), str(key)])
+        self.write_managed(Path("/etc/apt/keyrings/uvm-php.gpg"), binary_key.read_bytes())
+        self.write_managed(Path("/etc/apt/sources.list.d/uvm-php.list"),
+                           b"deb [signed-by=/etc/apt/keyrings/uvm-php.gpg] https://ppa.launchpadcontent.net/ondrej/php/ubuntu jammy main\n")
+
+    def payload(self, tool, temporary):
+        prefix = ROOT / tool["release"]
+        marker = prefix / ".uvm-tools.json"
+        identity = {"release": tool["release"], "sha256": tool["sha256"]}
+        if prefix.exists() or prefix.is_symlink():
+            safe_directory(prefix)
+            if not marker.exists():
+                raise ToolError(f"installation directory is already in use: {prefix}")
+            regular_file(marker)
+            previous = json.loads(marker.read_bytes())
+            if any(previous.get(key) != value for key, value in identity.items()):
+                raise ToolError(f"installation directory belongs to a different release: {prefix}")
+            if tool["kind"] != "conda" or previous.get("complete"):
+                return prefix
+        artifact = temporary / "download"
+        self.append(f"Downloading {tool['release']} and verifying SHA-256…\n".encode())
+        download(tool["url"], artifact, tool["sha256"])
+        if tool["kind"] == "archive":
+            extraction = temporary / "extract"
+            extraction.mkdir()
+            extract_archive(artifact, extraction)
+            source = extraction / tool["archive_root"]
+            if not source.is_dir() or source.is_symlink():
+                raise ToolError("tool archive has an unexpected layout")
+            atomic_write(source / ".uvm-tools.json", json.dumps(identity).encode(), 0o600)
+            os.replace(source, prefix)
+        else:
+            prefix.mkdir(exist_ok=True, mode=0o755)
+            atomic_write(marker, json.dumps(identity).encode(), 0o600)
+            if tool["kind"] == "conda":
+                self.run(["/bin/bash", str(artifact), "-b", "-f", "-p", str(prefix)])
+            else:
+                artifact.chmod(0o755)
+                os.replace(artifact, prefix / "micromamba")
+        return prefix
+
+    def configure(self, tool, prefix):
+        for name, target in tool["links"].items():
+            self.link(name, Path(target) if target.startswith("/") else prefix / target)
+        if tool["id"] == "python":
+            executable = prefix / "bin" / f"python{tool['version']}"
+            self.write_managed(BIN_DIR / f"pip{tool['version']}",
+                               f'#!/bin/sh\nexec {shlex.quote(str(executable))} -m pip "$@"\n'.encode(), 0o755)
+        elif tool["id"] == "jdk":
+            self.write_managed(PROFILE_DIR / "uvm-java.sh",
+                               f"export JAVA_HOME={shlex.quote(tool['java_home'])}\n".encode())
+        elif tool["id"] == "conda":
+            for key, value in (("auto_activate_base", "false"),):
+                self.run([str(prefix / "bin/conda"), "config", "--system", "--set", key, value])
+            for key, value in (("envs_dirs", "~/.conda/envs"), ("pkgs_dirs", "~/.conda/pkgs")):
+                self.run([str(prefix / "bin/conda"), "config", "--system", "--prepend", key, value])
+            self.write_managed(PROFILE_DIR / "uvm-conda.sh",
+                               f". {shlex.quote(str(prefix / 'etc/profile.d/conda.sh'))}\n".encode())
+        elif tool["id"] == "micromamba":
+            self.write_managed(PROFILE_DIR / "uvm-micromamba.sh",
+                               b'export MAMBA_ROOT_PREFIX="${MAMBA_ROOT_PREFIX:-$HOME/micromamba}"\n'
+                               b'eval "$(/usr/local/bin/micromamba shell hook --shell posix)"\n')
+
+    def install(self, tool):
+        safe_directory(LOCK.parent)
+        self.lock = os.open(LOCK, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
+        try:
+            metadata = os.fstat(self.lock)
+            if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1 or metadata.st_uid != os.geteuid():
+                raise ToolError("unsafe guest tool installation lock")
+            try:
+                fcntl.flock(self.lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except BlockingIOError:
+                raise ToolError("another installation is still running inside this guest; wait before retrying")
+            existing = probe(tool, self.username)
+            if existing["installed"]:
+                return {"status": "already_installed", "tool": existing}
+            safe_directory(ROOT)
+            manifest = ROOT / "managed.json"
+            if manifest.exists():
+                regular_file(manifest)
+                self.managed = json.loads(manifest.read_bytes())
+            with tempfile.TemporaryDirectory(prefix=".install-", dir=ROOT) as name:
+                temporary = Path(name)
+                if tool["kind"] == "php":
+                    self.php_repository(temporary)
+                self.apt(tool["packages"], refresh=tool["kind"] == "php")
+                prefix = self.payload(tool, temporary) if "url" in tool else None
+                self.configure(tool, prefix)
+                installed = probe(tool, self.username)
+                if not installed["installed"]:
+                    raise ToolError(installed["error"] or f"{tool['key']} did not pass its version check after installation")
+                if prefix is not None:
+                    atomic_write(prefix / ".uvm-tools.json", json.dumps({"release": tool["release"],
+                                 "sha256": tool["sha256"], "complete": True}).encode(), 0o600)
+                return {"status": "installed", "tool": installed}
+        finally:
+            os.close(self.lock)
+            self.lock = None
+
+
+def main(payload):
+    installer = Installer(payload["username"])
+    try:
+        if os.geteuid() != 0:
+            raise ToolError("tool installation and inspection require guest root")
+        info = platform_info()
+        if not info["supported"] or info["architecture"] != payload["architecture"]:
+            raise ToolError("one-click tools require an Ubuntu 22.04 guest on the selected x86_64/ARM64 architecture")
+        if payload.get("boot_id") and payload["boot_id"] != info["boot_id"]:
+            raise ToolError("the guest rebooted during installation; retry on the current boot")
+        if payload["action"] == "probe":
+            result = {**info, "busy": guest_busy(), "tools": [probe(tool, payload["username"]) for tool in payload["tools"]]}
+        elif payload["action"] == "install":
+            result = installer.install(payload["tools"][0])
+        else:
+            raise ToolError("unsupported tool operation")
+        result["ok"] = True
+    except (ToolError, OSError, ValueError, KeyError, tarfile.TarError, subprocess.SubprocessError) as error:
+        result = {"ok": False, "error": str(error)[:2048]}
+    result.update(output=installer.log.decode("utf-8", errors="replace"), truncated=installer.truncated)
+    print(json.dumps(result, ensure_ascii=True))
+
+
+if __name__ == "__main__":
+    main(json.loads(sys.argv[1]))

uvm/lifecycle.py

@@ -26,6 +26,8 @@ from .known_hosts import KnownHostsManager
 from .network import NetworkManager
 from .state import StateStore
 from .system import check_kvm, ensure_data_directories, require_root
+from .tool_catalog import normalize_tools
+from .tool_store import ToolJobStore
 from .validation import parse_name, parse_password, parse_ports, parse_timeout, parse_username
 
 
@@ -75,6 +77,7 @@ class LifecycleService:
         self._guest_connector = guest_connector
         self._terminal_connector = terminal_connector
         self._clock = clock
+        self._tool_store = ToolJobStore(settings, state_store, clock=clock)
 
     def create(self, spec: VmSpec) -> VmRecord:
         require_root()
@@ -84,6 +87,7 @@ class LifecycleService:
         if spec.name is not None:
             parse_name(spec.name)
         ports = self._validated_ports(spec.ports)
+        tools = normalize_tools(spec.tools, allow_empty=True)
         ensure_data_directories(self._settings)
         with self._state_store.operation_lock():
             self._check_boot()
@@ -100,6 +104,8 @@ class LifecycleService:
                 self._images.provision_guest_agent(disk, vm.id, vm.username)
                 vm.disk_ready = True
                 self._save(vm)
+                if tools:
+                    self._tool_store.enqueue_locked(vm.id, tools, source="create")
                 self._boot(vm, assets.kernel)
                 return vm
             except BaseException as error:
@@ -506,6 +512,8 @@ class LifecycleService:
     def execute(
         self, vm_id: str, command: str, *, sudo: bool = False,
         timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
+        expected_identity: tuple | None = None,
+        on_connect: Callable[[GuestSession], None] | None = None,
     ) -> ExecResult:
         """Pin a running VM's guest connection under the lock, then wait unlocked."""
         require_root()
@@ -516,14 +524,52 @@ class LifecycleService:
             if (vm.status != "running" or vm.operation is not None or vm.pending_resize is not None
                     or self._process.observe(vm) != "alive"):
                 raise ConflictError("guest execution requires a running VM with no pending lifecycle operation")
+            if expected_identity is not None and expected_identity != (vm.pid, vm.process_start_time, vm.boot_id):
+                raise ConflictError("the VM restarted during tool installation; retry on the current boot")
             session = self._guest_connector(self._settings.guest_socket_path(vm.id), vm.id, vm.username, vm.pid)
         try:
+            if on_connect is not None:
+                on_connect(session)
             # This session stays attached to that VMM even if the VM is stopped,
             # deleted, or restarted concurrently. Never reconnect/retry a command.
             return session.execute(command, sudo=sudo, timeout_seconds=timeout_seconds)
         finally:
             session.close()
 
+    def _validate_tool_target(self, vm: VmRecord, *, start_if_stopped: bool, restart: bool) -> str:
+        self._not_deleting(vm)
+        observation = self._process.observe(vm)
+        if vm.operation is not None or vm.pending_resize is not None or not vm.disk_ready:
+            raise ConflictError("finish pending lifecycle operations before installing tools")
+        if vm.status == "running" and observation == "alive":
+            return "running"
+        if vm.status == "stopped" and observation == "gone" and (start_if_stopped or restart):
+            self._validate_resume(vm)
+            return "stopped"
+        raise ConflictError("tool installation requires a running VM; use Start & install for a stopped VM")
+
+    def queue_tools(self, vm_id: str, selections, *, start_if_stopped=False, restart=False, request_id=None) -> dict:
+        require_root()
+        selections = normalize_tools(selections)
+        with self._state_store.operation_lock():
+            vm = self._load_for_operation(vm_id, exact_id=True)
+            self._validate_tool_target(vm, start_if_stopped=start_if_stopped, restart=restart)
+            return self._tool_store.enqueue_locked(vm.id, selections, start_if_stopped=start_if_stopped,
+                                                   restart=restart, request_id=request_id)
+
+    def prepare_tool_install(self, vm_id: str, *, start_if_stopped=False, restart=False) -> VmRecord:
+        """Perform requested startup/agent refresh atomically before a tool worker connects."""
+        require_root()
+        with self._state_store.operation_lock():
+            vm = self._load_for_operation(vm_id, exact_id=True)
+            status = self._validate_tool_target(vm, start_if_stopped=start_if_stopped, restart=restart)
+            if status == "running" and restart:
+                self._stop_locked(vm)
+            if status == "stopped" or restart:
+                # The normal retained-disk boot also provisions the current agent.
+                vm = self._start_locked(vm)
+            return vm
+
     def open_terminal(self, vm_id: str, *, rows: int = 24, cols: int = 80) -> tuple[GuestTerminal, str]:
         """Connect to the exact running VMM under the lock, then open its guest PTY."""
         require_root()

uvm/tool_store.py

@@ -0,0 +1,195 @@
+"""Atomic per-VM tool job records, serialized with lifecycle filesystem changes."""
+
+from __future__ import annotations
+
+import copy
+import json
+import os
+import re
+import stat
+import tempfile
+import time
+import uuid
+from contextlib import contextmanager
+
+from .config import Settings
+from .errors import ConflictError, NotFoundError, StateError
+from .state import StateStore
+from .tool_catalog import normalize_tools
+
+
+ACTIVE_JOB_STATUSES = frozenset({"queued", "starting", "waiting", "installing"})
+MAX_RETAINED_JOBS = 10
+MAX_DOCUMENT_BYTES = 16 * 1024 * 1024
+
+
+class ToolJobStore:
+    """Only short transactions hold operations.lock; guest execution never does."""
+
+    def __init__(self, settings: Settings, state_store: StateStore, *, clock=time.time) -> None:
+        self.settings = settings
+        self.state = state_store
+        self.clock = clock
+
+    def _path(self, vm_id):
+        vm = self.state.get(vm_id)  # Exact ID only, including while creating a VM.
+        if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,80}", vm.id):
+            raise StateError("invalid VM ID in tool history")
+        directory = self.settings.vm_dir(vm.id)
+        try:
+            metadata = directory.lstat()
+        except FileNotFoundError as error:
+            raise NotFoundError("VM runtime directory is no longer available") from error
+        if (not stat.S_ISDIR(metadata.st_mode) or metadata.st_uid != os.geteuid()
+                or metadata.st_mode & 0o077 or not vm.runtime_owned):
+            raise StateError("tool history requires a private, owned VM runtime directory")
+        return directory / "tools.json"
+
+    @staticmethod
+    def _empty():
+        return {"schema_version": 1, "jobs": [], "inventory": [], "checked_at": None,
+                "agent_ready": False, "guest_busy": False, "supported": None, "error": None}
+
+    def _read(self, vm_id):
+        path = self._path(vm_id)
+        try:
+            descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
+        except FileNotFoundError:
+            return self._empty()
+        try:
+            with os.fdopen(descriptor, "rb") as stream:
+                metadata = os.fstat(stream.fileno())
+                if (not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1
+                        or metadata.st_uid != os.geteuid() or metadata.st_mode & 0o077):
+                    raise StateError("unsafe tool history file")
+                raw = stream.read(MAX_DOCUMENT_BYTES + 1)
+                if len(raw) > MAX_DOCUMENT_BYTES:
+                    raise StateError("tool history exceeds its size limit")
+            data = json.loads(raw)
+            if (not isinstance(data, dict) or data.get("schema_version") != 1
+                    or not isinstance(data.get("jobs"), list) or not isinstance(data.get("inventory"), list)
+                    or len(data["jobs"]) > MAX_RETAINED_JOBS):
+                raise StateError("invalid tool history document")
+            for job in data["jobs"]:
+                if (not isinstance(job, dict) or job.get("vm_id") != vm_id
+                        or not re.fullmatch(r"tools-[0-9a-f]{32}", str(job.get("id", "")))
+                        or not isinstance(job.get("items"), list) or len(job["items"]) > 12):
+                    raise StateError("invalid tool installation job")
+            return data
+        except (OSError, ValueError) as error:
+            raise StateError("could not read tool installation history") from error
+
+    def _write(self, vm_id, data):
+        path = self._path(vm_id)  # Never recreate a deleted VM directory.
+        descriptor, name = tempfile.mkstemp(prefix=".tools-", suffix=".json", dir=path.parent)
+        try:
+            with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
+                os.fchmod(stream.fileno(), 0o600)
+                json.dump(data, stream, ensure_ascii=True, allow_nan=False)
+                stream.write("\n")
+                stream.flush()
+                os.fsync(stream.fileno())
+            os.replace(name, path)
+            self.state._fsync_parent(path)
+        except OSError as error:
+            raise StateError("could not save tool installation history") from error
+        finally:
+            try:
+                os.unlink(name)
+            except FileNotFoundError:
+                pass
+
+    @contextmanager
+    def transaction(self, vm_id):
+        with self.state.operation_lock():
+            data = self._read(vm_id)
+            yield data
+            self._write(vm_id, data)
+
+    def read(self, vm_id):
+        with self.state.operation_lock():
+            return self._read(vm_id)
+
+    def enqueue_locked(self, vm_id, selections, *, start_if_stopped=False, restart=False,
+                       request_id=None, source="user"):
+        """Called by lifecycle create/submit with operations.lock already held."""
+        selections = normalize_tools(selections)
+        data = self._read(vm_id)
+        wanted = [item.to_dict() for item in selections]
+        for previous in reversed(data["jobs"]):
+            same = (previous["selections"] == wanted and previous["start_if_stopped"] == start_if_stopped
+                    and previous.get("restart", False) == restart)
+            if request_id is not None and previous.get("request_id") == request_id:
+                if not same:
+                    raise ConflictError("this installation request ID was already used for different tools")
+                return copy.deepcopy(previous)
+            if previous["status"] in ACTIVE_JOB_STATUSES:
+                if same:
+                    return copy.deepcopy(previous)
+                raise ConflictError("a tool installation is already queued or running for this VM")
+        now = self.clock()
+        job = {"id": "tools-" + uuid.uuid4().hex, "vm_id": vm_id, "status": "queued", "source": source,
+               "selections": wanted, "start_if_stopped": start_if_stopped, "restart": restart,
+               "request_id": request_id, "created_at": now, "updated_at": now, "completed_at": None,
+               "error": None, "outcome_unknown": False,
+               "items": [{**item.to_dict(), "key": item.key, "status": "queued", "detected_version": None,
+                          "error": None, "output": "", "truncated": False} for item in selections]}
+        data["jobs"] = [*data["jobs"], job][-MAX_RETAINED_JOBS:]
+        self._write(vm_id, data)
+        return copy.deepcopy(job)
+
+    def get_job(self, vm_id, job_id):
+        for job in self.read(vm_id)["jobs"]:
+            if job["id"] == job_id:
+                return job
+        raise NotFoundError("tool installation job not found for this VM")
+
+    def update_job(self, vm_id, job_id, update):
+        with self.transaction(vm_id) as data:
+            for job in data["jobs"]:
+                if job["id"] == job_id:
+                    update(job, data)
+                    job["updated_at"] = self.clock()
+                    return copy.deepcopy(job)
+        raise NotFoundError("tool installation job not found for this VM")
+
+    def claim(self):
+        with self.state.operation_lock():
+            for vm in self.state.load().vms.values():
+                # A failed/interrupted create may not own a runtime directory.
+                if not vm.runtime_owned or not self.settings.vm_dir(vm.id).is_dir():
+                    continue
+                data = self._read(vm.id)
+                for job in data["jobs"]:
+                    if job["status"] == "queued":
+                        job["status"] = "starting" if vm.status == "stopped" or job.get("restart") else "waiting"
+                        job["updated_at"] = self.clock()
+                        self._write(vm.id, data)
+                        return copy.deepcopy(job)
+        return None
+
+    @staticmethod
+    def interrupt(job, message, now, *, outcome_unknown=True):
+        job.update(status="interrupted", error=message, outcome_unknown=outcome_unknown,
+                   updated_at=now, completed_at=now)
+        for item in job["items"]:
+            if item["status"] in {"installing", "queued"}:
+                item["status"] = "interrupted" if item["status"] == "installing" else "skipped"
+                item["error"] = message
+
+    def recover(self):
+        """The worker owner lock proves that previous in-flight host workers exited."""
+        with self.state.operation_lock():
+            for vm in self.state.load().vms.values():
+                if not vm.runtime_owned or not self.settings.vm_dir(vm.id).is_dir():
+                    continue
+                data = self._read(vm.id)
+                changed = False
+                for job in data["jobs"]:
+                    if job["status"] in ACTIVE_JOB_STATUSES - {"queued"}:
+                        self.interrupt(job, "Management server restarted. Refresh tool status before retrying; "
+                                       "the last installation may have completed inside the guest.", self.clock())
+                        changed = True
+                if changed:
+                    data.update(agent_ready=False, guest_busy=False)
+                    self._write(vm.id, data)

uvm/tools.py

@@ -0,0 +1,319 @@
+"""Bounded background installation workers using the existing guest exec channel."""
+
+from __future__ import annotations
+
+import fcntl
+import json
+import logging
+import os
+import re
+import shlex
+import stat
+import threading
+import time
+from functools import lru_cache
+from pathlib import Path
+
+from . import guest_tools
+from .errors import ConflictError, GuestAgentUnavailableError, GuestExecutionError, NotFoundError, UvmError, ValidationError
+from .guest_exec import validate_exec
+from .system import require_root
+from .tool_catalog import all_selections, normalize_tools, recipe
+from .tool_store import ToolJobStore
+
+
+@lru_cache(maxsize=1)
+def _guest_source():
+    return Path(guest_tools.__file__).read_text(encoding="utf-8")
+
+
+def guest_command(payload: dict) -> str:
+    command = shlex.join(["/usr/bin/python3", "-I", "-c", _guest_source(), json.dumps(payload, ensure_ascii=True)])
+    validate_exec(command, True, 60)
+    return command
+
+
+def _identity(vm):
+    return vm.pid, vm.process_start_time, vm.boot_id
+
+
+def _text(value, limit=16384):
+    return str(value or "").encode("utf-8", errors="replace")[-limit:].decode("utf-8", errors="replace")
+
+
+class ToolService:
+    def __init__(self, settings, state_store, lifecycle) -> None:
+        self.settings = settings
+        self.state = state_store
+        self.lifecycle = lifecycle
+        self.store = ToolJobStore(settings, state_store)
+        self._mutex = threading.RLock()
+        self._wake = threading.Condition()
+        self._stop = threading.Event()
+        self._threads = []
+        self._sessions = {}
+        self._owner = None
+        self._workers_left = 0
+
+    def start(self):
+        with self._mutex:
+            if self._stop.is_set():
+                raise ConflictError("the tool installer is shutting down")
+            if self._owner is not None:
+                return
+            with self.state.operation_lock():
+                descriptor = os.open(self.settings.base / "tool-worker.lock", os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
+                try:
+                    metadata = os.fstat(descriptor)
+                    if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1 or metadata.st_uid != os.geteuid():
+                        raise UvmError("unsafe tool worker ownership file")
+                    os.fchmod(descriptor, 0o600)
+                    fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                except BaseException as error:
+                    os.close(descriptor)
+                    if isinstance(error, BlockingIOError):
+                        raise ConflictError("another management server is already processing tool installations for this registry") from error
+                    raise
+            self._owner = descriptor
+            try:
+                self.store.recover()
+                self._workers_left = self.settings.tool_workers
+                self._threads = [threading.Thread(target=self._worker, name=f"uvm-tools-{index}", daemon=True)
+                                 for index in range(self.settings.tool_workers)]
+                for thread in self._threads:
+                    thread.start()
+            except BaseException:
+                self._release_owner()
+                raise
+
+    def _release_owner(self):
+        if self._owner is not None:
+            os.close(self._owner)
+            self._owner = None
+
+    def notify(self):
+        with self._wake:
+            self._wake.notify_all()
+
+    def shutdown(self):
+        self._stop.set()
+        self.notify()
+        with self._mutex:
+            sessions = list(self._sessions.values())
+        for session in sessions:
+            session.close()
+        deadline = time.monotonic() + 8
+        for thread in self._threads:
+            thread.join(max(0, deadline - time.monotonic()))
+        # A worker finishing a host start keeps ownership until that start ends.
+        # Its guest command is never submitted after shutdown was requested.
+        with self._mutex:
+            if not self._workers_left:
+                self._release_owner()
+
+    def submit(self, vm_id, selections, *, start_if_stopped=False, restart=False, request_id=None):
+        require_root()
+        selections = normalize_tools(selections)
+        if type(start_if_stopped) is not bool or type(restart) is not bool:
+            raise ValidationError("start_if_stopped and restart must be booleans")
+        if request_id is not None and (not isinstance(request_id, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", request_id)):
+            raise ValidationError("request_id must contain 1–80 letters, numbers, underscores, or hyphens")
+        self.start()
+        job = self.lifecycle.queue_tools(vm_id, selections, start_if_stopped=start_if_stopped,
+                                         restart=restart, request_id=request_id)
+        self.notify()
+        return job
+
+    def _execute(self, vm, payload, timeout):
+        if self._stop.is_set():
+            raise InterruptedError("management server is shutting down")
+        thread_id = threading.get_ident()
+
+        def connected(session):
+            with self._mutex:
+                self._sessions[thread_id] = session
+                if self._stop.is_set():
+                    session.close()
+
+        try:
+            result = self.lifecycle.execute(vm.id, guest_command(payload), sudo=True, timeout_seconds=timeout,
+                                            expected_identity=_identity(vm), on_connect=connected)
+        finally:
+            with self._mutex:
+                self._sessions.pop(thread_id, None)
+        if result.timed_out:
+            return {"ok": False, "error": f"installation/verification timed out after {timeout:g} seconds; retry to finish it",
+                    "output": _text(result.stderr), "truncated": result.truncated}
+        try:
+            value = json.loads(result.stdout)
+            if result.exit_code != 0 or result.truncated or not isinstance(value, dict) or type(value.get("ok")) is not bool:
+                raise ValueError("invalid result")
+            return value
+        except (ValueError, TypeError) as error:
+            raise GuestExecutionError("guest returned an incomplete tool result; refresh status before retrying",
+                                      outcome_unknown=payload["action"] == "install") from error
+
+    def _payload(self, vm, action, selections, *, boot_id=None):
+        return {"action": action, "username": vm.username, "architecture": self.settings.architecture,
+                "boot_id": boot_id, "tools": [recipe(selection, self.settings.architecture) for selection in selections]}
+
+    @staticmethod
+    def _inventory(result, selections):
+        expected = {item.key: item for item in selections}
+        records = result.get("tools")
+        if not isinstance(records, list) or len(records) != len(expected):
+            raise GuestExecutionError("invalid guest tool inventory")
+        inventory = []
+        for value in records:
+            if (not isinstance(value, dict) or value.get("key") not in expected
+                    or type(value.get("installed")) is not bool):
+                raise GuestExecutionError("invalid guest tool inventory")
+            selection = expected.pop(value["key"])
+            inventory.append({**selection.to_dict(), "key": selection.key, "installed": value["installed"],
+                              "detected_version": _text(value.get("detected_version"), 120) or None,
+                              "error": _text(value.get("error"), 512) or None,
+                              "usage": _text(value.get("usage"), 512)})
+        return inventory
+
+    def snapshot(self, vm_id, *, refresh=False):
+        require_root()
+        vm = self.state.get(vm_id)
+        data = self.store.read(vm_id)
+        running = self.lifecycle.inspect(vm.id).observed_status == "running"
+        if refresh and running:
+            try:
+                result = self._execute(vm, self._payload(vm, "probe", all_selections()), 90)
+                if not result["ok"]:
+                    raise UvmError(_text(result.get("error"), 2048))
+                inventory = self._inventory(result, all_selections())
+                with self.store.transaction(vm_id) as current:
+                    current.update(inventory=inventory, checked_at=time.time(), error=None, agent_ready=True,
+                                   guest_busy=result.get("busy") is True, supported=result.get("supported") is True)
+                data = self.store.read(vm_id)
+            except UvmError as error:
+                with self.store.transaction(vm_id) as current:
+                    current.update(error=_text(error, 2048), agent_ready=not isinstance(error, GuestAgentUnavailableError))
+                data = self.store.read(vm_id)
+        if not running:
+            data.update(agent_ready=False, guest_busy=False)
+        return {"vm_id": vm_id, **{key: data[key] for key in (
+            "inventory", "checked_at", "agent_ready", "guest_busy", "supported", "error")},
+            "stale": not running or data["checked_at"] is None or time.time() - data["checked_at"] > 60,
+            "latest_job": data["jobs"][-1] if data["jobs"] else None,
+            "jobs": [{key: job[key] for key in ("id", "status", "created_at", "completed_at")}
+                     for job in reversed(data["jobs"])]}
+
+    def _worker(self):
+        try:
+            while not self._stop.is_set():
+                try:
+                    job = self.store.claim()
+                    if job is not None:
+                        self._run_job(job)
+                        continue
+                except (UvmError, OSError):
+                    logging.getLogger(__name__).exception("Could not process a tool installation job")
+                with self._wake:
+                    self._wake.wait(timeout=1)
+        finally:
+            with self._mutex:
+                self._workers_left -= 1
+                if self._workers_left == 0 and self._stop.is_set():
+                    self._release_owner()
+
+    def _update(self, job, update):
+        return self.store.update_job(job["vm_id"], job["id"], update)
+
+    def _wait_ready(self, vm, selections):
+        deadline = time.monotonic() + self.settings.tool_ready_timeout_s
+        while not self._stop.is_set():
+            try:
+                result = self._execute(vm, self._payload(vm, "probe", selections), 90)
+                if not result["ok"]:
+                    raise UvmError(_text(result.get("error"), 2048))
+                if result.get("busy"):
+                    raise ConflictError("an earlier installation is still running inside this VM; wait and refresh before retrying")
+                if not isinstance(result.get("boot_id"), str) or not result["boot_id"]:
+                    raise GuestExecutionError("guest did not report its boot identity")
+                return result
+            except GuestAgentUnavailableError:
+                if time.monotonic() >= deadline:
+                    raise GuestAgentUnavailableError("Guest agent did not become ready. Restart & retry to apply agent setup, "
+                                                     "then check the VM log if startup still fails.")
+                self._stop.wait(min(2, max(0, deadline - time.monotonic())))
+        raise InterruptedError("management server is shutting down")
+
+    def _run_job(self, job):
+        submitted = False
+        try:
+            if self._stop.is_set():
+                raise InterruptedError("management server is shutting down")
+            selections = normalize_tools(job["selections"])
+            vm = self.lifecycle.prepare_tool_install(job["vm_id"], start_if_stopped=job["start_if_stopped"],
+                                                     restart=job.get("restart", False))
+            self._update(job, lambda current, data: current.update(status="waiting"))
+            info = self._wait_ready(vm, selections)
+            initial = self._inventory(info, selections)
+
+            def ready(current, data):
+                observed = {item["key"]: item for item in data["inventory"]}
+                observed.update({item["key"]: item for item in initial})
+                data.update(inventory=list(observed.values()), checked_at=time.time(), agent_ready=True,
+                            guest_busy=False, supported=True, error=None)
+            self._update(job, ready)
+            for index, selection in enumerate(selections):
+                if self._stop.is_set():
+                    raise InterruptedError("management server is shutting down")
+
+                def begin(current, data):
+                    current["status"] = "installing"
+                    current["items"][index]["status"] = "installing"
+                self._update(job, begin)
+                submitted = True
+                result = self._execute(vm, self._payload(vm, "install", [selection], boot_id=info["boot_id"]),
+                                       self.settings.tool_install_timeout_s)
+                observed = None
+                if result["ok"]:
+                    observed, = self._inventory({"tools": [result.get("tool")]}, [selection])
+                    if not observed["installed"] or result.get("status") not in {"installed", "already_installed"}:
+                        raise GuestExecutionError("guest installation did not verify the selected tool", outcome_unknown=True)
+
+                def finish_item(current, data):
+                    current["items"][index].update(
+                        status=result["status"] if result["ok"] else "failed",
+                        detected_version=observed["detected_version"] if observed else None,
+                        error=None if result["ok"] else _text(result.get("error"), 2048),
+                        output=_text(result.get("output")), truncated=result.get("truncated") is True)
+                    inventory = {item["key"]: item for item in data["inventory"]}
+                    if observed:
+                        inventory[selection.key] = observed
+                    elif selection.key in inventory:
+                        inventory[selection.key].update(installed=False, error="Refresh to verify after the failed installation.")
+                    data.update(inventory=list(inventory.values()), checked_at=time.time())
+                self._update(job, finish_item)
+                submitted = False
+
+            def finish(current, data):
+                failures = sum(item["status"] == "failed" for item in current["items"])
+                current.update(status="failed" if failures else "succeeded", completed_at=time.time(),
+                               error=f"{failures} tool installation(s) failed; see the per-tool output and retry." if failures else None)
+            self._update(job, finish)
+        except NotFoundError:
+            return  # Deletion owns removal of the whole runtime, including history.
+        except (InterruptedError, GuestExecutionError, ConflictError) as error:
+            try:
+                self._update(job, lambda current, data: self.store.interrupt(
+                    current, _text(error, 2048), time.time(),
+                    outcome_unknown=getattr(error, "outcome_unknown", submitted)))
+            except NotFoundError:
+                pass
+        except (UvmError, OSError, ValueError) as error:
+            def failed(current, data):
+                self.store.interrupt(current, _text(error, 2048), time.time(), outcome_unknown=submitted)
+                current["status"] = "failed" if not submitted else "interrupted"
+                if isinstance(error, GuestAgentUnavailableError):
+                    data.update(agent_ready=False, error=_text(error, 2048))
+            try:
+                self._update(job, failed)
+            except NotFoundError:
+                pass