electric-cherry/agent/ec_agent.py

#!/usr/bin/env python3
"""Electric Cherry host agent. Python 3 standard library; no remote shell API.

Each operation is recorded before execution. An interrupted operation becomes
unknown and is reported for reconciliation, never automatically executed twice.
"""

import argparse
import base64
from concurrent.futures import ThreadPoolExecutor, as_completed
import contextlib
from datetime import datetime, timezone
import errno
import math
import hashlib
import http.client
import fcntl
import json
import os
from pathlib import Path, PurePosixPath
import platform
import re
import shutil
import signal
import socket
import sqlite3
import stat as stat_module
import subprocess
import sys
import threading
import time
import types
import urllib.error
import urllib.parse
import urllib.request
import uuid

VERSION = "1.9.2"
LABEL = "io.electric-cherry"
MAX_FILE = 32 * 1024 * 1024
MANIFEST_MAX_ENTRIES = 10000
MANIFEST_MAX_BYTES = 16 * 1024 * 1024
MANIFEST_MAX_SECONDS = 15
MANIFEST_MAX_DEPTH = 64
LIBRARY_DOWNLOAD_WORKERS = 4
HEALTH_LOCK_TIMEOUT_SECONDS = 5
PLUTONIUM_MANIFEST_URL = "https://cdn.plutoniummod.com/updater/prod/info.json"
PLUTONIUM_MANIFEST_LIMIT = 2 * 1024 * 1024
PLUTONIUM_FILE_LIMIT = 512 * 1024 * 1024
PLUTONIUM_CACHE_LIMIT = 4 * 1024 * 1024 * 1024
BANK_GAME_ENDPOINT = "https://ec.thewrightserver.net/api/bank"
ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$")
AGENT_CONFIG = Path("/etc/electric-cherry/agent.json")
AGENT_EXECUTABLE = Path("/usr/local/libexec/electric-cherry-agent.py")
AGENT_UNIT = Path("/etc/systemd/system/electric-cherry-agent.service")
UNINSTALL_EXECUTABLE = Path("/usr/local/libexec/electric-cherry-uninstall.py")
UNINSTALL_UNIT = Path("/etc/systemd/system/electric-cherry-uninstall.service")
UNINSTALL_REQUEST = Path("/etc/electric-cherry/decommission.json")
UPDATE_EXECUTABLE = Path("/usr/local/libexec/electric-cherry-update.py")
UPDATE_UNIT = Path("/etc/systemd/system/electric-cherry-update.service")
UPDATE_TIMER = Path("/etc/systemd/system/electric-cherry-update.timer")
UPDATE_TIMER_ENABLED = Path("/etc/systemd/system/timers.target.wants/electric-cherry-update.timer")
UPDATE_JOURNAL = Path("/etc/electric-cherry/update/journal.json")
UPDATE_STATUS = Path("/etc/electric-cherry/update-status.json")
AGENT_HEALTH = Path("/etc/electric-cherry/agent-health.json")
ROOT_MARKER = ".electric-cherry.json"


class AgentError(Exception):
    pass


class Held(AgentError):
    """A reversible prerequisite prevented any disruptive action."""


class Rejected(AgentError):
    """A definitive precondition failure, before any destructive action."""


class AutoscalingInvalidated(Rejected):
    """This removal needs a new policy decision, never a later local retry."""


OPERATION_ERROR_PHASES = {
    "executing": "executing the operation",
    "checking-instance": "checking the existing instance",
    "validating-create-spec": "validating the container specification",
    "preparing-create-directory": "preparing the instance directory",
    "seeding-runtime": "seeding the runtime cache",
    "writing-create-files": "writing initial files",
    "persisting-create-metadata": "saving instance metadata",
    "creating-container": "creating the container",
    "starting-container": "starting the container",
    "recording-created-instance": "recording the started container",
    "inspecting-created-instance": "checking the created container",
    "removing-old-instance": "removing the old instance directory",
}


def unexpected_operation_error(error, phase):
    """Return bounded diagnostics without exception text, paths, or runtime data."""
    def attribute(value, name):
        try:
            return getattr(value, name, None)
        except Exception:
            return None

    def category(value):
        # Names come from this fixed list, never an arbitrary exception class.
        for kind in (urllib.error.HTTPError, urllib.error.URLError, http.client.IncompleteRead,
                     http.client.RemoteDisconnected, http.client.HTTPException,
                     subprocess.TimeoutExpired, TimeoutError,
                     PermissionError, FileNotFoundError, FileExistsError, IsADirectoryError,
                     NotADirectoryError, ConnectionError, OSError, sqlite3.OperationalError, sqlite3.Error,
                     json.JSONDecodeError, UnicodeError, KeyError, TypeError, ValueError,
                     MemoryError, RuntimeError):
            if isinstance(value, kind):
                return kind.__name__
        return "Exception"

    phase = phase if isinstance(phase, str) and phase in OPERATION_ERROR_PHASES else "executing"
    details = {"type": category(error), "phase": phase, "agentVersion": VERSION}
    evidence = error
    if isinstance(error, urllib.error.HTTPError):
        status = attribute(error, "code")
        if type(status) is int and 100 <= status <= 599:
            details["httpStatus"] = status
    elif isinstance(error, urllib.error.URLError):
        reason = attribute(error, "reason")
        if isinstance(reason, BaseException):
            details["causeType"] = category(reason)
            evidence = reason
    number = attribute(evidence, "errno") if isinstance(evidence, OSError) else None
    number = number if type(number) is int and number in errno.errorcode else None
    if number is not None:
        details.update(errno=number, errnoName=errno.errorcode[number])

    frames = []
    trace = attribute(error, "__traceback__")
    while isinstance(trace, types.TracebackType):
        # Only agent-owned code locations are safe to report. Exclude source
        # lines, filenames, other modules, frame locals, and exception chains.
        if trace.tb_frame.f_globals is globals():
            frames.append({"function": trace.tb_frame.f_code.co_name, "line": trace.tb_lineno})
        trace = trace.tb_next
    details["frames"] = frames[-8:]

    if details.get("httpStatus") in (401, 403):
        guidance = "Check host enrollment and controller artifact access permissions."
    elif details.get("httpStatus") == 404:
        guidance = "Check that the requested artifact is still available on the controller."
    elif isinstance(error, urllib.error.HTTPError):
        guidance = "Check controller availability and artifact delivery."
    elif number == errno.ENOSPC:
        guidance = "Check free disk space and inodes on the host."
    elif number == errno.EDQUOT:
        guidance = "Check the host filesystem quota."
    elif number in (errno.EACCES, errno.EPERM) or isinstance(evidence, PermissionError):
        guidance = "Check permissions on the host's managed files and directories."
    elif number == errno.EROFS:
        guidance = "Check whether the host filesystem is mounted read-only."
    elif isinstance(error, (urllib.error.URLError, http.client.HTTPException)) or isinstance(evidence, (TimeoutError, ConnectionError)):
        guidance = "Check host network, DNS, TLS, and controller connectivity."
    elif isinstance(error, subprocess.TimeoutExpired):
        guidance = "Check whether the host service command is stalled or unavailable."
    elif isinstance(evidence, FileNotFoundError):
        guidance = "Check that the required host files and directories are present."
    elif isinstance(evidence, OSError):
        guidance = "Check host storage, filesystem access, and service availability."
    else:
        guidance = "Inspect the saved operation diagnostics and host setup before retrying."
    labels = [details["type"]]
    if "httpStatus" in details:
        labels.append(f"HTTP {details['httpStatus']}")
    if "errnoName" in details:
        labels.append(f"{details['errnoName']} ({details['errno']})")
    if "causeType" in details:
        labels.append("cause " + details["causeType"])
    if frames:
        labels.append(f"at {frames[-1]['function']}:{frames[-1]['line']}")
    return f"Operation error while {OPERATION_ERROR_PHASES[phase]} ({'; '.join(labels)}). {guidance}", details


def systemctl(*args):
    try:
        result = subprocess.run(["systemctl", *args], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=60, check=False)
    except (OSError, subprocess.TimeoutExpired):
        raise AgentError("Host service management is unavailable") from None
    if result.returncode:
        raise AgentError("Host service management did not complete")


def protected_path(path, missing=False):
    """Deletion authority must never come from a runtime-writable path."""
    path = Path(path)
    for current in (*reversed(path.parents), path):
        try:
            info = current.lstat()
        except FileNotFoundError:
            if missing:
                continue
            raise AgentError("Host uninstall files are unavailable; reinstall the agent") from None
        if stat_module.S_ISLNK(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o022:
            raise AgentError("Host uninstall paths must be owned and writable only by root")
    return path


def protected_file(path, missing=False):
    path = protected_path(path, missing=missing)
    try:
        if not stat_module.S_ISREG(path.lstat().st_mode):
            raise AgentError("Host uninstall metadata must be a regular file")
    except FileNotFoundError:
        if not missing:
            raise
    return path


def sync_parent(path):
    descriptor = os.open(Path(path).parent, os.O_RDONLY | os.O_DIRECTORY)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def uninstall_root(value, identity=None):
    """Accept only a canonical, dedicated directory with no nested mounts."""
    if not isinstance(value, str) or not re.fullmatch(r"/[A-Za-z0-9/_.-]+", value):
        raise AgentError("Host data directory is unsafe to remove")
    root = Path(value)
    forbidden = {"/etc", "/usr", "/var", "/var/lib", "/var/log", "/var/cache", "/home", "/root", "/opt", "/srv", "/tmp", "/run", "/mnt", "/media", "/usr/local", "/usr/local/libexec"}
    if str(root) != value or ".." in root.parts or "." in root.parts or len(root.parts) < 3 or value in forbidden:
        raise AgentError("Host data directory is unsafe to remove")
    for installed in (AGENT_CONFIG, AGENT_EXECUTABLE, AGENT_UNIT, UNINSTALL_EXECUTABLE, UNINSTALL_UNIT, UPDATE_EXECUTABLE, UPDATE_UNIT, UPDATE_TIMER, AGENT_EXECUTABLE.with_suffix(".py.update"), UNINSTALL_EXECUTABLE.with_suffix(".py.update")):
        if installed == root or root in installed.parents:
            raise AgentError("Host data directory overlaps the agent installation")
    protected_path(root, missing=True)
    if root.exists():
        info = root.stat()
        if not stat_module.S_ISDIR(info.st_mode):
            raise AgentError("Host data directory is not a directory")
        if identity and [info.st_dev, info.st_ino] != identity:
            raise AgentError("Host data directory changed since deletion was authorized")
    try:
        # ProtectSystem/ReadWritePaths give the running agent a private bind
        # mount at its data directory. Inspect the host systemd namespace,
        # otherwise every legitimate installation appears to be a mount root.
        mount_lines = Path("/proc/1/mountinfo").read_text().splitlines()
    except FileNotFoundError:
        mount_lines = []  # Non-Linux tests; the installed service requires Linux.
    for line in mount_lines:
        fields = line.split()
        if len(fields) < 5:
            continue
        mount = re.sub(r"\\([0-7]{3})", lambda match: chr(int(match[1], 8)), fields[4])
        if mount == value or mount.startswith(value + "/"):
            raise AgentError("Unmount filesystems inside the host data directory before deletion")
    return root


def uninstall_ownership(config):
    root = uninstall_root(config.get("root"))
    marker = protected_file(root / ROOT_MARKER)
    try:
        saved = json.loads(marker.read_text())
    except (OSError, ValueError):
        raise AgentError("Host data ownership is unavailable; reinstall the agent") from None
    expected = {"version": 1, "root": str(root), "hostId": config.get("hostId"), "controller": config.get("controller", "").rstrip("/")}
    if saved != expected or not expected["hostId"]:
        raise AgentError("Host data ownership does not match this enrollment")
    return root


def digest(data):
    return hashlib.sha256(data).hexdigest()


def file_digest(path):
    h = hashlib.sha256()
    with open(path, "rb") as stream:
        for block in iter(lambda: stream.read(1024 * 1024), b""):
            h.update(block)
    return h.hexdigest()


def status_players(text):
    lines = text.splitlines()
    separator = next((index for index, line in enumerate(lines) if re.match(r"^\s*-+\s+-+", line)), None)
    # T6 includes a bot column; T5 places ping directly after score.
    if separator is not None and any(re.search(r"\bnum\s+score\s+(?:bot\s+)?ping\b", line, re.I) for line in lines[:separator]):
        rows = [line.strip() for line in lines[separator + 1:] if line.strip()]
        if any(not re.match(r"^\d+\s+", row) for row in rows):
            return None
        return len(rows)
    info = next((index for index, line in enumerate(lines) if line.startswith("\\") and "\\mapname\\" in line), None)
    if info is not None:
        return sum(bool(line.strip()) for line in lines[info + 1:])
    # A truncated header or an unfamiliar dialect is never proof of emptiness.
    return None


def identifier(value):
    if not isinstance(value, str) or not ID_RE.fullmatch(value):
        raise AgentError("Invalid identifier")
    return value


def instance_identity(instance_id, generation):
    try:
        if str(uuid.UUID(instance_id)) != instance_id.lower():
            raise ValueError()
    except (ValueError, AttributeError, TypeError):
        raise AgentError("Instance ID must be a UUID") from None
    if isinstance(generation, bool) or not isinstance(generation, int) or generation < 1:
        raise AgentError("Invalid instance generation")
    return instance_id.lower(), generation


def scoped(root, value, allow_root=False):
    """Reject traversal AND existing symlinks, including dangling symlinks."""
    if not isinstance(value, str) or "\\" in value or "\x00" in value:
        raise AgentError("Invalid managed path")
    rel = PurePosixPath(value)
    if rel.is_absolute() or any(p in ("..", "") for p in rel.parts):
        raise AgentError("Path is outside the managed directory")
    if value not in ("", ".") and str(rel) != value:
        raise AgentError("Managed paths must use their canonical relative form")
    if (not rel.parts or str(rel) == ".") and not allow_root:
        raise AgentError("A file path is required")
    root = Path(root)
    current = root
    if current.is_symlink():
        raise AgentError("Managed root must not be a symlink")
    for part in rel.parts:
        current = current / part
        if current.is_symlink():
            raise AgentError("Symlinks are not allowed in managed file operations")
    return current


def atomic_bytes(path, data, mode=0o600):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(path.name + "." + uuid.uuid4().hex + ".tmp")
    try:
        with open(tmp, "xb") as stream:
            os.chmod(tmp, mode)
            stream.write(data)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(tmp, path)
        fd = os.open(path.parent, os.O_RDONLY)
        try:
            os.fsync(fd)
        finally:
            os.close(fd)
    finally:
        with contextlib.suppress(FileNotFoundError):
            tmp.unlink()


def atomic_json(path, value):
    atomic_bytes(path, json.dumps(value, sort_keys=True).encode())


@contextlib.contextmanager
def managed_parent(root, value, create=False, uid=1000, gid=1000):
    """Use dirfds so an instance cannot race a symlink into a host-root write."""
    scoped(root, value)
    parts = PurePosixPath(value).parts
    flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
    fd = os.open(root, flags)
    try:
        for part in parts[:-1]:
            if create:
                with contextlib.suppress(FileExistsError):
                    os.mkdir(part, 0o755, dir_fd=fd)
                    if os.geteuid() == 0:
                        os.chown(part, uid, gid, dir_fd=fd, follow_symlinks=False)
            child = os.open(part, flags, dir_fd=fd)
            os.close(fd)
            fd = child
        yield fd, parts[-1]
    finally:
        os.close(fd)


@contextlib.contextmanager
def managed_open(root, value):
    with managed_parent(root, value) as (parent, name):
        fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent)
        try:
            if not stat_module.S_ISREG(os.fstat(fd).st_mode):
                raise AgentError("Requested managed path is not a regular file")
            with os.fdopen(fd, "rb", closefd=False) as stream:
                yield stream
        finally:
            os.close(fd)


def managed_read(root, value):
    try:
        with managed_open(root, value) as stream:
            if os.fstat(stream.fileno()).st_size > MAX_FILE:
                raise Rejected("Managed file exceeds the 32 MiB read limit")
            content = stream.read(MAX_FILE + 1)
            if len(content) > MAX_FILE:
                raise Rejected("Managed file exceeds the 32 MiB read limit")
            return content
    except FileNotFoundError:
        return None


def managed_manifest(root, payload, stopping):
    """Hash a bounded managed tree without following runtime-controlled symlinks."""
    deadline = time.monotonic() + MANIFEST_MAX_SECONDS

    def check_budget():
        if stopping.is_set() or time.monotonic() >= deadline:
            raise AgentError("Manifest scan exceeded its time limit or was interrupted")

    roots = payload.get("roots")
    ignored = payload.get("ignoreNames")
    paths = payload.get("paths", [])
    if not isinstance(roots, list) or not roots or len(roots) > 64:
        raise AgentError("Manifest roots must contain between 1 and 64 managed directories")
    if not isinstance(ignored, list) or len(ignored) > 256 or any(
        not isinstance(name, str) or not name or name in (".", "..") or any(char in name for char in ("/", "\\", "\x00"))
        for name in ignored
    ):
        raise AgentError("Manifest ignored names must be plain file or directory names")
    if not isinstance(paths, list) or len(paths) > 10000:
        raise AgentError("Manifest managed paths must contain at most 10000 files")
    for value in roots:
        check_budget()
        scoped(root, value)
    roots = sorted(set(roots))
    roots = [value for index, value in enumerate(roots) if not any(value.startswith(parent + "/") for parent in roots[:index])]
    ignored = set(ignored)
    flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
    entries = 0
    result_bytes = len(json.dumps({"files": []}).encode())
    files = []

    def validate_paths():
        for value in paths:
            check_budget()
            try:
                with managed_open(root, value):
                    pass
            except FileNotFoundError:
                pass

    def identity(info):
        return info.st_dev, info.st_ino, info.st_mode, info.st_size, info.st_mtime_ns, info.st_ctime_ns

    def open_directory(app, relative):
        directory = os.dup(app)
        try:
            for part in PurePosixPath(relative).parts:
                nested = os.open(part, flags, dir_fd=directory)
                os.close(directory)
                directory = nested
            return directory
        except BaseException:
            os.close(directory)
            raise

    def scan(directory, relative, depth):
        nonlocal entries, result_bytes
        check_budget()
        if depth > MANIFEST_MAX_DEPTH:
            raise AgentError("Manifest scan exceeded its directory depth limit")
        before = os.fstat(directory)
        with os.scandir(directory) as children:
            for child in children:
                check_budget()
                entries += 1
                if entries > MANIFEST_MAX_ENTRIES:
                    raise AgentError("Manifest scan exceeded its entry limit")
                if child.name in ignored:
                    continue
                info = os.stat(child.name, dir_fd=directory, follow_symlinks=False)
                if stat_module.S_ISLNK(info.st_mode):
                    continue
                path = relative + "/" + child.name
                if stat_module.S_ISDIR(info.st_mode):
                    nested = os.open(child.name, flags, dir_fd=directory)
                    try:
                        if identity(info) != identity(os.fstat(nested)):
                            raise AgentError("Manifest directory changed during the scan")
                        scan(nested, path, depth + 1)
                        if identity(os.fstat(nested)) != identity(os.stat(child.name, dir_fd=directory, follow_symlinks=False)):
                            raise AgentError("Manifest directory changed during the scan")
                    finally:
                        os.close(nested)
                elif stat_module.S_ISREG(info.st_mode):
                    descriptor = os.open(child.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory)
                    try:
                        if identity(info) != identity(os.fstat(descriptor)):
                            raise AgentError("Manifest file changed during the scan")
                        sha = hashlib.sha256()
                        size = 0
                        with os.fdopen(descriptor, "rb", closefd=False) as stream:
                            while True:
                                check_budget()
                                block = stream.read(1024 * 1024)
                                if not block:
                                    break
                                sha.update(block)
                                size += len(block)
                        if size != info.st_size or identity(info) != identity(os.fstat(descriptor)) or identity(info) != identity(os.stat(child.name, dir_fd=directory, follow_symlinks=False)):
                            raise AgentError("Manifest file changed during the scan")
                    finally:
                        os.close(descriptor)
                    item = {"path": path, "size": size, "sha256": sha.hexdigest()}
                    result_bytes += len(json.dumps(item).encode()) + (2 if files else 0)
                    if result_bytes > MANIFEST_MAX_BYTES:
                        raise AgentError("Manifest scan exceeded its result size limit")
                    files.append(item)
                else:
                    raise AgentError("Manifest scan encountered a non-regular managed file")
        if identity(before) != identity(os.fstat(directory)):
            raise AgentError("Manifest directory changed during the scan")

    try:
        app = os.open(root, flags)
        try:
            app_identity = os.fstat(app)
            validate_paths()
            for relative in roots:
                check_budget()
                try:
                    directory = open_directory(app, relative)
                except FileNotFoundError:
                    continue  # A destination directory need not exist yet.
                try:
                    scan(directory, relative, len(PurePosixPath(relative).parts))
                    current_directory = open_directory(app, relative)
                    try:
                        if identity(os.fstat(directory)) != identity(os.fstat(current_directory)):
                            raise AgentError("Manifest directory changed during the scan")
                    finally:
                        os.close(current_directory)
                finally:
                    os.close(directory)
            validate_paths()
            current = os.stat(root, follow_symlinks=False)
            if (current.st_dev, current.st_ino, current.st_mode) != (app_identity.st_dev, app_identity.st_ino, app_identity.st_mode):
                raise AgentError("Manifest managed root changed during the scan")
            check_budget()
        finally:
            os.close(app)
    except OSError:
        raise AgentError("Manifest scan could not read the complete managed tree") from None
    return {"files": sorted(files, key=lambda item: item["path"])}


def managed_write(root, value, content, uid=1000, gid=1000):
    if content is None:
        try:
            with managed_parent(root, value) as (parent, name):
                with contextlib.suppress(FileNotFoundError):
                    os.unlink(name, dir_fd=parent)
                os.fsync(parent)
        except FileNotFoundError:
            pass
        return
    with managed_parent(root, value, create=True, uid=uid, gid=gid) as (parent, name):
        prior = None
        try:
            prior = os.stat(name, dir_fd=parent, follow_symlinks=False)
            if not stat_module.S_ISREG(prior.st_mode):
                raise Rejected("Managed target is not a regular file")
        except FileNotFoundError:
            pass
        temp_name = ".ec-" + uuid.uuid4().hex
        fd = os.open(temp_name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=parent)
        try:
            os.fchmod(fd, prior.st_mode & 0o777 if prior else 0o644)
            if os.geteuid() == 0:
                os.fchown(fd, prior.st_uid if prior else uid, prior.st_gid if prior else gid)
            with os.fdopen(fd, "wb", closefd=False) as stream:
                stream.write(content)
                stream.flush()
                os.fsync(fd)
            os.replace(temp_name, name, src_dir_fd=parent, dst_dir_fd=parent)
            os.fsync(parent)
        finally:
            os.close(fd)
            with contextlib.suppress(FileNotFoundError):
                os.unlink(temp_name, dir_fd=parent)


def managed_copy(root, value, source, uid=1000, gid=1000):
    """Copy a private file, using Linux reflinks when supported, never hardlinks."""
    with managed_open(Path(source).parent, Path(source).name) as incoming:
        with managed_parent(root, value, create=True, uid=uid, gid=gid) as (parent, name):
            temp_name = ".ec-" + uuid.uuid4().hex
            fd = os.open(temp_name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=parent)
            try:
                os.fchmod(fd, 0o644)
                if os.geteuid() == 0:
                    os.fchown(fd, uid, gid)
                try:
                    if not sys.platform.startswith("linux"):
                        raise OSError("Reflinks unavailable")
                    fcntl.ioctl(fd, 0x40049409, incoming.fileno())  # FICLONE
                except OSError:
                    os.ftruncate(fd, 0)
                    with os.fdopen(fd, "wb", closefd=False) as outgoing:
                        shutil.copyfileobj(incoming, outgoing, 1024 * 1024)
                        outgoing.flush()
                os.fsync(fd)
                os.replace(temp_name, name, src_dir_fd=parent, dst_dir_fd=parent)
                os.fsync(parent)
            finally:
                os.close(fd)
                with contextlib.suppress(FileNotFoundError):
                    os.unlink(temp_name, dir_fd=parent)


class ReadWriteGate:
    """Let observations run together; global mutations wait for existing readers."""
    def __init__(self):
        self.condition = threading.Condition()
        self.readers = 0
        self.writers = 0

    @contextlib.contextmanager
    def read(self):
        with self.condition:
            while self.writers:
                self.condition.wait()
            self.readers += 1
        try:
            yield
        finally:
            with self.condition:
                self.readers -= 1
                self.condition.notify_all()

    @contextlib.contextmanager
    def write(self):
        # The mutation lane is serial, so there is only one writer.
        with self.condition:
            self.writers += 1
            while self.readers:
                self.condition.wait()
        try:
            yield
        finally:
            with self.condition:
                self.writers -= 1
                self.condition.notify_all()


def read_operation(operation):
    if not isinstance(operation, dict):
        return False
    if operation.get("kind") in ("host.inspect", "instance.inspect", "rcon.status-batch", "files.read", "files.list", "files.stat", "files.manifest", "logs.read"):
        return True
    payload = operation.get("payload")
    return operation.get("kind") == "rcon.execute" and isinstance(payload, dict) and payload.get("command") in ("status", "getstatus", "version")


def valid_observation_keep_days(value):
    return type(value) is int and 0 <= value <= 36500


class Journal:
    def __init__(self, path, recover=True):
        self.db = sqlite3.connect(path)
        self.db.row_factory = sqlite3.Row
        if not recover:
            return
        self.db.execute("PRAGMA journal_mode=WAL")
        self.db.execute("PRAGMA synchronous=FULL")
        self.db.execute("CREATE TABLE IF NOT EXISTS operations (id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, request TEXT NOT NULL, state TEXT NOT NULL, result TEXT, error TEXT, updated REAL NOT NULL)")
        columns = {row[1] for row in self.db.execute("PRAGMA table_info(operations)")}
        if "checkpoint" not in columns:
            self.db.execute("ALTER TABLE operations ADD COLUMN checkpoint TEXT")
        if "reported_at" not in columns:
            self.db.execute("ALTER TABLE operations ADD COLUMN reported_at REAL")
        if "retention_eligible" not in columns:
            self.db.execute("ALTER TABLE operations ADD COLUMN retention_eligible INTEGER NOT NULL DEFAULT 0")
        self.db.execute("CREATE INDEX IF NOT EXISTS operations_observation_retention ON operations(updated,id) WHERE state='completed' AND reported_at IS NOT NULL AND retention_eligible=1")
        self.db.execute("UPDATE operations SET state='unknown', error='Agent restarted during execution; inspect before reconciliation', updated=? WHERE state='running'", (time.time(),))
        self.db.commit()

    def accept(self, request):
        op_id = identifier(request.get("id"))
        immutable = {key: request.get(key) for key in ("id", "kind", "instanceId", "generation", "payload")}
        canonical = json.dumps(immutable, sort_keys=True, separators=(",", ":"))
        fingerprint = digest(canonical.encode())
        row = self.db.execute("SELECT * FROM operations WHERE id=?", (op_id,)).fetchone()
        if row:
            if row["fingerprint"] != fingerprint:
                raise AgentError("Operation ID was reused with different content")
            return dict(row), False
        self.db.execute("INSERT INTO operations(id,fingerprint,request,state,updated) VALUES(?,?,?,'accepted',?)", (op_id, fingerprint, canonical, time.time()))
        self.db.commit()
        return self.get(op_id), True

    def get(self, op_id):
        row = self.db.execute("SELECT * FROM operations WHERE id=?", (op_id,)).fetchone()
        if row is None:
            raise AgentError("Unknown operation")
        return dict(row)

    def update(self, op_id, state, result=None, error=None):
        self.db.execute("UPDATE operations SET state=?, result=?, error=?, updated=?, reported_at=NULL, retention_eligible=0 WHERE id=?", (state, json.dumps(result) if result is not None else None, error, time.time(), op_id))
        self.db.commit()
        return self.get(op_id)

    def confirm_completed(self, row):
        """Only an acknowledged, unchanged completion can become disposable."""
        if row["state"] != "completed":
            return
        eligible = False
        try:
            result = json.loads(row["result"]) if row.get("result") is not None else None
            eligible = read_operation(json.loads(row["request"])) and not row.get("error") and not (
                isinstance(result, dict) and (result.get("ok") is False or result.get("healthy") is False or result.get("error") not in (None, ""))
            )
        except (ValueError, TypeError, KeyError):
            pass  # Malformed historical records are retained for inspection.
        self.db.execute("""UPDATE operations SET reported_at=COALESCE(reported_at,?), retention_eligible=?
            WHERE id=? AND state='completed' AND updated=? AND result IS ? AND error IS ?""",
            (time.time(), int(bool(eligible)), row["id"], row["updated"], row.get("result"), row.get("error")))
        self.db.commit()

    def prune_observations(self, keep_days):
        if not valid_observation_keep_days(keep_days) or keep_days == 0:
            return 0
        cutoff = time.time() - keep_days * 86400
        removed = self.db.execute("""DELETE FROM operations WHERE id IN (
            SELECT id FROM operations WHERE state='completed' AND reported_at IS NOT NULL AND retention_eligible=1
                AND updated<? ORDER BY updated,id LIMIT 500)""", (cutoff,)).rowcount
        self.db.commit()
        return removed

    def close(self):
        self.db.close()

    def checkpoint(self, op_id, value):
        self.db.execute("UPDATE operations SET checkpoint=?, updated=? WHERE id=?", (json.dumps(value), time.time(), op_id))
        self.db.commit()


class Docker:
    def call(self, *args, timeout=120, tolerate_missing=False, merge_stderr=False):
        try:
            result = subprocess.run(["docker", *args], stdout=subprocess.PIPE, stderr=subprocess.STDOUT if merge_stderr else subprocess.PIPE, timeout=timeout, check=False)
        except (OSError, subprocess.TimeoutExpired):
            raise AgentError("Docker command unavailable or timed out") from None
        if result.returncode:
            # Never include arbitrary Docker stderr: create errors may echo secrets.
            if tolerate_missing and any(message in (result.stderr or b"") for message in (b"No such object", b"No such container", b"No such image")):
                return None
            raise AgentError("Docker rejected the requested operation")
        return result.stdout.decode("utf-8", "replace")

    def inspect(self, name):
        raw = self.call("container", "inspect", name, tolerate_missing=True)
        return json.loads(raw)[0] if raw is not None else None


class Uninstaller:
    """A separate systemd service owns cleanup after stopping the host agent.

    Its root-only request and executable stay on disk until the controller
    acknowledges data removal. Completion is reported only after removing those
    final files. A lost final response can be retried from memory; a power loss
    in that last window deliberately leaves controller status unconfirmed.
    """

    def __init__(self, request_path, docker=None):
        if Path(request_path) != UNINSTALL_REQUEST:
            raise AgentError("Invalid host uninstall request location")
        protected_file(UNINSTALL_REQUEST)
        self.job = json.loads(UNINSTALL_REQUEST.read_text())
        self.docker = docker or Docker()
        self.root = uninstall_root(self.job.get("root"), self.job.get("rootIdentity"))
        host_id = identifier(self.job.get("hostId"))
        operation_id = identifier(self.job.get("operationId"))
        parsed = urllib.parse.urlsplit(self.job.get("controller", ""))
        if parsed.scheme != "https" or not parsed.netloc or parsed.username or parsed.password or parsed.query or parsed.fragment:
            raise AgentError("Host uninstall controller must use HTTPS")
        if self.job.get("receiptUrl") != "/api/agent/decommission/" + operation_id or not isinstance(self.job.get("receiptToken"), str) or not self.job["receiptToken"]:
            raise AgentError("Invalid host uninstall receipt")
        if not isinstance(self.job.get("rootIdentity"), list) or len(self.job["rootIdentity"]) != 2 or not all(isinstance(value, int) and not isinstance(value, bool) for value in self.job["rootIdentity"]):
            raise AgentError("Invalid host data ownership proof")
        if self.root.exists():
            # The authorized device/inode is persisted before deletion so a
            # crash halfway through rmtree does not require a deleted marker.
            if not self.job.get("removalStarted"):
                uninstall_ownership({"root": str(self.root), "controller": self.job["controller"], "hostId": host_id})

    def save(self, **changes):
        self.job.update(changes)
        atomic_json(UNINSTALL_REQUEST, self.job)

    def receipt(self, stage, error=None):
        payload = {"stage": stage}
        if error:
            payload["error"] = error
        request = urllib.request.Request(self.job["controller"].rstrip("/") + self.job["receiptUrl"], data=json.dumps(payload).encode(), headers={"Content-Type": "application/json", "Authorization": "Bearer " + self.job["receiptToken"], "User-Agent": "electric-cherry-agent/" + VERSION}, method="POST")
        class NoRedirect(urllib.request.HTTPRedirectHandler):
            def redirect_request(self, req, fp, code, msg, hdrs, newurl):
                raise AgentError("Controller redirects are not allowed")
        try:
            with urllib.request.build_opener(NoRedirect()).open(request, timeout=35) as response:
                if json.load(response).get("acknowledged") is not True:
                    raise AgentError("Controller has not acknowledged host removal")
        except (OSError, ValueError):
            raise AgentError("Host removal receipt is waiting for controller contact") from None

    def containers(self):
        ids = self.docker.call("container", "ls", "--all", "--quiet", "--no-trunc").splitlines()
        values = []
        for container_id in ids:
            info = self.docker.inspect(container_id)
            if info:
                values.append(info)
        return values

    def owned_container(self, info):
        labels = info.get("Config", {}).get("Labels", {}) or {}
        if labels.get(LABEL + ".managed") != "true":
            return False
        if labels.get(LABEL + ".host") not in (None, self.job["hostId"]):
            return False
        try:
            instance_id, generation = instance_identity(labels.get(LABEL + ".instance"), int(labels.get(LABEL + ".generation", "")))
        except (AgentError, ValueError, TypeError):
            return False
        expected = str(self.root / "instances" / instance_id / str(generation) / "app")
        # Legacy agents did not label the host; its exact managed bind mount
        # proves ownership without touching another installation's containers.
        return any(mount.get("Type") == "bind" and mount.get("Source") == expected and mount.get("Destination") == "/home/plutainer/app" for mount in info.get("Mounts", []))

    def uses_root(self, info):
        for mount in info.get("Mounts", []):
            if mount.get("Type") != "bind":
                continue
            source = Path(mount.get("Source", "/")).resolve()
            if source == self.root or self.root in source.parents:
                return True
        return False

    def remove_docker_resources(self):
        containers = self.containers()
        if any(self.uses_root(info) and not self.owned_container(info) for info in containers):
            raise AgentError("An unrelated container uses the host data directory; detach it before retrying deletion")
        owned = [info for info in containers if self.owned_container(info)]
        if any(info.get("State", {}).get("Running") is not False or info.get("State", {}).get("Status") not in ("created", "exited", "dead") for info in owned):
            raise AgentError("A managed game container is still running or its state is unknown; retire it before deleting this host")
        for info in owned:
            # Do not force removal: Docker must also reject a container that
            # starts again after inspection rather than interrupting players.
            self.docker.call("container", "rm", "--volumes", info["Id"], timeout=120, tolerate_missing=True)
        remaining = self.containers()
        if any(self.uses_root(info) for info in remaining):
            raise AgentError("Containers still use the host data directory")
        # Never prune Docker globally. Only remove this agent's recorded image
        # references, and preserve images used by any unrelated container.
        used_images = {info.get("Image") for info in remaining}
        images = self.job.get("runtimeImages", {})
        if not isinstance(images, dict):
            raise AgentError("Managed runtime image ownership is unavailable")
        for reference, expected_id in images.items():
            if not isinstance(reference, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/:@-]{0,511}", reference) or not isinstance(expected_id, str):
                raise AgentError("Managed runtime image ownership is invalid")
            raw = self.docker.call("image", "inspect", reference, tolerate_missing=True)
            if raw is None:
                continue
            found = json.loads(raw)
            if not found or found[0].get("Id") != expected_id or expected_id in used_images:
                continue
            self.docker.call("image", "rm", reference, tolerate_missing=True)

    def remove_data(self):
        protected_path(AGENT_CONFIG.parent, missing=True)
        # Prevent a scheduled update from recreating agent files during removal.
        # Stopping the worker is safe here: the installation and any backups are
        # being removed together, and the durable decommission request blocks
        # future update attempts.
        if not self.job.get("updaterStopped"):
            for unit in (UPDATE_TIMER, UPDATE_UNIT):
                protected_file(unit, missing=True)
                if unit.exists():
                    if unit == UPDATE_TIMER:
                        systemctl("disable", "--now", unit.name)
                    else:
                        systemctl("stop", unit.name)
            self.save(updaterStopped=True)
        if not self.job.get("agentStopped"):
            systemctl("disable", "--now", AGENT_UNIT.name)
            self.save(agentStopped=True)
        self.remove_docker_resources()
        root = uninstall_root(str(self.root), self.job["rootIdentity"])
        # This checkpoint also authorizes safe continuation if a power loss
        # removes the ownership marker before the rest of the tree.
        self.save(removalStarted=True)
        if root.exists():
            if not shutil.rmtree.avoids_symlink_attacks:
                raise AgentError("This platform cannot safely remove managed host files")
            shutil.rmtree(root)
            sync_parent(root)
        for path in (AGENT_CONFIG, AGENT_EXECUTABLE, AGENT_UNIT, UPDATE_EXECUTABLE, UPDATE_UNIT, UPDATE_TIMER, AGENT_EXECUTABLE.with_suffix(".py.update"), UNINSTALL_EXECUTABLE.with_suffix(".py.update")):
            protected_file(path, missing=True)
            path.unlink(missing_ok=True)
            sync_parent(path)
        systemctl("daemon-reload")
        self.save(dataRemoved=True)

    def remove_helper(self):
        # Disable does not stop this process: it must outlive its own unit and
        # executable to deliver the final receipt after the files are gone.
        protected_file(UNINSTALL_UNIT, missing=True)
        if UNINSTALL_UNIT.exists():
            systemctl("disable", UNINSTALL_UNIT.name)
            UNINSTALL_UNIT.unlink()
        sync_parent(UNINSTALL_UNIT)
        systemctl("daemon-reload")
        protected_file(UNINSTALL_EXECUTABLE, missing=True)
        UNINSTALL_EXECUTABLE.unlink(missing_ok=True)
        sync_parent(UNINSTALL_EXECUTABLE)
        protected_path(UNINSTALL_REQUEST.parent, missing=True)
        if UNINSTALL_REQUEST.parent.exists():
            shutil.rmtree(UNINSTALL_REQUEST.parent)
        sync_parent(UNINSTALL_REQUEST.parent)

    def run(self):
        if not self.job.get("dataRemoved"):
            try:
                self.remove_data()
            except (AgentError, OSError, ValueError, KeyError) as error:
                with contextlib.suppress(AgentError):
                    self.receipt("failed", str(error) if isinstance(error, AgentError) else "Host cleanup could not finish. Check the uninstall service and resolve filesystem errors; cleanup will retry.")
                raise
        self.receipt("cleaned")
        while True:
            try:
                self.remove_helper()
                break
            except (AgentError, OSError, ValueError):
                # Once its unit/executable has been unlinked systemd can no
                # longer restart this worker. Finish idempotently in memory.
                with contextlib.suppress(AgentError):
                    self.receipt("failed", "Host files were removed, but removing the uninstall service is still retrying")
                time.sleep(15)
        # The controller never marks a host deleted on the earlier receipt.
        # No credentials or retry state are left on disk at this point.
        while True:
            try:
                self.receipt("completed")
                return
            except AgentError:
                time.sleep(15)


class LibraryTransfer:
    """Durable phase progress; throughput counts only work in this attempt."""

    def __init__(self, progress, marker, previous=None):
        self.progress = progress
        self.marker = marker
        self.lock = threading.RLock()
        prior = (previous or {}).get("transfer") or {}
        self.prior_phase = prior.get("phase")
        self.prior_phase_elapsed = prior.get("phaseElapsedSeconds", 0)
        self.started_at = prior.get("startedAt") or time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        self.elapsed = prior.get("elapsedSeconds", 0)
        self.phase_elapsed = prior.get("phaseElapsedSeconds", 0) if prior.get("phase") == progress["status"] else 0
        self.started = self.phase_started = time.monotonic()
        self.phase = progress["status"]
        self.work = 0
        self.finished = False
        self.failed_bytes_downloaded = None
        self.persisted_at = self.started
        self.persist(force=True)

    def snapshot(self):
        with self.lock:
            if not self.finished:
                now = time.monotonic()
                sample_seconds = max(0, now - self.phase_started)
                rate = self.work / sample_seconds if sample_seconds >= 1 and self.work else None
                done = self.progress["bytesDownloaded"] if self.phase == "downloading" else self.progress["bytesVerified"]
                self.progress["transfer"] = {
                    "phase": self.phase, "startedAt": self.started_at,
                    "elapsedSeconds": self.elapsed + max(0, now - self.started),
                    "phaseElapsedSeconds": self.phase_elapsed + sample_seconds,
                    "bytesPerSecond": rate,
                    "etaSeconds": max(0, self.progress["bytesTotal"] - done) / rate if rate else None,
                }
            return {**self.progress, "transfer": self.progress["transfer"].copy()}

    def persist(self, force=False):
        with self.lock:
            now = time.monotonic()
            self.snapshot()
            if force or now - self.persisted_at >= 5:
                atomic_json(self.marker, self.progress)
                self.persisted_at = now

    def advance(self, key, count, work):
        with self.lock:
            self.progress[key] += count
            self.work += work
            self.persist()

    def begin_verification(self):
        with self.lock:
            self.progress.update(status="verifying", bytesVerified=0)
            if self.phase != "verifying":
                self.phase_elapsed = self.prior_phase_elapsed if self.prior_phase == "verifying" else 0
            self.phase = "verifying"
            self.phase_started = time.monotonic()
            self.work = 0
            self.persist(force=True)

    def finish(self, status, error=None):
        with self.lock:
            self.snapshot()
            self.finished = True
            self.progress["status"] = status
            if status == "failed" and self.failed_bytes_downloaded is not None:
                self.progress["bytesDownloaded"] = self.failed_bytes_downloaded
            self.progress["transfer"]["etaSeconds"] = 0 if status == "ready" else None
            if error:
                self.progress["error"] = error
            self.persist(force=True)


class Agent:
    def __init__(self, config, docker=None):
        self.config = config
        self.root = Path(config.get("root", "/var/lib/electric-cherry")).resolve()
        for directory in ("agent", "instances", "libraries"):
            (self.root / directory).mkdir(parents=True, exist_ok=True)
        os.chmod(self.root / "agent", 0o700)
        self.journal = Journal(self.root / "agent" / "operations.sqlite3")
        self.docker = docker or Docker()
        self.stopping = threading.Event()
        self.current_operation = None
        self.cpu_sample_lock = threading.Lock()
        self.cpu_sample = None
        self.instance_locks = {}
        self.read_gate = ReadWriteGate()
        self.read_claim_lock = threading.Lock()
        self.read_claims = set()
        self.health_probe_gate = threading.BoundedSemaphore(2)
        self.cancelled_creations = frozenset()
        self.defer_plutonium_preparation = False
        self.observation_keep_days = 0  # Older controllers do not authorize cleanup.
        self.library_progress = {}
        self.library_transfers = {}
        self.inventory_manifest_cache = {}
        self.inventory_manifest_lock = threading.Lock()
        self.plutonium_lock = threading.RLock()
        self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
        self.executing_id = None
        self.operation_phase = None
        self.lane_lock = threading.Lock()
        self.lanes = {name: {"thread": None, "lastPollAt": None, "polled": time.monotonic(), "busy": False}
                      for name in ("health-1", "health-2", "read", "write")}

    def lane_started(self, name):
        with self.lane_lock:
            self.lanes[name]["thread"] = threading.current_thread()

    def lane_polled(self, name):
        with self.lane_lock:
            self.lanes[name].update(lastPollAt=datetime.now(timezone.utc).isoformat(timespec="milliseconds"), polled=time.monotonic())

    def lane_busy(self, name, busy):
        with self.lane_lock:
            self.lanes[name]["busy"] = busy

    def lane_inventory(self):
        with self.lane_lock:
            return [{"name": name, "alive": lane["thread"] is None or lane["thread"].is_alive(),
                     "busy": lane["busy"], "lastPollAt": lane["lastPollAt"],
                     "pollAgeSeconds": max(0, time.monotonic() - lane["polled"])}
                    for name, lane in self.lanes.items()]

    def can_uninstall(self):
        try:
            if Path(self.config_path) != AGENT_CONFIG:
                return False
            uninstall_ownership(self.config)
            protected_file(UNINSTALL_EXECUTABLE)
            protected_file(UNINSTALL_UNIT)
            return True
        except (AgentError, OSError, ValueError):
            return False

    def can_update(self):
        try:
            if Path(self.config_path) != AGENT_CONFIG:
                return False
            uninstall_ownership(self.config)
            for path in (UPDATE_EXECUTABLE, UPDATE_UNIT, UPDATE_TIMER):
                protected_file(path)
            protected_path(UPDATE_TIMER_ENABLED.parent)
            return UPDATE_TIMER_ENABLED.is_symlink() and UPDATE_TIMER_ENABLED.resolve() == UPDATE_TIMER
        except (AgentError, OSError, ValueError):
            return False

    def update_inventory(self):
        result = {"enabled": self.can_update(), "state": "current", "currentVersion": VERSION}
        if Path(self.config_path) != AGENT_CONFIG:
            return result
        try:
            protected_file(UPDATE_STATUS, missing=True)
            if not UPDATE_STATUS.exists():
                return result
            if UPDATE_STATUS.stat().st_size > 16384:
                raise ValueError("Invalid update status")
            saved = json.loads(UPDATE_STATUS.read_text())
            if saved.get("state") in ("current", "checking", "deferred", "updating", "updated", "failed"):
                result["state"] = saved["state"]
            for key in ("targetVersion", "error"):
                if isinstance(saved.get(key), str):
                    result[key] = saved[key][:2000 if key == "error" else 80]
            checked_at = saved.get("checkedAt")
            if isinstance(checked_at, (int, float)) and not isinstance(checked_at, bool):
                result["checkedAt"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(checked_at))
            elif isinstance(checked_at, str):
                result["checkedAt"] = checked_at[:80]
        except (AgentError, OSError, ValueError, TypeError, AttributeError, OverflowError):
            result.update(state="failed", error="Agent update status is unavailable; inspect the update service.")
        return result

    def update_pending(self):
        # A persisted transaction survives crashes and is cleared only by the
        # independent worker after health confirmation or a completed rollback.
        return Path(self.config_path) == AGENT_CONFIG and (UPDATE_JOURNAL.exists() or UPDATE_JOURNAL.is_symlink())

    @contextlib.contextmanager
    def activity(self, read_only=False):
        path = self.root / "agent" / "activity.lock"
        descriptor = os.open(path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
        mutation_descriptor = None
        try:
            try:
                # All work excludes the updater. Mutations additionally exclude
                # other mutation batches while read lanes may continue.
                fcntl.flock(descriptor, fcntl.LOCK_SH | fcntl.LOCK_NB)
                if not read_only:
                    mutation_descriptor = os.open(self.root / "agent" / "mutation.lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
                    fcntl.flock(mutation_descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                yield False
                return
            yield True
        finally:
            if mutation_descriptor is not None:
                os.close(mutation_descriptor)
            os.close(descriptor)

    def record_health(self):
        if Path(self.config_path) != AGENT_CONFIG or not self.can_update():
            return
        value = {"version": VERSION, "pid": os.getpid(), "checkedAt": time.time()}
        protected_file(AGENT_HEALTH, missing=True)
        if self.update_pending():
            protected_file(UPDATE_JOURNAL)
            journal = json.loads(UPDATE_JOURNAL.read_text())
            if isinstance(journal.get("id"), str):
                value["updateId"] = journal["id"]
        atomic_json(AGENT_HEALTH, value)

    def decommission(self, operation):
        payload = operation.get("payload", {})
        if payload.get("hostId") != self.config.get("hostId") or payload.get("operationId") != operation["id"]:
            raise Rejected("Host removal identity does not match this agent")
        if payload.get("receiptUrl") != "/api/agent/decommission/" + identifier(operation["id"]) or not isinstance(payload.get("receiptToken"), str) or not payload["receiptToken"]:
            raise Rejected("Invalid host removal receipt")
        if not self.can_uninstall():
            raise Rejected("Reinstall the agent to enable safe removal of its dedicated host files")
        root = uninstall_ownership(self.config)
        info = root.stat()
        protected_file(UNINSTALL_REQUEST, missing=True)
        aliases_path = root / "agent" / "runtime-images.json"
        protected_file(aliases_path, missing=True)
        aliases = json.loads(aliases_path.read_text()) if aliases_path.exists() else {}
        job = {"hostId": payload["hostId"], "operationId": operation["id"], "receiptUrl": payload["receiptUrl"], "receiptToken": payload["receiptToken"], "controller": self.config["controller"].rstrip("/"), "root": str(root), "rootIdentity": [info.st_dev, info.st_ino], "runtimeImages": aliases}
        if UNINSTALL_REQUEST.exists():
            previous = json.loads(UNINSTALL_REQUEST.read_text())
            if any(previous.get(key) != job[key] for key in ("hostId", "operationId", "receiptUrl", "receiptToken", "controller", "root", "rootIdentity")):
                raise Rejected("A different host removal is already pending")
        self.checkpoint({"phase": "uninstall-service"})
        if not UNINSTALL_REQUEST.exists():
            atomic_json(UNINSTALL_REQUEST, job)
        systemctl("enable", "--now", "--no-block", UNINSTALL_UNIT.name)
        self.stopping.set()
        return {"decommissioning": True, "message": "The uninstall service will report completion after host files and the agent are removed"}

    def instance_dir(self, instance_id, generation):
        instance_id, generation = instance_identity(instance_id, generation)
        return self.root / "instances" / instance_id / str(generation)

    def container_name(self, instance_id, generation):
        instance_id, generation = instance_identity(instance_id, generation)
        return f"ec-{instance_id}-g{generation}"

    def inspect(self, instance_id, generation):
        name = self.container_name(instance_id, generation)
        info = self.docker.inspect(name)
        if info is None:
            return {"exists": False, "running": False, "healthy": False, "state": "missing", "instanceId": instance_id, "generation": generation}
        labels = info.get("Config", {}).get("Labels", {}) or {}
        if labels.get(LABEL + ".managed") != "true" or labels.get(LABEL + ".instance") != instance_id or labels.get(LABEL + ".generation") != str(generation):
            raise AgentError("Container ownership or generation does not match")
        state = info.get("State", {})
        status = state.get("Status", "unknown")
        healthy = state.get("Status") == "running" and state.get("Running") is True and state.get("Health", {}).get("Status", "healthy") == "healthy"
        if status == "running" and not healthy:
            status = state.get("Health", {}).get("Status", "unknown")
        return {"exists": True, "running": state.get("Running") is True, "healthy": healthy, "state": status, "containerId": info.get("Id"), "instanceId": instance_id, "generation": generation, "startedAt": state.get("StartedAt"), "creationOperation": labels.get(LABEL + ".operation"), "specSha256": labels.get(LABEL + ".spec-sha256")}

    def checkpoint(self, value):
        if self.executing_id:
            self.journal.checkpoint(self.executing_id, value)

    def mark_removed(self, instance_id, generation):
        marker = self.instance_dir(instance_id, generation).parent / "current.json"
        if not marker.exists() or json.loads(marker.read_text())["generation"] <= generation:
            atomic_json(marker, {"generation": generation, "removed": True})

    def make_writable(self, path, directory=False):
        if os.geteuid() == 0:
            os.chown(path, int(self.config.get("runtimeUid", 1000)), int(self.config.get("runtimeGid", 1000)))
            os.chmod(path, 0o755 if directory else 0o644)
        elif directory:
            os.chmod(path, 0o777)

    def spec(self, instance_id, generation):
        path = self.instance_dir(instance_id, generation) / "spec.json"
        try:
            return json.loads(path.read_text())
        except (OSError, ValueError):
            raise AgentError("Managed instance metadata is unavailable") from None

    def request(self, method, path, payload=None, credential=True):
        url = self.config["controller"].rstrip("/") + path
        if urllib.parse.urlsplit(url).scheme != "https" and not self.config.get("allowInsecureLocalhost", False):
            raise AgentError("Controller URL must use HTTPS")
        if urllib.parse.urlsplit(url).scheme != "https" and urllib.parse.urlsplit(url).hostname not in ("127.0.0.1", "localhost", "::1"):
            raise AgentError("Insecure development transport is limited to loopback")
        headers = {"Content-Type": "application/json", "User-Agent": "electric-cherry-agent/" + VERSION}
        if credential:
            headers["Authorization"] = "Bearer " + self.config["credential"]
        data = None if payload is None else json.dumps(payload).encode()
        req = urllib.request.Request(url, data=data, headers=headers, method=method)
        try:
            class NoRedirect(urllib.request.HTTPRedirectHandler):
                def redirect_request(self, req, fp, code, msg, hdrs, newurl):
                    raise AgentError("Controller redirects are not allowed")
            with urllib.request.build_opener(NoRedirect()).open(req, timeout=35) as response:
                return json.load(response)
        except (OSError, ValueError, http.client.HTTPException):
            raise AgentError("Controller request failed") from None

    def cpu_usage_percent(self):
        """Host-wide busy time since the previous inventory sample, without waiting."""
        with self.cpu_sample_lock:
            previous = self.cpu_sample
            self.cpu_sample = None
            try:
                with Path("/proc/stat").open() as source:
                    fields = source.readline().split()
                if not fields or fields[0] != "cpu" or len(fields) < 9:
                    return None
                # Guest time is already included in user/nice. Idle and I/O
                # wait are not busy; steal counts as unavailable CPU capacity.
                counters = tuple(int(value) for value in fields[1:9])
                if any(value < 0 for value in counters):
                    return None
            except (OSError, ValueError):
                return None
            self.cpu_sample = counters
            if previous is None:
                return None
            deltas = [current - old for current, old in zip(counters, previous)]
            total = sum(deltas)
            # Counters can reset and Linux can revise I/O wait downwards.
            # Rebase rather than reporting a misleading utilization value.
            if total <= 0 or any(delta < 0 for delta in deltas):
                return None
            return round(100 * (total - deltas[3] - deltas[4]) / total, 1)

    def instance_resources(self, observations=None):
        """One bounded Docker stats batch; unavailable samples are never zero load."""
        ids = self.docker.call("container", "ls", "--all", "--filter", "label=" + LABEL + ".managed=true",
                               "--no-trunc", "--format", "{{.ID}}", timeout=10).splitlines()[:1000]
        if not ids:
            return []
        observed_at = datetime.now(timezone.utc).isoformat(timespec="milliseconds")
        containers = json.loads(self.docker.call("container", "inspect", *ids, timeout=10))
        owned = {}
        for info in containers:
            try:
                labels = info.get("Config", {}).get("Labels", {}) or {}
                if labels.get(LABEL + ".managed") != "true":
                    continue
                instance_id, generation = instance_identity(labels.get(LABEL + ".instance"), int(labels.get(LABEL + ".generation", "")))
                name = self.container_name(instance_id, generation)
                if info.get("Name", "").lstrip("/") != name:
                    continue
                expected = str(self.instance_dir(instance_id, generation) / "app")
                if not any(mount.get("Type") == "bind" and mount.get("Source") == expected and mount.get("Destination") == "/home/plutainer/app" for mount in info.get("Mounts", [])):
                    continue
                state = info.get("State", {})
                healthy = state.get("Status") == "running" and state.get("Running") is True and state.get("Health", {}).get("Status", "healthy") == "healthy"
                status = state.get("Status", "unknown")
                if status == "running" and not healthy:
                    status = state.get("Health", {}).get("Status", "unknown")
                if observations is not None:
                    observations.append({"instanceId": instance_id, "generation": generation,
                                         "state": status, "healthy": healthy, "observedAt": observed_at})
                if not state.get("Running"):
                    continue
                started = datetime.fromisoformat(info["State"]["StartedAt"].replace("Z", "+00:00"))
                uptime = max(0, int((datetime.now(timezone.utc) - started).total_seconds()))
                owned[name] = {"instanceId": instance_id, "generation": generation, "uptimeSeconds": uptime}
            except (AgentError, ValueError, KeyError, TypeError):
                continue
        if not owned:
            return []
        rows = self.docker.call("stats", "--no-stream", "--no-trunc", "--format", "{{json .}}", *owned, timeout=15)
        samples = []
        units = {"B": 1, "KiB": 1024, "MiB": 1024 ** 2, "GiB": 1024 ** 3,
                 "TiB": 1024 ** 4, "kB": 1000, "MB": 1000 ** 2, "GB": 1000 ** 3}
        for line in rows.splitlines():
            try:
                row = json.loads(line)
                identity = owned.get(row.get("Name"))
                if not identity or int(row["PIDs"]) <= 0:
                    continue
                # Docker Linux CPU percent is core-based: 150% means 1.5 cores.
                percent = row["CPUPerc"]
                if not percent.endswith("%"):
                    continue
                cores = float(percent[:-1]) / 100
                used = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)\s*([A-Za-z]+)", row["MemUsage"].split("/", 1)[0].strip())
                if not used or used[2] not in units or not math.isfinite(cores) or cores < 0:
                    continue
                memory = math.ceil(float(used[1]) * units[used[2]])
                if memory > 0:
                    samples.append({**identity, "cpuCores": cores, "memoryBytes": memory})
            except (ValueError, KeyError, TypeError, AttributeError, OverflowError):
                continue
        return samples

    def inventory_manifest(self, marker):
        # Inventory only: launch/download verification must still read disk.
        # Include file identity so atomic replacements invalidate even when
        # their size and modification time match the previous manifest.
        with self.inventory_manifest_lock:
            try:
                metadata = marker.stat()
                signature = (metadata.st_mtime_ns, metadata.st_size, metadata.st_dev, metadata.st_ino)
                cached = self.inventory_manifest_cache.get(marker)
                if cached is not None and cached[0] == signature:
                    return cached[1]
                self.inventory_manifest_cache.pop(marker, None)
                value = json.loads(marker.read_text())
                self.inventory_manifest_cache[marker] = (signature, value)
                return value
            except (OSError, ValueError):
                self.inventory_manifest_cache.pop(marker, None)
                raise

    def inventory(self):
        cpu_usage = self.cpu_usage_percent()
        usage = shutil.disk_usage(self.root)
        memory = {}
        try:
            for line in Path("/proc/meminfo").read_text().splitlines():
                key, value = line.split(":", 1)
                memory[key] = int(value.strip().split()[0]) * 1024
        except (OSError, ValueError):
            pass
        docker_version, docker_running = "", False
        try:
            info = json.loads(self.docker.call("info", "--format", "{{json .}}", timeout=15))
            docker_version, docker_running = info.get("ServerVersion", ""), True
        except (AgentError, ValueError):
            pass
        libraries, caches = [], []
        progress = {key: value.copy() for key, value in list(self.library_progress.items())}
        progress.update({key: transfer.snapshot() for key, transfer in list(self.library_transfers.items())})
        manifests = list((self.root / "libraries").glob("*/*/manifest.json"))
        partial_manifests = list((self.root / "libraries").glob("*/.*.partial/source.json"))
        with self.inventory_manifest_lock:
            for marker in self.inventory_manifest_cache.keys() - set(manifests + partial_manifests):
                del self.inventory_manifest_cache[marker]
        for marker in manifests:
            # A crash between writing a manifest and its atomic promotion must
            # never expose the staging directory as a ready library.
            if marker.parent.name.startswith("."):
                continue
            try:
                value = self.inventory_manifest(marker)
                # Rechecking an existing cache must revoke readiness until its
                # entire contents have passed verification again.
                if value["libraryId"] not in progress and (marker.parent / "transfer.json").exists():
                    saved_progress = json.loads((marker.parent / "transfer.json").read_text())
                    if saved_progress.get("status") != "ready":
                        progress[value["libraryId"]] = {**saved_progress, "status": "failed" if saved_progress.get("status") == "failed" else "missing"}
                if value["libraryId"] in progress:
                    continue
                libraries.append({"id": value["libraryId"], "state": "ready"})
                cache = {"libraryVersionId": value["libraryId"], "status": "ready", "bytesDownloaded": value["size"], "bytesVerified": value["size"], "bytesTotal": value["size"], "manifestSha256": value.get("manifestSha256")}
                if value.get("transfer"):
                    cache["transfer"] = value["transfer"]
                caches.append(cache)
            except (OSError, ValueError, KeyError):
                continue
        ready = {item["id"] for item in libraries}
        for marker in partial_manifests:
            try:
                value = self.inventory_manifest(marker)
                library_id = identifier(value["libraryId"])
                if library_id in ready or library_id in progress:
                    continue
                expected = self.library_dir(value["game"], library_id).with_name("." + library_id + ".partial")
                if marker.parent != expected or marker.is_symlink() or marker.parent.is_symlink():
                    continue
                files = value["files"]
                canonical = json.dumps([{ "path": item["path"], "size": item["size"], "sha256": item["sha256"] } for item in files], separators=(",", ":"), ensure_ascii=False)
                if digest(canonical.encode("utf-8")) != value["manifestSha256"]:
                    continue
                retained, size, paths = 0, 0, set()
                for item in files:
                    if item["path"] in paths or type(item["size"]) is not int or item["size"] < 0:
                        raise AgentError("Invalid partial library manifest")
                    paths.add(item["path"])
                    target = scoped(marker.parent / "content", item["path"])
                    size += item["size"]
                    if target.exists() and stat_module.S_ISREG(target.stat().st_mode):
                        retained += min(target.stat().st_size, item["size"])
                # These are resumable bytes, not verified content. Reporting
                # them survives restarts without requiring space for two copies.
                cache = {"libraryVersionId": library_id, "status": "missing", "bytesDownloaded": retained, "bytesVerified": 0, "bytesTotal": size, "manifestSha256": value["manifestSha256"]}
                transfer_marker = marker.parent / "transfer.json"
                if transfer_marker.exists():
                    saved_progress = json.loads(transfer_marker.read_text())
                    if saved_progress.get("manifestSha256") == value["manifestSha256"]:
                        cache.update(bytesVerified=saved_progress.get("bytesVerified", 0), transfer=saved_progress.get("transfer"))
                        if saved_progress.get("status") == "failed":
                            cache.update(status="failed", error=saved_progress.get("error"))
                progress[library_id] = cache
            except (AgentError, OSError, ValueError, KeyError, TypeError):
                continue
        progress = list(progress.values())
        libraries.extend({"id": item["libraryVersionId"], "state": item["status"]} for item in progress)
        caches.extend(progress)
        images, prepared_images = [], []
        if docker_running:
            with contextlib.suppress(AgentError):
                lines = self.docker.call("image", "ls", "--no-trunc", "--digests", "--format", "{{.Repository}}|{{.Tag}}|{{.Digest}}|{{.ID}}", timeout=15).splitlines()
                image_ids = set()
                for line in lines:
                    parts = line.split("|")
                    if len(parts) != 4:
                        continue
                    repository, tag, image_digest, image_id = parts
                    image_ids.add(image_id)
                    if repository != "<none>" and tag != "<none>":
                        images.append(repository + ":" + tag)
                    if repository != "<none>" and image_digest != "<none>":
                        images.append(repository + "@" + image_digest)
                aliases = self.root / "agent" / "runtime-images.json"
                if aliases.exists():
                    with contextlib.suppress(OSError, ValueError):
                        saved = json.loads(aliases.read_text())
                        if isinstance(saved, dict):
                            # Only runtime.pull records administrator-prepared
                            # references. Other Docker cache entries are useful
                            # diagnostics, but do not grant launch permission.
                            prepared_images = [alias for alias, image_id in saved.items() if isinstance(image_id, str) and image_id in image_ids]
                            images.extend(prepared_images)
        resources = {}
        if docker_running:
            observations = []
            try:
                resources["instanceResources"] = self.instance_resources(observations)
            except (AgentError, ValueError, TypeError, KeyError):
                resources["instanceResourcesError"] = "Container resource sampling failed. Check Docker access and stats availability on this host."
            # A failed stats call does not discard a successful inspection.
            # Missing entries remain unknown; they are never inferred stopped.
            resources["instanceStates"] = observations
        os_name = platform.system()
        with contextlib.suppress(OSError):
            for line in Path("/etc/os-release").read_text().splitlines():
                if line.startswith("PRETTY_NAME="):
                    os_name = line.split("=", 1)[1].strip('"')
        update = self.update_inventory()
        capabilities = ["library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.unhealthy-recovery", "instance.rebuild-files", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs", "logs.read-offset", "bank.settlement"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
        plutonium_cache = self.plutonium_cache_inventory()
        return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "preparedRuntimeImages": sorted(set(prepared_images)), **({"plutoniumCache": plutonium_cache} if plutonium_cache else {}), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}

    def safe_to_disrupt(self, instance_id, generation, *, allow_unhealthy=False):
        health = self.inspect(instance_id, generation)
        if health["state"] in ("missing", "stopped", "exited", "created", "dead"):
            return
        # Provisioning recovery may treat explicitly unhealthy containers as
        # empty. Reinspect here so recovered containers still require occupancy.
        if allow_unhealthy and health["state"] == "unhealthy":
            return
        if not health["running"]:
            raise Held("Container state is unknown; disruption held")
        result = self.rcon(instance_id, generation, "status")
        if not result.get("ok") or result.get("players") is None:
            # Keep diagnostics bounded and do not include raw replies: they may
            # contain player addresses or echoed credentials.
            reason = {
                "unconfigured": "local RCON password is not configured",
                "unreachable": "local RCON endpoint is unreachable",
                "invalid-reply": "local RCON reply has an invalid packet prefix",
                "timeout": "local RCON status query timed out",
                "authentication": "local RCON authentication failed",
            }.get(result.get("reason"), "local RCON status reply has no recognized player count" if result.get("ok") else "local RCON query failed")
            raise Held(f"Fresh local player occupancy is unavailable: {reason}; disruption held")
        if result["players"] != 0:
            raise Held("Players are present; disruption held")

    def rcon_status_batch(self, operation):
        targets = operation.get("payload", {}).get("instances")
        if not isinstance(targets, list) or not 1 <= len(targets) <= 1000 or any(not isinstance(target, dict) for target in targets):
            raise AgentError("Invalid status batch")
        identities = [instance_identity(target.get("instanceId"), target.get("generation")) for target in targets]
        if len(set(identities)) != len(identities):
            raise AgentError("Duplicate status batch target")
        deadline = time.monotonic() + 20

        def probe(identity):
            instance_id, generation = identity
            target = {"instanceId": instance_id, "generation": generation}
            reply = {"ok": False, "reason": "deadline", "text": ""}
            # A batch does not hold every instance lock while one server waits.
            # Bound concurrency across overlapping batches as well as within one.
            lock = self.instance_lock(target)
            # Use the same lock order as individual health-lane operations.
            if lock.acquire(timeout=max(0, deadline - time.monotonic())):
                try:
                    if self.health_probe_gate.acquire(timeout=max(0, deadline - time.monotonic())):
                        try:
                            marker = self.instance_dir(instance_id, generation).parent / "current.json"
                            current = json.loads(marker.read_text()) if marker.exists() else {}
                            if current.get("generation", generation) != generation or current.get("removed"):
                                reply = {"ok": False, "reason": "superseded", "text": ""}
                            elif time.monotonic() < deadline:
                                reply = self.rcon(instance_id, generation, "status", deadline=deadline, status_attempts=1)
                        except (AgentError, OSError, ValueError, KeyError):
                            reply = {"ok": False, "reason": "unreachable", "text": ""}
                        finally:
                            self.health_probe_gate.release()
                finally:
                    lock.release()
            return {**target, "reply": reply}

        with ThreadPoolExecutor(max_workers=2) as executor:
            results = list(executor.map(probe, identities))
        # Completion describes the batch; individual failures remain in each reply.
        return {"ok": True, "results": results}

    def rcon(self, instance_id, generation, command, deadline=None, status_attempts=3):
        if not isinstance(command, str) or not command or len(command) > 2048 or any(ch in command for ch in ("\x00", "\r", "\n")):
            raise AgentError("Invalid RCON command")
        saved = self.spec(instance_id, generation)
        password = saved.get("rconPassword") or saved.get("env", {}).get("PLUTAINER_RCON_PASSWORD")
        if not password:
            return {"ok": False, "reason": "unconfigured", "players": None, "text": ""}
        bind = saved.get("bindAddress", "0.0.0.0")
        host = "127.0.0.1" if bind in ("0.0.0.0", "") else "::1" if bind == "::" else bind
        family = socket.AF_INET6 if ":" in host else socket.AF_INET
        # Retry only the read-only occupancy query. A missing UDP reply must
        # never cause commands with effects to be replayed.
        attempts = status_attempts if command == "status" else 1
        for attempt in range(attempts):
            if deadline is not None and time.monotonic() >= deadline:
                return {"ok": False, "reason": "deadline" if attempt == 0 else "timeout", "players": None, "text": ""}
            result = self.rcon_attempt(family, host, int(saved["port"]), password, command, **({"deadline": deadline} if deadline is not None else {}))
            if result["ok"] or result.get("reason") not in ("timeout", "unreachable") or attempt == attempts - 1:
                return result
            time.sleep(min(0.25 * 2 ** attempt, max(0, deadline - time.monotonic())) if deadline is not None else 0.25 * 2 ** attempt)

    def rcon_attempt(self, family, host, port, password, command, deadline=None):
        chunks = []
        with socket.socket(family, socket.SOCK_DGRAM) as sock:
            sock.settimeout(max(0.001, min(3, deadline - time.monotonic())) if deadline is not None else 3)
            try:
                sock.connect((host, port))
                sock.send(b"\xff\xff\xff\xffrcon " + password.encode() + b" " + command.encode() + b"\n")
                while len(chunks) < 64 and sum(map(len, chunks)) < 1024 * 1024:
                    if deadline is not None and time.monotonic() >= deadline:
                        break
                    chunks.append(sock.recv(65535))
                    sock.settimeout(max(0.001, min(0.15, deadline - time.monotonic())) if deadline is not None else 0.15)
            except socket.timeout:
                pass
            except OSError:
                return {"ok": False, "reason": "unreachable", "players": None, "text": ""}
        if any(not chunk.startswith(b"\xff\xff\xff\xff") for chunk in chunks):
            return {"ok": False, "reason": "invalid-reply", "players": None, "text": ""}
        text = "\n".join(chunk[4:].decode("utf-8", "replace").removeprefix("print\n") for chunk in chunks)
        if not text or re.search(r"(bad|invalid|wrong).{0,20}(password|rcon)", text, re.I):
            return {"ok": False, "reason": "timeout" if not text else "authentication", "players": None, "text": text}
        players = status_players(text) if command == "status" else None
        return {"ok": True, "text": text, "players": players}

    def validate_spec(self, payload):
        game = payload.get("game")
        if game not in ("t5", "t6"):
            raise AgentError("Only T5 and T6 are supported")
        library = self.library_dir(game, payload.get("libraryId"))
        self.verify_library(library)
        image = payload.get("image", "")
        if not isinstance(image, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/:@-]{0,511}", image):
            raise AgentError("Invalid runtime image")
        port = payload.get("port")
        allowed = self.config.get("portRange", {"min": 28960, "max": 29060})
        if isinstance(port, bool) or not isinstance(port, int) or not allowed["min"] <= port <= allowed["max"]:
            raise AgentError("Port is outside the configured host range")
        bind = payload.get("bindAddress", "0.0.0.0")
        try:
            import ipaddress
            ipaddress.ip_address(bind)
        except (ValueError, TypeError):
            raise AgentError("Invalid local bind address") from None
        env = dict(payload.get("env", {}))
        if payload.get("rconPassword"):
            env["PLUTAINER_RCON_PASSWORD"] = payload["rconPassword"]
        if not isinstance(env, dict) or any(not re.fullmatch(r"[A-Z][A-Z0-9_]*", key) or not isinstance(value, str) or "\x00" in value or "\n" in value for key, value in env.items()):
            raise AgentError("Invalid runtime environment")
        # Runtime download is a separate preparation operation. A deployment
        # uses the already prepared image even during a registry outage.
        image_id = self.docker.call("image", "inspect", "--format", "{{.Id}}", image).strip()
        if not re.fullmatch(r"sha256:[a-f0-9]{64}", image_id):
            raise AgentError("Docker did not report a valid runtime image ID")
        try:
            raw = managed_read(self.root / "agent", "runtime-images.json")
            prepared = json.loads(raw) if raw is not None else {}
        except (OSError, ValueError):
            raise AgentError("Runtime preparation records are unavailable") from None
        if not isinstance(prepared, dict):
            raise AgentError("Runtime preparation records are invalid")
        # The controller also authorizes its configured default without a pull
        # record. Whenever preparation was recorded, enforce that exact image.
        if image in prepared and prepared[image] != image_id:
            raise AgentError("Prepared runtime image has changed. An admin must prepare it again before launching.")
        files = self.initial_files(payload)
        if not isinstance(files, list) or len(files) > 10000:
            raise Rejected("Invalid initial file list")
        total, paths = 0, set()
        for item in files:
            scoped(self.root / "agent" / "validation", item["path"])
            if item["path"] in paths:
                raise Rejected("Duplicate initial file path")
            paths.add(item["path"])
            total += len(self.decode_file(item))
            if total > 64 * 1024 * 1024:
                raise Rejected("Initial files exceed the 64 MiB operation limit")
        return game, library, image, image_id, port, bind, env

    @staticmethod
    def plutonium_manifest(data):
        """Validate the official updater format before using any paths or URLs."""
        if len(data) > PLUTONIUM_MANIFEST_LIMIT:
            raise AgentError("Plutonium manifest is too large")
        try:
            manifest = json.loads(data)
        except (ValueError, UnicodeError):
            raise AgentError("Invalid Plutonium manifest") from None
        if not isinstance(manifest, dict) or manifest.get("product") != "plutonium-core-prod":
            raise AgentError("Invalid Plutonium production manifest")
        revision = manifest.get("revision")
        if type(revision) is not int or not 1 <= revision <= 2147483647:
            raise AgentError("Invalid Plutonium revision")
        base = manifest.get("baseUrl")
        if not isinstance(base, str) or not re.fullmatch(r"https://cdn\.plutoniummod\.com/updater/prod/(?:[A-Za-z0-9_-]+/)*", base):
            raise AgentError("Plutonium downloads must use the official production CDN")
        entries = manifest.get("files")
        if not isinstance(entries, list) or not 1 <= len(entries) <= 4096:
            raise AgentError("Invalid Plutonium file list")
        names, total = set(), 0
        for entry in entries:
            if not isinstance(entry, dict):
                raise AgentError("Invalid Plutonium file entry")
            name, size, checksum = entry.get("name"), entry.get("size"), entry.get("hash")
            if not isinstance(name, str) or len(name) > 512 or re.search(r'[\x00-\x1f\x7f\\:*?"<>|]', name):
                raise AgentError("Invalid Plutonium file path")
            path = PurePosixPath(name)
            if not path.parts or path.is_absolute() or str(path) != name or any(part in (".", "..") or part.endswith((".", " ")) for part in path.parts) or name.lower() == "cdn_info.json":
                raise AgentError("Invalid Plutonium file path")
            if name.lower() in names:
                raise AgentError("Duplicate Plutonium file path")
            names.add(name.lower())
            if type(size) is not int or not 0 <= size <= PLUTONIUM_FILE_LIMIT:
                raise AgentError("Invalid Plutonium file size")
            if not isinstance(checksum, str) or not re.fullmatch(r"[a-f0-9]{40}", checksum):
                raise AgentError("Invalid Plutonium file hash")
            total += size
            if total > PLUTONIUM_CACHE_LIMIT:
                raise AgentError("Plutonium cache exceeds the size limit")
        for name in names:
            if any(str(parent) in names for parent in PurePosixPath(name).parents if str(parent) != "."):
                raise AgentError("Conflicting Plutonium file paths")
        if "bin/plutonium-bootstrapper-win32.exe" not in {entry["name"] for entry in entries}:
            raise AgentError("Plutonium manifest is missing the server bootstrapper")
        # Match plutonium-updater's default: launcher assets are optional.
        return manifest, [entry for entry in entries if not entry["name"].startswith("launcher")]

    @staticmethod
    def plutonium_file_matches(root, entry):
        try:
            with managed_open(root, entry["name"]) as stream:
                if os.fstat(stream.fileno()).st_size != entry["size"]:
                    return False
                checksum = hashlib.sha1()
                for chunk in iter(lambda: stream.read(1024 * 1024), b""):
                    checksum.update(chunk)
                return checksum.hexdigest() == entry["hash"]
        except (OSError, AgentError):
            return False

    def plutonium_open(self, url):
        if not isinstance(url, str) or not re.fullmatch(r"https://cdn\.plutoniummod\.com/updater/prod/[A-Za-z0-9_./-]+", url):
            raise AgentError("Plutonium downloads must use the official production CDN")
        class NoRedirect(urllib.request.HTTPRedirectHandler):
            def redirect_request(self, req, fp, code, msg, hdrs, newurl):
                raise AgentError("Plutonium download redirects are not allowed")
        # Never send the controller's host credentials to the public CDN.
        return urllib.request.build_opener(NoRedirect()).open(url, timeout=30)

    def plutonium_snapshot(self):
        root = scoped(self.root / "agent", "plutonium-cache")
        pointer = json.loads(managed_read(root, "current.json") or b"null")
        if not isinstance(pointer, dict) or not isinstance(pointer.get("key"), str) or not re.fullmatch(r"[a-f0-9]{64}", pointer["key"]):
            raise AgentError("Plutonium cache is not prepared")
        content = scoped(root, pointer["key"] + "/content")
        raw = managed_read(content, "cdn_info.json")
        if raw is None or digest(raw) != pointer["key"]:
            raise AgentError("Plutonium cache manifest is invalid")
        manifest, files = self.plutonium_manifest(raw)
        if pointer.get("revision") != manifest["revision"] or pointer.get("bytes") != sum(entry["size"] for entry in files):
            raise AgentError("Plutonium cache metadata is invalid")
        # Heartbeat checks are cheap. Every file is hashed before it is seeded.
        for entry in files:
            info = scoped(content, entry["name"]).lstat()
            if not stat_module.S_ISREG(info.st_mode) or info.st_size != entry["size"]:
                raise AgentError("Plutonium cache files are incomplete")
        return pointer, content, raw, files

    def plutonium_cache_inventory(self):
        with self.plutonium_lock:
            try:
                pointer, _, _, _ = self.plutonium_snapshot()
                return {"revision": pointer["revision"], "bytes": pointer["bytes"]}
            except (OSError, ValueError, AgentError):
                return None

    def prepare_plutonium(self, payload):
        target, reserve = payload.get("revision", 0), payload.get("minimumFreeBytes", 0)
        if type(target) is not int or not 0 <= target <= 2147483647 or type(reserve) is not int or reserve < 0:
            raise Rejected("Invalid Plutonium preparation request")
        root = scoped(self.root / "agent", "plutonium-cache")
        root.mkdir(mode=0o700, exist_ok=True)
        os.chmod(root, 0o700)
        deadline = time.monotonic() + 1800
        def active():
            if self.stopping.is_set() or time.monotonic() >= deadline:
                raise AgentError("Plutonium cache preparation was interrupted or timed out")
            if self.defer_plutonium_preparation:
                raise Held("Plutonium cache preparation is waiting for other host operations")
        active()
        try:
            with self.plutonium_open(PLUTONIUM_MANIFEST_URL) as response:
                raw = response.read(PLUTONIUM_MANIFEST_LIMIT + 1)
            manifest, files = self.plutonium_manifest(raw)
            if manifest["revision"] < target:
                raise AgentError("Plutonium CDN has not reached the requested revision")
            key = digest(raw)
            previous = None
            with self.plutonium_lock:
                with contextlib.suppress(OSError, ValueError, AgentError):
                    _, previous, _, _ = self.plutonium_snapshot()
            if previous is not None and previous.parent.name == key and all(self.plutonium_file_matches(previous, entry) for entry in files):
                return {"revision": manifest["revision"], "bytes": sum(entry["size"] for entry in files)}
            staging = scoped(root, key + ".partial/content")
            staging.mkdir(parents=True, exist_ok=True)
            expected = {entry["name"] for entry in files}
            # SIGKILL/power loss bypasses download/copy finally blocks. Remove
            # abandoned temporary files before counting space or publishing.
            for directory, subdirs, filenames in os.walk(staging, followlinks=False):
                for name in subdirs + filenames:
                    path = Path(directory) / name
                    if path.is_symlink() or (name in filenames and path.relative_to(staging).as_posix() not in expected):
                        path.unlink()
            # Keep at most the current snapshot and this resumable preparation.
            current_key = previous.parent.name if previous is not None else None
            for directory in root.iterdir():
                if re.fullmatch(r"[a-f0-9]{64}(?:\.partial)?", directory.name) and directory.name not in (current_key, key + ".partial"):
                    if directory.is_symlink():
                        directory.unlink()
                    else:
                        shutil.rmtree(directory)
            missing = sum(entry["size"] for entry in files if not self.plutonium_file_matches(staging, entry))
            if shutil.disk_usage(root).free < reserve + missing + len(raw):
                raise AgentError("Insufficient free storage for the Plutonium cache and host reserve")
            for entry in files:
                active()
                if self.plutonium_file_matches(staging, entry):
                    continue
                if previous is not None and self.plutonium_file_matches(previous, entry):
                    managed_copy(staging, entry["name"], scoped(previous, entry["name"]), os.geteuid(), os.getegid())
                    continue
                destination = scoped(staging, entry["name"])
                destination.parent.mkdir(parents=True, exist_ok=True)
                temporary = destination.with_name(".download-" + uuid.uuid4().hex)
                try:
                    with self.plutonium_open(manifest["baseUrl"] + entry["hash"]) as incoming, temporary.open("xb") as outgoing:
                        size, checksum = 0, hashlib.sha1()
                        while True:
                            active()
                            chunk = incoming.read(min(1024 * 1024, entry["size"] - size + 1))
                            if not chunk:
                                break
                            size += len(chunk)
                            if size > entry["size"]:
                                raise AgentError("Plutonium download exceeds its expected size")
                            checksum.update(chunk)
                            outgoing.write(chunk)
                        if size != entry["size"] or checksum.hexdigest() != entry["hash"]:
                            raise AgentError("Plutonium download failed size or checksum verification")
                        outgoing.flush()
                        os.fsync(outgoing.fileno())
                    os.replace(temporary, destination)
                    sync_parent(destination)
                finally:
                    with contextlib.suppress(FileNotFoundError):
                        temporary.unlink()
            # Publish metadata last. Interrupted transfers can never claim ready.
            active()
            atomic_bytes(staging / "cdn_info.json", raw)
            pointer = {"key": key, "revision": manifest["revision"], "bytes": sum(entry["size"] for entry in files)}
            with self.plutonium_lock:
                final = scoped(root, key)
                if final.exists():
                    # Invalidate before replacing a damaged/same-revision cache.
                    if current_key == key:
                        (root / "current.json").unlink(missing_ok=True)
                    shutil.rmtree(final)
                os.replace(staging.parent, final)
                sync_parent(final)
                atomic_json(root / "current.json", pointer)
                if previous is not None and current_key != key:
                    shutil.rmtree(previous.parent)
            return {"revision": pointer["revision"], "bytes": pointer["bytes"]}
        except (urllib.error.URLError, TimeoutError, http.client.HTTPException):
            raise AgentError("Plutonium CDN download is unavailable; the previous cache is retained") from None

    def seed_plutonium(self, app, image):
        if not re.fullmatch(r"ghcr\.io/ayymoss/plutainer(?::[A-Za-z0-9_.-]+)?(?:@sha256:[a-f0-9]{64})?", image):
            return False
        with self.plutonium_lock:
            try:
                _, content, raw, files = self.plutonium_snapshot()
                if not all(self.plutonium_file_matches(content, entry) for entry in files):
                    scoped(self.root / "agent", "plutonium-cache/current.json").unlink(missing_ok=True)
                    raise AgentError("Plutonium cache checksum verification failed")
            except (OSError, ValueError, AgentError):
                return False
            uid, gid = int(self.config.get("runtimeUid", 1000)), int(self.config.get("runtimeGid", 1000))
            marker = "runtime/plutonium/cdn_info.json"
            managed_write(app, marker, None, uid, gid)
            for entry in files:
                self.check_creation_cancellation()
                managed_copy(app, "runtime/plutonium/" + entry["name"], scoped(content, entry["name"]), uid, gid)
            managed_write(app, marker, raw, uid, gid)
            return True

    def create(self, instance_id, generation, payload, *, before=None):
        self.operation_phase = "checking-instance"
        self.check_creation_cancellation()
        directory = self.instance_dir(instance_id, generation)
        current_marker = directory.parent / "current.json"
        if current_marker.exists():
            current = json.loads(current_marker.read_text())
            if current["generation"] > generation or current["generation"] == generation and current.get("removed"):
                raise AgentError("Instance generation has been superseded or removed")
        existing = self.inspect(instance_id, generation)
        if existing["exists"]:
            if self.spec(instance_id, generation) != payload:
                raise AgentError("Instance already exists with a different specification")
            return existing
        self.operation_phase = "validating-create-spec"
        game, library, image, image_id, port, bind, env = self.validate_spec(payload)
        app = directory / "app"
        self.operation_phase = "preparing-create-directory"
        app.mkdir(parents=True, exist_ok=True)
        self.make_writable(app, directory=True)
        self.operation_phase = "seeding-runtime"
        self.seed_plutonium(app, image)
        self.operation_phase = "writing-create-files"
        self.write_initial_files(app, self.initial_files(payload))
        self.operation_phase = "persisting-create-metadata"
        atomic_json(directory / "spec.json", payload)
        env_path = directory / "environment"
        atomic_bytes(env_path, "".join(f"{key}={value}\n" for key, value in env.items()).encode())
        name = self.container_name(instance_id, generation)
        spec_sha = digest(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode())
        # Use the validated immutable ID so retagging after validation cannot
        # substitute another image. A deleted image must not trigger a pull.
        args = ["container", "create", "--pull=never", "--name", name, "--restart", "unless-stopped", "--label", LABEL + ".managed=true", "--label", LABEL + ".instance=" + instance_id, "--label", LABEL + ".generation=" + str(generation), "--label", LABEL + ".operation=" + (self.executing_id or "direct"), "--label", LABEL + ".spec-sha256=" + spec_sha, "--mount", f"type=bind,src={app},dst=/home/plutainer/app", "--mount", f"type=bind,src={library / 'content'},dst=/home/plutainer/gamefiles,readonly", "--env-file", str(env_path)]
        if self.config.get("hostId"):
            args.extend(["--label", LABEL + ".host=" + identifier(self.config["hostId"])])
        bind_arg = "[" + bind + "]" if ":" in bind else bind
        for protocol in ("udp", "tcp"):
            args.extend(["--publish", f"{bind_arg}:{port}:{port}/{protocol}"])
        args.append(image_id)
        self.check_creation_cancellation()
        self.operation_phase = "creating-container"
        self.checkpoint({"phase": "creating", "generation": generation, "specSha256": spec_sha, **({"before": before} if before is not None else {})})
        self.docker.call(*args, timeout=120)
        self.check_creation_cancellation()
        self.operation_phase = "starting-container"
        self.docker.call("container", "start", name)
        self.check_creation_cancellation()
        self.operation_phase = "recording-created-instance"
        atomic_json(current_marker, {"generation": generation})
        self.operation_phase = "inspecting-created-instance"
        return self.inspect(instance_id, generation)

    def check_creation_cancellation(self):
        if self.executing_id and self.executing_id in self.cancelled_creations:
            raise AgentError("Creation cancelled; waiting for instance cleanup")

    def initial_files(self, payload):
        files = payload.get("files", [])
        if not isinstance(files, list):
            raise Rejected("Invalid initial file list")
        bank = payload.get("bank")
        if bank is None:
            return files
        if not isinstance(bank, dict) or not isinstance(bank.get("token"), str) or not re.fullmatch(r"ec_bank_[A-Za-z0-9_-]{32,128}", bank["token"]):
            raise Rejected("Invalid instance bank credential")
        path = bank.get("configPath")
        scoped(self.root / "agent" / "validation", path)
        if PurePosixPath(path).name != "electric_cherry_bank.cfg":
            raise Rejected("Invalid private bank config path")
        # The game scripts use the public bank API; the host controller address
        # is only for agent transport and must not become a game URL override.
        content = f'set tws_bank_token "{bank["token"]}"\n'.encode()
        return files + [{"path": path, "contentBase64": base64.b64encode(content).decode()}]

    def write_initial_files(self, root, files):
        if not isinstance(files, list) or len(files) > 10000:
            raise AgentError("Invalid file list")
        for item in files:
            content = self.decode_file(item)
            managed_write(root, item["path"], content, int(self.config.get("runtimeUid", 1000)), int(self.config.get("runtimeGid", 1000)))

    def decode_file(self, item):
        if "contentSha256" in item:
            sha = item["contentSha256"]
            if not isinstance(sha, str) or not re.fullmatch(r"[a-f0-9]{64}", sha):
                raise AgentError("Invalid file content hash")
            url = self.config["controller"].rstrip("/") + "/api/agent/blobs/" + sha
            with self.artifact_open("GET", url) as response:
                content = response.read(MAX_FILE + 1)
            if len(content) > MAX_FILE:
                raise AgentError("File exceeds staging size limit")
            if digest(content) != sha or item.get("sha256", sha) != sha:
                raise AgentError("Staged file checksum mismatch")
            return content
        try:
            content = base64.b64decode(item["contentBase64"], validate=True)
        except (ValueError, KeyError, TypeError):
            raise AgentError("Invalid encoded file content") from None
        if len(content) > MAX_FILE:
            raise AgentError("File exceeds staging size limit")
        if item.get("sha256") and item["sha256"] != digest(content):
            raise AgentError("Staged file checksum mismatch")
        return content

    def authorize_autoscaling_removal(self, instance_id, generation, payload):
        authorization = payload.get("autoscaling")
        if not isinstance(authorization, dict) or not self.executing_id or payload.get("safetyOverride") or payload.get("stageId"):
            raise AutoscalingInvalidated("Invalid autoscaling removal authorization")
        started = time.monotonic()
        try:
            response = self.request("POST", "/api/agent/operations/" + urllib.parse.quote(self.executing_id, safe="") + "/authorize-removal", authorization)
        except AgentError:
            raise AutoscalingInvalidated("Autoscaling removal authorization unavailable; request a new decision") from None
        ttl = response.get("validForMs") if isinstance(response, dict) else None
        if not isinstance(response, dict) or response.get("authorized") is not True or isinstance(ttl, bool) or not isinstance(ttl, (int, float)) or not 0 < ttl <= 5000:
            raise AutoscalingInvalidated("Autoscaling removal authorization was withdrawn or expired")
        # Occupancy can change during the controller round trip. Check again
        # within the short claim window, immediately before the durable checkpoint.
        self.safe_to_disrupt(instance_id, generation)
        if (time.monotonic() - started) * 1000 >= ttl:
            raise AutoscalingInvalidated("Autoscaling removal authorization expired before mutation")

    def settle_bank(self, instance_id, generation, operation_id=None):
        """Drain stopped runtime journals before discarding any generation's storage.

        Use only our configured controller and host credential. Never follow an
        endpoint or symlink supplied by a game journal. Lost replies are retried
        with the original request identity; files remain until teardown succeeds.
        """
        app = scoped(self.instance_dir(instance_id, generation), "app")
        if not app.exists():
            return
        if self.inspect(instance_id, generation)["running"]:
            raise AgentError("Bank settlement requires a stopped instance")
        operation_id = operation_id or self.executing_id
        visited = 0
        deadline = time.monotonic() + 30
        journals = []

        def unreadable(_error):
            raise AgentError("Bank journal discovery failed; runtime storage retained")

        for directory, directories, files in os.walk(app, followlinks=False, onerror=unreadable):
            relative = Path(directory).relative_to(app)
            inside_bank = "tws-bank" in relative.parts
            visited += 1
            if visited > 100000 or time.monotonic() > deadline:
                raise AgentError("Bank journal discovery limit exceeded; runtime storage retained")
            for name in directories + files:
                if (inside_bank or name == "tws-bank") and (Path(directory) / name).is_symlink():
                    raise AgentError("Bank journal paths must not be symlinks")
            if not inside_bank:
                continue
            for name in files:
                if name.endswith(".json"):
                    journals.append((relative / name).as_posix())
                    if len(journals) > 4096:
                        raise AgentError("Too many bank journals; runtime storage retained")
        for path in sorted(journals):
            try:
                raw = managed_read(app, path)
            except Rejected:
                raise AgentError("Bank journal cannot be read; runtime storage retained") from None
            if raw is None or len(raw) > 16 * 1024 * 1024:
                raise AgentError("Bank journal is unavailable or oversized")
            try:
                journal = json.loads(raw)
                reports = journal["reports"]
                # Accept the fixed game endpoint and this host's legacy generated
                # endpoint so rebuild/removal can drain pre-upgrade decisions.
                # Neither value controls transport: settlement still uses the
                # enrolled controller, host credential and immutable operation.
                endpoints = {BANK_GAME_ENDPOINT, self.config["controller"].rstrip("/") + "/api/bank"}
                if journal["endpoint"].rstrip("/") not in endpoints:
                    raise AgentError("Bank journal has an unrecognized endpoint; runtime storage retained")
                if type(journal["version"]) is not int or journal["version"] != 1 or not isinstance(reports, list) or len(reports) > 4096:
                    raise ValueError()
                # Controller independently binds this operation to the instance/server.
                for report in reports:
                    if not isinstance(report, str) or len(report.encode()) > 20000:
                        raise ValueError()
                    transfer = json.loads(report)
                    action = transfer["decision"]
                    if action not in ("commit", "cancel") or transfer["server"] != journal["server"] or not operation_id:
                        raise ValueError()
                    response = self.request("POST", "/api/agent/bank/settle", {
                        "operationId": operation_id, "action": action, "transfer": transfer})
                    expected = ("committed",) if action == "commit" else ("cancelled", "rejected")
                    if not isinstance(response, dict) or response.get("request_id") != transfer["request_id"] or response.get("state") not in expected:
                        raise AgentError("Bank settlement was not acknowledged; runtime storage retained")
            except (ValueError, TypeError, KeyError, AttributeError):
                raise AgentError("Bank journal is invalid; runtime storage retained") from None

    def lifecycle(self, kind, instance_id, generation, payload):
        unhealthy_recovery = payload.get("unhealthyRecovery") is True
        if unhealthy_recovery and (kind not in ("instance.restart", "instance.rebuild") or payload.get("stageId") or "autoscaling" in payload):
            raise AgentError("Unhealthy recovery only permits restart or rebuild without file activation or autoscaling")
        if kind == "instance.cancel-create":
            return self.cancel_creation(instance_id, generation, payload)
        if kind == "instance.create":
            return self.create(instance_id, generation, payload)
        if kind == "instance.inspect":
            return self.inspect(instance_id, generation)
        health = self.inspect(instance_id, generation)
        name = self.container_name(instance_id, generation)
        before = []
        if kind == "instance.start":
            if not health["exists"]:
                raise AgentError("Instance does not exist")
            if payload.get("stageId"):
                self.safe_to_disrupt(instance_id, generation)
                self.checkpoint({"phase": "activating", "containerId": health.get("containerId"), "startedAt": health.get("startedAt"), "restartRequired": health["running"]})
                before = self.activate(instance_id, generation, payload["stageId"])
                health = self.inspect(instance_id, generation)
                self.safe_to_disrupt(instance_id, generation)
            restart_required = bool(payload.get("stageId") and health["running"])
            self.checkpoint({"phase": "starting", "containerId": health.get("containerId"), "startedAt": health.get("startedAt"), "restartRequired": restart_required})
            if restart_required:
                self.docker.call("container", "restart", "--time", "30", name)
            else:
                self.docker.call("container", "start", name)
        elif kind in ("instance.stop", "instance.restart", "instance.remove", "instance.rebuild"):
            # The authenticated controller records explicit administrator approval
            # for this immutable operation. Ownership and generation checks still apply.
            override_safety = payload.get("safetyOverride") is True
            if "autoscaling" in payload and (kind != "instance.remove" or override_safety):
                raise AutoscalingInvalidated("Autoscaling only permits removal with player safety checks")
            if kind == "instance.rebuild":
                next_generation = payload.get("nextGeneration", payload.get("newGeneration"))
                if isinstance(next_generation, bool) or not isinstance(next_generation, int) or next_generation <= generation:
                    raise AgentError("Rebuild must use a newer immutable generation")
                spec = payload.get("spec") or {key: value for key, value in payload.items() if key not in ("nextGeneration", "newGeneration", "unhealthyRecovery")}
                self.validate_spec(spec)
            if not override_safety:
                self.safe_to_disrupt(instance_id, generation, allow_unhealthy=unhealthy_recovery)
            if "autoscaling" in payload:
                self.authorize_autoscaling_removal(instance_id, generation, payload)
            if kind == "instance.rebuild" and payload.get("stageId"):
                before = self.rebuild_before(instance_id, generation, payload["stageId"], spec)
                if not override_safety:
                    self.safe_to_disrupt(instance_id, generation)
            self.checkpoint({"phase": "disrupting", "containerId": health.get("containerId"), "startedAt": health.get("startedAt"), **({"before": before} if kind == "instance.rebuild" and payload.get("stageId") else {})})
            if payload.get("stageId") and kind == "instance.restart":
                before = self.activate(instance_id, generation, payload["stageId"], override_safety=override_safety)
            if kind in ("instance.remove", "instance.rebuild"):
                if health["exists"]:
                    self.docker.call("container", "stop", "--time", "30", name)
                    self.docker.call("container", "rm", name)
                self.settle_bank(instance_id, generation)
                if kind == "instance.rebuild":
                    result = self.create(instance_id, next_generation, spec, before=before if payload.get("stageId") else None)
                    self.operation_phase = "removing-old-instance"
                    old = self.instance_dir(instance_id, generation)
                    if old.exists():
                        shutil.rmtree(old, ignore_errors=False)
                    return {**result, "before": before}
                directory = self.instance_dir(instance_id, generation)
                if directory.exists():
                    shutil.rmtree(directory)
                self.mark_removed(instance_id, generation)
            elif health["exists"]:
                if kind == "instance.restart" and payload.get("stageId") and not override_safety:
                    self.safe_to_disrupt(instance_id, generation)
                self.docker.call("container", "stop" if kind.endswith("stop") else "restart", "--time", "30", name)
        else:
            raise AgentError("Unsupported instance operation")
        return {**self.inspect(instance_id, generation), "before": before}

    def cancel_creation(self, instance_id, generation, payload, operation_id=None):
        """Explicit launch cancellation, fenced to its immutable creation identity.

        The operation loop serializes this with create. Once delivered, removal
        needs no RCON response and a durable tombstone prevents a delayed create.
        Repeating this cleanup after an ambiguous Docker result is safe.
        """
        create_id = identifier(payload.get("createOperationId"))
        targets = [(generation, create_id)]
        if "recoveryRebuildOperationId" in payload or "recoveryGeneration" in payload:
            rebuild_id = identifier(payload.get("recoveryRebuildOperationId"))
            recovery_generation = payload.get("recoveryGeneration")
            if isinstance(recovery_generation, bool) or not isinstance(recovery_generation, int) or recovery_generation != generation + 1:
                raise Rejected("Cancellation recovery must target the next generation")
            saved = self.journal.db.execute("SELECT request FROM operations WHERE id=?", (rebuild_id,)).fetchone()
            if saved:
                rebuild = json.loads(saved["request"])
                rebuild_payload = rebuild.get("payload", {})
                if rebuild["kind"] != "instance.rebuild" or rebuild["instanceId"] != instance_id or rebuild["generation"] != generation or rebuild_payload.get("nextGeneration", rebuild_payload.get("newGeneration")) != recovery_generation or rebuild_payload.get("unhealthyRecovery") is not True:
                    raise Rejected("Cancellation does not match the creation recovery rebuild")
            targets.append((recovery_generation, rebuild_id))
        marker = self.instance_dir(instance_id, generation).parent / "current.json"
        if marker.exists() and json.loads(marker.read_text())["generation"] > targets[-1][0]:
            raise Rejected("Cancelled creation was superseded by a newer generation")
        saved = self.journal.db.execute("SELECT request FROM operations WHERE id=?", (create_id,)).fetchone()
        if saved:
            original = json.loads(saved["request"])
            if original["kind"] != "instance.create" or original["instanceId"] != instance_id or original["generation"] != generation:
                raise Rejected("Cancellation does not match the original creation")
        health_by_generation = {}
        for target_generation, operation_id in targets:
            health = self.inspect(instance_id, target_generation)
            if health["exists"] and health.get("creationOperation") != operation_id:
                raise Rejected("Container belongs to a different creation operation")
            health_by_generation[target_generation] = health
        self.checkpoint({"phase": "cancelling-creation", "createOperationId": create_id})
        # Fence both generations before deleting: delayed create/rebuild work
        # must never recreate either target after cancellation or a crash.
        self.mark_removed(instance_id, targets[-1][0])
        for target_generation, _ in targets:
            if health_by_generation[target_generation]["exists"]:
                self.docker.call("container", "rm", "--force", "--volumes", self.container_name(instance_id, target_generation))
            if self.inspect(instance_id, target_generation)["exists"]:
                raise AgentError("Cancelled creation container still exists")
            self.settle_bank(instance_id, target_generation, operation_id)
            directory = self.instance_dir(instance_id, target_generation)
            if directory.exists():
                shutil.rmtree(directory)
        return {**self.inspect(instance_id, generation), "cancelledCreation": create_id}

    def stage(self, instance_id, generation, payload, operation_id):
        directory = self.instance_dir(instance_id, generation)
        self.spec(instance_id, generation)
        stage_id = identifier(payload.get("stageId", operation_id))
        stage = directory / "stages" / stage_id
        files = payload.get("files")
        if not isinstance(files, list) or not files or len(files) > 10000:
            raise AgentError("Invalid staged file list")
        planned = []
        paths = set()
        total_size = 0
        for item in files:
            scoped(directory / "app", item["path"])
            if item["path"] in paths:
                raise AgentError("Duplicate staged file path")
            paths.add(item["path"])
            deleting = item.get("delete") is True or (item.get("contentBase64") is None and "contentSha256" not in item)
            content = None if deleting else self.decode_file(item)
            total_size += len(content) if content is not None else 0
            if total_size > 64 * 1024 * 1024:
                raise Rejected("Staged file content exceeds the 64 MiB operation limit")
            planned.append({"id": item.get("id"), "path": item["path"], "delete": deleting, "sha256": None if deleting else digest(content), "checkBefore": item.get("checkBefore", "expectedSha256" in item), "expectedBeforeSha": item.get("expectedBeforeSha", item.get("expectedSha256")), "contentBase64": None if deleting else base64.b64encode(content).decode()})
        manifest = {"stageId": stage_id, "files": planned, "state": "staged"}
        marker = stage / "manifest.json"
        if marker.exists():
            existing = json.loads(marker.read_text())
            if existing["files"] != planned:
                raise AgentError("Stage ID already contains different files")
            return {"stageId": stage_id, "files": len(planned), "state": existing["state"]}
        atomic_json(marker, manifest)
        return {"stageId": stage_id, "files": len(planned), "state": "staged"}

    def rebuild_before(self, instance_id, generation, stage_id, spec):
        """Validate requested replacement bytes and snapshot the old generation before teardown."""
        directory = self.instance_dir(instance_id, generation)
        marker = directory / "stages" / identifier(stage_id) / "manifest.json"
        saved = json.loads(marker.read_text())
        if saved["state"] != "staged":
            raise Rejected("Rebuild requires an unactivated file stage")
        replacement = {item["path"]: item for item in self.initial_files(spec)}
        before = []
        for item in saved["files"]:
            target = replacement.get(item["path"])
            if item["delete"]:
                if target is not None:
                    raise Rejected("Rebuild includes a file requested for removal")
            elif target is None or digest(self.decode_file(target)) != item["sha256"]:
                raise Rejected("Rebuild does not match staged file content")
            content = managed_read(directory / "app", item["path"])
            actual_sha = None if content is None else digest(content)
            if item.get("checkBefore") and actual_sha != item.get("expectedBeforeSha"):
                raise Rejected("Target changed after campaign review")
            before.append({"id": item.get("id"), "path": item["path"], "contentBase64": None if content is None else base64.b64encode(content).decode()})
        return before

    def activate(self, instance_id, generation, stage_id, *, override_safety=False):
        directory = self.instance_dir(instance_id, generation)
        stage = directory / "stages" / identifier(stage_id)
        marker = stage / "manifest.json"
        saved = json.loads(marker.read_text())
        if saved["state"] == "activated":
            return saved["before"]
        if saved["state"] != "staged":
            raise AgentError("Interrupted activation requires explicit rollback")
        before = []
        for item in saved["files"]:
            content = managed_read(directory / "app", item["path"])
            actual_sha = None if content is None else digest(content)
            if item.get("checkBefore") and actual_sha != item.get("expectedBeforeSha"):
                raise Rejected("Target changed after campaign review")
            before.append({"id": item.get("id"), "path": item["path"], "contentBase64": None if content is None else base64.b64encode(content).decode()})
        # Snapshot preparation may take time. Occupancy is rechecked at the
        # local activation boundary, after all file I/O and integrity checks.
        if not override_safety:
            self.safe_to_disrupt(instance_id, generation)
        saved.update(state="activating", before=before)
        atomic_json(marker, saved)
        try:
            for item in saved["files"]:
                managed_write(directory / "app", item["path"], None if item["delete"] else self.decode_file(item), int(self.config.get("runtimeUid", 1000)), int(self.config.get("runtimeGid", 1000)))
            saved["state"] = "activated"
            atomic_json(marker, saved)
        except Exception:
            self.restore_files(directory / "app", before)
            saved["state"] = "rolled-back"
            atomic_json(marker, saved)
            raise
        return before

    def restore_files(self, app, before):
        for item in before:
            managed_write(app, item["path"], None if item["contentBase64"] is None else self.decode_file(item), int(self.config.get("runtimeUid", 1000)), int(self.config.get("runtimeGid", 1000)))

    def file_operation(self, kind, instance_id, generation, payload, operation_id):
        directory = self.instance_dir(instance_id, generation)
        app = directory / "app"
        if kind == "files.stage":
            return self.stage(instance_id, generation, payload, operation_id)
        if kind in ("files.activate", "files.rollback"):
            self.safe_to_disrupt(instance_id, generation)
            stage_id = identifier(payload["stageId"])
            health = self.inspect(instance_id, generation)
            self.checkpoint({"phase": "activating" if kind == "files.activate" else "rolling-back", "containerId": health.get("containerId"), "startedAt": health.get("startedAt")})
            if kind == "files.activate":
                before = self.activate(instance_id, generation, stage_id)
                result = {"activated": True, "before": before}
            else:
                marker = directory / "stages" / stage_id / "manifest.json"
                saved = json.loads(marker.read_text())
                if saved["state"] not in ("activated", "activating", "rolled-back"):
                    raise AgentError("Stage has no activation to roll back")
                if saved["state"] != "rolled-back":
                    self.restore_files(app, saved["before"])
                    saved["state"] = "rolled-back"
                    atomic_json(marker, saved)
                result = {"rolledBack": True}
            if payload.get("restart", True) and self.inspect(instance_id, generation)["exists"]:
                self.safe_to_disrupt(instance_id, generation)
                self.docker.call("container", "restart", "--time", "30", self.container_name(instance_id, generation))
            return result
        if kind == "logs.read":
            return self.read_log(instance_id, generation, payload)
        if kind == "files.manifest":
            return managed_manifest(app, payload, self.stopping)
        path = scoped(app, payload.get("path", ""), allow_root=kind in ("files.list", "files.stat"))
        if kind == "files.read":
            content = managed_read(app, payload["path"])
            if content is None:
                return {"contentBase64": None}
            return {"contentBase64": base64.b64encode(content).decode(), "sha256": digest(content), "size": len(content)}
        if kind == "files.stat":
            if not path.exists():
                return None
            stat = path.stat()
            return {"size": stat.st_size, "mtime": stat.st_mtime, "isDirectory": path.is_dir(), "isFile": path.is_file()}
        if kind == "files.list":
            if not path.exists():
                return {"entries": []}
            return {"entries": [{"name": child.name, "path": child.relative_to(app).as_posix(), "isDirectory": child.is_dir(), "size": child.stat().st_size} for child in sorted(path.iterdir()) if not child.is_symlink()][:10000]}
        raise AgentError("Unsupported file operation")

    def read_log(self, instance_id, generation, payload):
        app = self.instance_dir(instance_id, generation) / "app"
        value = payload["path"]
        count = min(2000, max(1, int(payload.get("lines", 200))))
        # Validate the original relative spelling independently of symlinks.
        scoped(self.root / "agent" / "log-validation", value)
        try:
            resolved = (app / value).resolve(strict=True)
            relative = resolved.relative_to(app.resolve(strict=True)).as_posix()
        except FileNotFoundError:
            if value != "logs/console.log" or not self.inspect(instance_id, generation)["exists"]:
                return None
            text = self.docker.call("container", "logs", "--tail", str(count), "--timestamps", self.container_name(instance_id, generation), timeout=30, merge_stderr=True)
            lines = text.splitlines()
            return {"lines": lines[-count:], "size": len(text.encode()), "mtime": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "truncated": len(lines) >= count, "source": "container"}
        except ValueError:
            raise AgentError("Log symlink resolves outside the managed instance") from None
        # Reopen the resolved relative target through no-follow directory FDs;
        # swapping either the original link or a target component cannot escape.
        with managed_open(app, relative) as stream:
            stat = os.fstat(stream.fileno())
            if payload.get("mode") == "incremental":
                return self.read_log_chunk(stream, stat, payload.get("cursor"))
            start = max(0, stat.st_size - 256 * 1024)
            stream.seek(start)
            lines = stream.read(256 * 1024).decode("utf-8", "replace").splitlines()
        if start:
            lines = lines[1:]
        return {"lines": lines[-count:], "size": stat.st_size, "mtime": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(stat.st_mtime)), "truncated": start > 0}

    @staticmethod
    def read_log_chunk(stream, stat, cursor):
        file_id, modified = f"{stat.st_dev}:{stat.st_ino}", f"{stat.st_mtime_ns}:{stat.st_ctime_ns}"
        size = stat.st_size
        mtime = datetime.fromtimestamp(stat.st_mtime, timezone.utc).isoformat()
        if cursor is not None and (not isinstance(cursor, dict)
                or any(type(cursor.get(key)) is not int or cursor[key] < 0 for key in ("offset", "size", "anchorLength"))
                or cursor["anchorLength"] > min(256, cursor["offset"])
                or any(not isinstance(cursor.get(key), str) for key in ("fileId", "modified", "anchorHash"))):
            raise AgentError("Invalid log cursor")
        if cursor and cursor["fileId"] == file_id and cursor["modified"] == modified and cursor["size"] == size and cursor["offset"] == size:
            return {"contentBase64": "", "size": size, "mtime": mtime, "truncated": size > 0, "cursor": cursor, "reset": False}
        reset = not cursor or cursor["fileId"] != file_id or cursor["offset"] > size
        if not reset:
            stream.seek(cursor["offset"] - cursor["anchorLength"])
            anchor = stream.read(cursor["anchorLength"])
            reset = len(anchor) != cursor["anchorLength"] or digest(anchor) != cursor["anchorHash"]
        start = max(0, size - 256 * 1024) if reset else cursor["offset"]
        stream.seek(start)
        content = stream.read(min(256 * 1024, size - start))
        offset = start + len(content)
        stream.seek(max(0, offset - 256))
        anchor = stream.read(min(256, offset))
        return {"contentBase64": base64.b64encode(content).decode(), "size": size, "mtime": mtime, "truncated": start > 0, "reset": reset,
                "cursor": {"fileId": file_id, "modified": modified, "size": size, "offset": offset, "anchorLength": len(anchor), "anchorHash": digest(anchor)}}

    def library_dir(self, game, library_id):
        if game not in ("t5", "t6"):
            raise AgentError("Unsupported library game")
        return self.root / "libraries" / game / identifier(library_id)

    def verify_library(self, directory):
        marker = directory / "manifest.json"
        if not marker.exists():
            raise AgentError("Required game library is not ready")
        manifest = json.loads(marker.read_text())
        for item in manifest["files"]:
            path = scoped(directory / "content", item["path"])
            if not path.is_file() or path.stat().st_size != item["size"] or file_digest(path) != item["sha256"]:
                raise AgentError("Game library integrity verification failed")
        return manifest

    def artifact_url(self, value):
        controller = self.config["controller"].rstrip("/")
        url = urllib.parse.urljoin(controller + "/", value)
        parsed = urllib.parse.urlsplit(url)
        origin = urllib.parse.urlsplit(controller)
        if (parsed.scheme, parsed.netloc) != (origin.scheme, origin.netloc) or parsed.username or parsed.password:
            raise AgentError("Library transport must remain on the authenticated controller origin")
        if parsed.scheme != "https" and not (self.config.get("allowInsecureLocalhost") and parsed.hostname in ("127.0.0.1", "localhost", "::1")):
            raise AgentError("Library transport must use HTTPS")
        return url

    def artifact_open(self, method, path, data=None, extra_headers=None, timeout=60):
        headers = {"Authorization": "Bearer " + self.config["credential"], **(extra_headers or {})}
        req = urllib.request.Request(self.artifact_url(path), data=data, method=method, headers=headers)
        # Redirects cannot forward host credentials to another origin.
        class NoRedirect(urllib.request.HTTPRedirectHandler):
            def redirect_request(self, req, fp, code, msg, hdrs, newurl):
                raise AgentError("Artifact redirects are not allowed")
        return urllib.request.build_opener(NoRedirect()).open(req, timeout=timeout)

    def download_library_file(self, item, target, files_url, update_progress, aborted):
        def check_active():
            if aborted.is_set():
                raise AgentError("Library download stopped after a transfer failed")

        check_active()
        target.parent.mkdir(parents=True, exist_ok=True)
        offset = target.stat().st_size if target.exists() else 0
        if offset > item["size"]:
            target.unlink()
            offset = 0
        # Saved prefixes and complete files are intentionally not read here.
        # The second phase verifies every byte after all transfers have ended.
        if offset == item["size"] and target.exists():
            update_progress(item["path"], offset)
            return
        update_progress(item["path"], offset)
        check_active()
        url = files_url.rstrip("/") + "/" + urllib.parse.quote(item["path"], safe="/")
        with self.artifact_open("GET", url, extra_headers={"Range": f"bytes={offset}-"} if offset else {}) as response:
            if response.status == 206:
                expected = f"bytes {offset}-{item['size'] - 1}/{item['size']}"
                if response.headers.get("Content-Range") != expected:
                    raise AgentError("Invalid resumed download response")
            elif response.status != 200:
                raise AgentError("Invalid library download response")
            elif offset:
                # A server may ignore Range and send the whole file instead.
                offset = 0
                update_progress(item["path"], 0)
            with target.open("ab" if offset else "wb") as output:
                received = offset
                while True:
                    check_active()
                    chunk = response.read(1024 * 1024)
                    if not chunk:
                        break
                    received += len(chunk)
                    if received > item["size"]:
                        raise AgentError("Library download exceeded its manifest size")
                    output.write(chunk)
                    update_progress(item["path"], received)
                output.flush()
                os.fsync(output.fileno())
        if target.stat().st_size != item["size"]:
            raise AgentError("Downloaded library file size failed integrity verification")

    def verify_downloaded_files(self, files, targets, transfer, repair=False):
        invalid = False
        retained = transfer.progress["bytesDownloaded"]
        for item in files:
            target = targets[item["path"]]
            checksum = hashlib.sha256()
            actual_size = target.stat().st_size if target.is_file() else 0
            valid_size = target.is_file() and actual_size == item["size"]
            if valid_size:
                with target.open("rb") as stream:
                    for chunk in iter(lambda: stream.read(1024 * 1024), b""):
                        checksum.update(chunk)
                        transfer.advance("bytesVerified", len(chunk), len(chunk))
            if not valid_size or checksum.hexdigest() != item["sha256"]:
                invalid = True
                if repair and target.exists():
                    # A corrupt complete file (including a corrupt saved
                    # prefix) must not be resumed indefinitely on every retry.
                    target.unlink()
                    actual_size = 0
                retained -= item["size"] - min(actual_size, item["size"])
        if invalid:
            # Verification describes a completed download phase. Keep that
            # total stable while other files are checked, then publish retained
            # bytes atomically with failure (including removed corrupt files).
            transfer.failed_bytes_downloaded = retained
            raise AgentError("Downloaded library file failed integrity verification")

    def publish_library_progress(self, payload, transfer):
        if payload.get("statusUrl"):
            # Phase boundaries are useful even when shorter than the heartbeat
            # interval. Telemetry failure must never undo a verified download.
            with contextlib.suppress(AgentError, OSError, ValueError):
                with self.artifact_open("POST", payload["statusUrl"], json.dumps(transfer.snapshot()).encode(), {"Content-Type": "application/json"}, timeout=5) as response:
                    response.read()

    def library_download(self, payload):
        library_id = identifier(payload.get("libraryVersionId", payload.get("libraryId")))
        with self.artifact_open("GET", payload["manifestUrl"]) as response:
            manifest = json.load(response)
        if manifest.get("id") != library_id or not isinstance(manifest.get("files"), list) or not manifest["files"]:
            raise AgentError("Library manifest identity mismatch")
        files = manifest["files"]
        canonical = json.dumps([{ "path": item["path"], "size": item["size"], "sha256": item["sha256"] } for item in files], separators=(",", ":"), ensure_ascii=False)
        manifest_sha = manifest.get("manifestSha256")
        if not isinstance(manifest_sha, str) or not re.fullmatch("[0-9a-f]{64}", manifest_sha):
            raise AgentError("Library manifest checksum is missing")
        if digest(canonical.encode("utf-8")) != manifest_sha:
            raise AgentError("Library manifest checksum mismatch")
        if payload.get("manifestSha256") and payload["manifestSha256"] != manifest_sha:
            raise AgentError("Library manifest differs from the requested immutable version")
        directory = self.library_dir(manifest["game"], library_id)
        existing = directory.exists()
        if existing:
            saved = json.loads((directory / "manifest.json").read_text())
            if saved.get("manifestSha256") != manifest_sha:
                raise AgentError("Published library changed unexpectedly")
        staging = directory if existing else directory.with_name("." + library_id + ".partial")
        staging.mkdir(parents=True, exist_ok=True)
        source_marker = staging / "source.json"
        if source_marker.exists() and json.loads(source_marker.read_text()).get("manifestSha256") != manifest_sha:
            raise Rejected("Partial library download belongs to a different manifest")
        size = sum(item["size"] for item in files)
        atomic_json(source_marker, {"libraryId": library_id, "game": manifest["game"], "size": size, "files": files, "manifestSha256": manifest_sha})
        targets = {}
        progress = {"libraryVersionId": library_id, "status": "verifying" if existing else "downloading", "bytesDownloaded": size if existing else 0, "bytesVerified": 0, "bytesTotal": size, "manifestSha256": manifest_sha}
        self.library_progress[library_id] = progress
        marker = staging / "transfer.json"
        previous = json.loads(marker.read_text()) if marker.exists() else None
        if previous and (previous.get("manifestSha256") != manifest_sha or previous.get("status") == "ready"):
            previous = None
        timing = LibraryTransfer(progress, marker, previous)
        self.library_transfers[library_id] = timing
        try:
            # Validate every path before starting concurrent writes. In addition
            # to duplicates, a manifest cannot treat a file as a parent folder.
            for item in files:
                if item["path"] in targets or type(item["size"]) is not int or item["size"] < 0 or not re.fullmatch("[0-9a-f]{64}", item["sha256"]):
                    raise AgentError("Invalid library file manifest")
                targets[item["path"]] = scoped(staging / "content", item["path"])
            for path in targets:
                if any(str(parent) in targets for parent in PurePosixPath(path).parents):
                    raise AgentError("Library file paths overlap")
            progress_lock = threading.Lock()
            offsets = {item["path"]: min(targets[item["path"]].stat().st_size, item["size"]) if targets[item["path"]].exists() else 0 for item in files}
            if not existing:
                progress["bytesDownloaded"] = sum(offsets.values())
            if not existing and shutil.disk_usage(staging).free < size - progress["bytesDownloaded"]:
                raise AgentError("Insufficient library storage")
            timing.persist(force=True)
            self.publish_library_progress(payload, timing)
            aborted = threading.Event()

            def update_progress(path, offset):
                with progress_lock:
                    delta = offset - offsets.get(path, 0)
                    offsets[path] = offset
                    timing.advance("bytesDownloaded", delta, max(0, delta))

            def transfer(item):
                try:
                    self.download_library_file(item, targets[item["path"]], payload["filesBaseUrl"], update_progress, aborted)
                except Exception:
                    aborted.set()
                    raise

            # Bound sockets, memory and disk pressure while allowing other files
            # to progress during network round trips and per-file fsyncs.
            if not existing:
                # A crash after setting read-only permissions but before the
                # atomic rename leaves a valid, resumable staging directory.
                content_root = staging / "content"
                if content_root.exists():
                    os.chmod(content_root, 0o755)
                    for path in content_root.rglob("*"):
                        if path.is_symlink():
                            raise AgentError("Symlinks are not allowed in library content")
                        os.chmod(path, 0o755 if path.is_dir() else 0o644)
                with ThreadPoolExecutor(max_workers=min(LIBRARY_DOWNLOAD_WORKERS, len(files))) as pool:
                    futures = [pool.submit(transfer, item) for item in files]
                    for future in as_completed(futures):
                        future.result()
                timing.begin_verification()
                self.publish_library_progress(payload, timing)
            self.verify_downloaded_files(files, targets, timing, repair=not existing)
            saved = {"libraryId": library_id, "game": manifest["game"], "size": size, "files": files, "manifestSha256": manifest_sha}
            atomic_json(staging / "manifest.json", saved)
            if not existing:
                for path in (staging / "content").rglob("*"):
                    if path.is_symlink():
                        raise AgentError("Symlinks are not allowed in library content")
                    os.chmod(path, 0o555 if path.is_dir() else 0o444)
                os.chmod(staging / "content", 0o555)
                os.replace(staging, directory)
                timing.marker = directory / "transfer.json"
            timing.finish("ready")
            saved["transfer"] = timing.snapshot()["transfer"]
            saved["bytesVerified"] = size
            atomic_json(directory / "manifest.json", saved)
            self.publish_library_progress(payload, timing)
            self.library_progress.pop(library_id, None)
            self.library_transfers.pop(library_id, None)
            return {"libraryId": library_id, "state": "ready", "size": size, "manifestSha256": manifest_sha}
        except Exception:
            timing.finish("failed", "Library download or integrity verification failed")
            self.publish_library_progress(payload, timing)
            raise

    def library_remove(self, payload):
        library_id = identifier(payload.get("libraryVersionId"))
        directory = self.library_dir(payload.get("game"), library_id)
        staging = directory.with_name("." + library_id + ".partial")
        # Never follow substituted storage parents outside the managed tree.
        for path in (self.root / "libraries", directory.parent, directory, staging):
            if path.is_symlink():
                raise Rejected("Refusing library cleanup through a symlink")
        # Include stopped and unmanaged containers: any bind mount pins the files.
        # Docker failures must stop cleanup, never masquerade as an empty host.
        for container_id in self.docker.call("container", "ls", "--all", "--quiet").splitlines():
            container = self.docker.inspect(container_id)
            if container is None:
                raise Rejected("Container inventory changed during library cleanup")
            for mount in container["Mounts"]:
                if not mount.get("Source"):  # tmpfs has no host filesystem source.
                    continue
                source = Path(mount["Source"]).resolve()
                for target in (directory.resolve(), staging.resolve()):
                    if source == target or target in source.parents or source in target.parents:
                        raise Rejected("Game files are still mounted by a container")
        self.checkpoint({"phase": "removing-library"})
        for path in (directory, staging):
            if path.exists():
                shutil.rmtree(path)
        self.library_progress.pop(library_id, None)
        self.library_transfers.pop(library_id, None)
        return {"libraryVersionId": library_id, "removed": True}

    def library_import(self, payload):
        game = payload.get("game")
        volume = payload.get("sourceVolume")
        if game not in ("t5", "t6") or volume != game + "_game_files":
            raise AgentError("Only the explicitly approved game-file Docker volumes can be imported")
        raw = self.docker.call("volume", "inspect", volume)
        source = Path(json.loads(raw)[0]["Mountpoint"])
        if not source.is_dir() or source.is_symlink():
            raise AgentError("Import volume is unavailable")
        files = []
        for path in sorted(source.rglob("*")):
            if path.is_symlink():
                raise AgentError("Import source contains a symlink")
            if path.is_file():
                files.append({"path": path.relative_to(source).as_posix(), "size": path.stat().st_size, "sha256": file_digest(path)})
            elif not path.is_dir():
                raise AgentError("Import source contains a special file")
        if not files:
            raise AgentError("Import volume is empty")
        with self.artifact_open("PUT", payload["manifestUrl"], json.dumps({"files": files}).encode(), {"Content-Type": "application/json"}) as response:
            response.read()
        for item in files:
            path = scoped(source, item["path"])
            if path.stat().st_size != item["size"] or file_digest(path) != item["sha256"]:
                raise AgentError("Import source changed while publishing")
            url = payload["uploadBaseUrl"].rstrip("/") + "/" + urllib.parse.quote(item["path"], safe="/")
            with path.open("rb") as stream, self.artifact_open("PUT", url, stream, {"Content-Type": "application/octet-stream", "Content-Length": str(item["size"])}) as response:
                response.read()
        self.checkpoint({"phase": "publishing-import"})
        return self.complete_import(payload, start=True)

    def complete_import(self, payload, start=False):
        method = "POST" if start else "GET"
        while not self.stopping.is_set():
            with self.artifact_open(method, payload["completeUrl"], b"{}" if method == "POST" else None, {"Content-Type": "application/json"}) as response:
                result = json.load(response)
            if result.get("status") == "published":
                return result
            if result.get("status") == "failed":
                raise Rejected("Controller rejected the game-file import after integrity verification")
            if result.get("status") not in ("pending", "verifying"):
                raise AgentError("Controller returned an invalid import publication state")
            method = "POST" if result["status"] == "pending" else "GET"
            self.stopping.wait(3)
        raise AgentError("Agent stopped while the library import was being published")

    def dispatch(self, operation):
        kind, payload = operation.get("kind"), operation.get("payload", {})
        if not isinstance(kind, str) or not isinstance(payload, dict):
            raise AgentError("Invalid operation")
        if kind == "host.decommission":
            return self.decommission(operation)
        if kind == "credentials.rotate":
            pending = "ec_host_" + base64.urlsafe_b64encode(os.urandom(32)).rstrip(b"=").decode()
            self.config["pendingCredential"] = pending
            atomic_json(self.config_path, self.config)
            self.checkpoint({"phase": "rotating"})
            try:
                self.request("POST", "/api/agent/credentials", {"credential": pending})
            except AgentError:
                if not self.recover_credential():
                    raise
            self.config["credential"] = pending
            self.config.pop("pendingCredential", None)
            atomic_json(self.config_path, self.config)
            return {"rotated": True}
        if kind == "runtime.pull":
            image = payload.get("image")
            if not isinstance(image, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/:@-]{0,511}", image):
                raise AgentError("Invalid runtime image")
            self.docker.call("image", "pull", image, timeout=3600)
            image_id = self.docker.call("image", "inspect", "--format", "{{.Id}}", image).strip()
            aliases_file = self.root / "agent" / "runtime-images.json"
            aliases = json.loads(aliases_file.read_text()) if aliases_file.exists() else {}
            aliases[image] = image_id
            atomic_json(aliases_file, aliases)
            return {"image": image, "ready": True}
        if kind == "host.inspect":
            return self.inventory()
        if kind == "plutonium.prepare":
            return self.prepare_plutonium(payload)
        if kind == "rcon.status-batch":
            return self.rcon_status_batch(operation)
        if kind == "library.download":
            return self.library_download(payload)
        if kind == "library.remove":
            return self.library_remove(payload)
        if kind == "library.import":
            return self.library_import(payload)
        if kind == "library.inspect":
            return self.verify_library(self.library_dir(payload["game"], payload["libraryId"]))
        instance_id, generation = instance_identity(operation.get("instanceId"), operation.get("generation"))
        current_marker = self.instance_dir(instance_id, generation).parent / "current.json"
        if current_marker.exists():
            current = json.loads(current_marker.read_text())
            if kind != "instance.cancel-create" and (current["generation"] > generation or current["generation"] == generation and current.get("removed") and kind not in ("instance.inspect", "instance.remove")):
                raise AgentError("Instance generation has been superseded or removed")
        if kind.startswith("instance."):
            return self.lifecycle(kind, instance_id, generation, payload)
        if kind.startswith("files.") or kind == "logs.read":
            return self.file_operation(kind, instance_id, generation, payload, operation["id"])
        if kind == "rcon.execute":
            if payload["command"] not in ("status", "getstatus", "version"):
                self.checkpoint({"phase": "rcon-command"})
            else:
                with self.health_probe_gate:
                    return self.rcon(instance_id, generation, payload["command"])
            return self.rcon(instance_id, generation, payload["command"])
        raise AgentError("Unsupported operation kind")

    def execute(self, operation, report_running=False):
        row, _ = self.journal.accept(operation)
        if row["state"] == "unknown":
            return self.reconcile(operation, row)
        if row["state"] != "accepted":
            return row
        running = self.journal.update(operation["id"], "running")
        if report_running:
            try:
                self.report(running)
            except AgentError:
                return self.journal.update(operation["id"], "accepted", {"blocked": "controller-unavailable", "message": "Waiting for controller contact before executing the operation"})
        self.executing_id = operation["id"]
        self.operation_phase = "executing"
        try:
            result = self.dispatch(operation)
        except Held as err:
            if operation.get("kind") == "instance.remove" and "autoscaling" in operation.get("payload", {}):
                return self.journal.update(operation["id"], "failed", {"autoscalingInvalidated": True}, error=str(err))
            return self.journal.update(operation["id"], "accepted", {"blocked": "occupied" if "Players" in str(err) else "unknown", "message": str(err)})
        except AutoscalingInvalidated as err:
            return self.journal.update(operation["id"], "failed", {"autoscalingInvalidated": True}, error=str(err))
        except Exception as err:
            # The command may already have changed Docker or files. Never infer
            # failure means nothing happened, and never retry without inspection.
            diagnostic = None
            if isinstance(err, AgentError):
                message = str(err)
            else:
                message, details = unexpected_operation_error(err, self.operation_phase)
                diagnostic = {"unexpectedError": details}
            checkpoint = self.journal.get(operation["id"]).get("checkpoint")
            return self.journal.update(operation["id"], "failed" if isinstance(err, Rejected) or not checkpoint else "unknown", diagnostic, error=message)
        finally:
            self.executing_id = None
            self.operation_phase = None
        return self.journal.update(operation["id"], "running" if operation["kind"] == "host.decommission" else "completed", result)

    def recover_credential(self):
        pending = self.config.get("pendingCredential")
        if not pending:
            return False
        previous = self.config["credential"]
        self.config["credential"] = pending
        try:
            self.request("POST", "/api/agent/credentials", {"credential": pending})
        except AgentError:
            self.config["credential"] = previous
            try:
                self.request("POST", "/api/agent/credentials", {"credential": pending})
            except AgentError:
                return False
            self.config["credential"] = pending
        self.config.pop("pendingCredential", None)
        atomic_json(self.config_path, self.config)
        return True

    def reconcile_uncreated_rebuild(self, operation, row):
        """Settle removal only when the replacement create was never submitted.

        The write lane and instance lock exclude the original Python execution.
        A terminal journal receipt prevents any later replay of that command.
        """
        if row["state"] != "unknown":
            return row
        instance_id, generation = instance_identity(operation.get("instanceId"), operation.get("generation"))
        payload = operation.get("payload", {})
        next_generation = payload.get("nextGeneration", payload.get("newGeneration"))
        if type(next_generation) is not int or next_generation <= generation:
            return row
        previous = json.loads(row["result"]) if row.get("result") else {}
        if not isinstance(previous, dict):
            return row
        directory = scoped(self.root, f"instances/{instance_id}/{next_generation}")

        def unchanged_generation():
            try:
                marker = scoped(self.root, f"instances/{instance_id}/current.json")
                current = json.loads(marker.read_text())
            except FileNotFoundError:
                return True
            return isinstance(current, dict) and type(current.get("generation")) is int and current["generation"] == generation and current.get("removed", False) is False

        def both_missing():
            for target in (generation, next_generation):
                health = self.inspect(instance_id, target)
                if health.get("exists") is not False or health.get("running") is not False or health.get("state") != "missing":
                    return False
            return True

        if not unchanged_generation() or not both_missing():
            return row
        self.settle_bank(instance_id, generation, operation["id"])
        # Preparation can leave runtime/configuration files before create's
        # checkpoint. A fresh operation must not inherit those partial files.
        expected_spec = payload.get("spec") or {key: value for key, value in payload.items() if key not in ("nextGeneration", "newGeneration", "unhealthyRecovery")}
        try:
            saved_spec = json.loads(scoped(directory, "spec.json").read_text())
        except FileNotFoundError:
            pass
        else:
            if saved_spec != expected_spec:
                return row
        try:
            metadata = directory.lstat()
        except FileNotFoundError:
            pass
        else:
            if not stat_module.S_ISDIR(metadata.st_mode):
                return row
            shutil.rmtree(directory)
        # Also sync when a previous reconciliation removed the directory but
        # crashed before recording the terminal receipt.
        try:
            sync_parent(directory)
        except FileNotFoundError:
            pass
        if not unchanged_generation() or not both_missing():
            return row
        original_error = row.get("error")
        result = {**previous, "reconciled": True, "originalError": original_error,
                  "rebuildFailure": {"phase": "disrupting", "generation": generation, "nextGeneration": next_generation,
                                     "oldContainerMissing": True, "replacementContainerMissing": True, "replacementFilesRemoved": True}}
        message = "Old container removed; replacement creation failed."
        if original_error:
            message += " " + original_error
        return self.journal.update(operation["id"], "failed", result, error=message)

    def reconcile(self, operation, row):
        """Prove desired state with immutable ownership; never resubmit a command."""
        kind, payload = operation["kind"], operation.get("payload", {})
        instance_id, generation = operation.get("instanceId"), operation.get("generation")
        checkpoint = json.loads(row["checkpoint"]) if row.get("checkpoint") else {}
        try:
            if kind == "instance.rebuild" and checkpoint.get("phase") == "disrupting":
                return self.reconcile_uncreated_rebuild(operation, row)
            if kind == "instance.remove" and "autoscaling" in payload and not checkpoint:
                return self.journal.update(operation["id"], "failed", {"autoscalingInvalidated": True}, error="Interrupted autoscaling removal requires a new decision")
            if kind == "instance.cancel-create":
                return self.journal.update(operation["id"], "completed", self.cancel_creation(instance_id, generation, payload, operation["id"]))
            if kind == "instance.create":
                marker = self.instance_dir(instance_id, generation).parent / "current.json"
                if marker.exists():
                    current = json.loads(marker.read_text())
                    if current["generation"] > generation or current["generation"] == generation and current.get("removed"):
                        return self.journal.update(operation["id"], "failed", error="Creation was cancelled or superseded")
            if kind == "host.decommission":
                self.executing_id = operation["id"]
                try:
                    result = self.decommission(operation)
                    return self.journal.update(operation["id"], "running", result)
                finally:
                    self.executing_id = None
            if kind == "credentials.rotate":
                if not self.config.get("pendingCredential") or self.recover_credential():
                    return self.journal.update(operation["id"], "completed", {"rotated": True, "reconciled": True})
                return row
            if kind in ("runtime.pull", "plutonium.prepare"):
                # Docker resumes already downloaded layers. Pull again under
                # the same operation ID: an existing tag can still refer to an
                # older image and cannot prove an interrupted pull completed.
                self.journal.update(operation["id"], "running")
                try:
                    result = self.dispatch(operation)
                    return self.journal.update(operation["id"], "completed", {**result, "reconciled": True})
                except Held as err:
                    return self.journal.update(operation["id"], "accepted", {"blocked": "host-busy", "message": str(err)})
                except Exception:
                    return self.journal.update(operation["id"], "failed", error="Interrupted runtime download could not be resumed")
            if kind == "library.import":
                try:
                    with self.artifact_open("GET", payload["completeUrl"]) as response:
                        status = json.load(response)
                    if status.get("status") == "published":
                        return self.journal.update(operation["id"], "completed", status)
                    if checkpoint.get("phase") == "publishing-import" or status.get("status") == "verifying":
                        return self.journal.update(operation["id"], "completed", self.complete_import(payload))
                    # Import uploads use an immutable import ID and per-file
                    # checksums. Reconcile their staged files after interruption.
                    self.journal.update(operation["id"], "running")
                    self.executing_id = operation["id"]
                    return self.journal.update(operation["id"], "completed", self.library_import(payload))
                except Rejected as err:
                    return self.journal.update(operation["id"], "failed", error=str(err))
                except Exception:
                    return self.journal.update(operation["id"], "unknown", error="Import reconciliation is waiting for controller contact")
                finally:
                    self.executing_id = None
            if kind == "library.remove":
                return self.journal.update(operation["id"], "completed", self.library_remove(payload))
            if kind == "library.download":
                # This is a resumable immutable transfer, not a repeated
                # container command. Full files skip transfer; verification
                # always starts again so interrupted reads cannot prove ready.
                self.journal.update(operation["id"], "running")
                try:
                    return self.journal.update(operation["id"], "completed", self.library_download(payload))
                except Exception:
                    return self.journal.update(operation["id"], "failed", error="Interrupted library download could not be resumed")
            if kind == "files.stage":
                return self.journal.update(operation["id"], "completed", self.stage(instance_id, generation, payload, operation["id"]))
            if self.read_operation(operation):
                return self.journal.update(operation["id"], "completed", self.dispatch(operation))
            if kind in ("files.activate", "files.rollback") and checkpoint:
                stage = self.instance_dir(instance_id, generation) / "stages" / identifier(payload["stageId"]) / "manifest.json"
                saved = json.loads(stage.read_text())
                expected = "activated" if kind == "files.activate" else "rolled-back"
                if saved["state"] == expected:
                    health = self.inspect(instance_id, generation)
                    restarted = health["running"] and health.get("containerId") == checkpoint.get("containerId") and bool(checkpoint.get("startedAt")) and health.get("startedAt") != checkpoint["startedAt"]
                    if not payload.get("restart", True) or restarted:
                        return self.journal.update(operation["id"], "completed", {"activated": expected == "activated", "rolledBack": expected == "rolled-back", "before": saved["before"], "reconciled": True})
                return row
            if not kind.startswith("instance.") or not checkpoint:
                return row
            observed_generation = payload.get("nextGeneration", payload.get("newGeneration")) if kind == "instance.rebuild" else generation
            health = self.inspect(instance_id, observed_generation)
            before = []
            if payload.get("stageId"):
                if kind == "instance.rebuild":
                    # The creating checkpoint retains snapshots after old storage is deleted.
                    if not isinstance(checkpoint.get("before"), list):
                        return row
                    before = checkpoint["before"]
                else:
                    stage = self.instance_dir(instance_id, generation) / "stages" / identifier(payload["stageId"]) / "manifest.json"
                    saved = json.loads(stage.read_text())
                    if saved["state"] != "activated":
                        return row
                    before = saved["before"]
            proven = False
            if kind in ("instance.create", "instance.rebuild"):
                proven = health["running"] and health.get("creationOperation") == operation["id"] and health.get("specSha256") == checkpoint.get("specSha256")
                if proven:
                    directory = self.instance_dir(instance_id, observed_generation)
                    atomic_json(directory.parent / "current.json", {"generation": observed_generation})
                    if kind == "instance.rebuild" and not self.inspect(instance_id, generation)["exists"]:
                        self.settle_bank(instance_id, generation, operation["id"])
                        old = self.instance_dir(instance_id, generation)
                        if old.exists():
                            shutil.rmtree(old)
            elif kind in ("instance.start", "instance.stop"):
                proven = health.get("containerId") == checkpoint.get("containerId") and (health["running"] if kind == "instance.start" else health["state"] in ("stopped", "exited", "created", "dead"))
                if kind == "instance.start":
                    proven = proven and checkpoint.get("phase") == "starting"
                    if checkpoint.get("restartRequired"):
                        proven = proven and bool(checkpoint.get("startedAt")) and bool(health.get("startedAt")) and health["startedAt"] != checkpoint["startedAt"]
            elif kind == "instance.restart":
                proven = health["running"] and health.get("containerId") == checkpoint.get("containerId") and bool(checkpoint.get("startedAt")) and bool(health.get("startedAt")) and health["startedAt"] != checkpoint["startedAt"]
            elif kind == "instance.remove":
                proven = not health["exists"]
                if proven:
                    self.settle_bank(instance_id, generation, operation["id"])
                    directory = self.instance_dir(instance_id, generation)
                    if directory.exists():
                        shutil.rmtree(directory)
                    self.mark_removed(instance_id, generation)
            if proven:
                return self.journal.update(operation["id"], "completed", {**health, "before": before, "reconciled": True})
        except (AgentError, OSError, ValueError, KeyError):
            pass
        return row

    def report(self, row, journal=None):
        payload = {"state": row["state"]}
        if row.get("result"):
            payload["result"] = json.loads(row["result"])
            if row["state"] == "accepted" and isinstance(payload["result"], dict) and payload["result"].get("blocked"):
                # The controller sees a durable running/waiting operation;
                # locally accepted means its mutation gate can safely retry.
                payload["state"] = "running"
        if row.get("error"):
            payload["error"] = row["error"]
        response = self.request("POST", "/api/agent/operations/" + urllib.parse.quote(row["id"], safe=""), payload)
        confirmed = response.get("operation") if isinstance(response, dict) else None
        if row["state"] == "completed" and isinstance(confirmed, dict) and confirmed.get("id") == row["id"] and confirmed.get("state") == "completed":
            (journal if journal is not None else self.journal).confirm_completed(row)
        return response

    def heartbeat(self):
        response = self.request("POST", "/api/agent/heartbeat", {"inventory": self.inventory(), "operationId": self.current_operation})
        # Only the heartbeat thread replaces this immutable snapshot; it never
        # mutates the operation journal or races Docker cleanup against create.
        self.cancelled_creations = frozenset(value for value in response.get("cancelledCreations", []) if isinstance(value, str))
        self.defer_plutonium_preparation = response.get("deferPlutoniumPreparation") is True
        keep_days = response.get("observationKeepDays")
        self.observation_keep_days = keep_days if valid_observation_keep_days(keep_days) else 0
        host = response.get("host", {})
        if isinstance(host.get("portMin"), int) and isinstance(host.get("portMax"), int):
            self.config["portRange"] = {"min": host["portMin"], "max": host["portMax"]}
        # Written only after the controller accepted this process's authenticated
        # heartbeat. Service liveness alone cannot prove an update works.
        self.record_health()

    def heartbeat_loop(self):
        while not self.stopping.wait(15):
            try:
                self.heartbeat()
            except (AgentError, OSError, ValueError):
                pass
            except Exception as err:
                print(f"Heartbeat recovered from {type(err).__name__}", file=sys.stderr)

    @staticmethod
    def read_operation(operation):
        return read_operation(operation)

    def instance_lock(self, operation):
        key = operation.get("instanceId")
        return self.instance_locks.setdefault(key, threading.RLock()) if key else contextlib.nullcontext()

    @contextlib.contextmanager
    def observation_lock(self, operation, health=False):
        lock = self.instance_lock(operation)
        if not health or not operation.get("instanceId"):
            with lock:
                yield
            return
        if not lock.acquire(timeout=HEALTH_LOCK_TIMEOUT_SECONDS):
            raise AgentError("Health check timed out waiting for the instance operation")
        try:
            yield
        finally:
            lock.release()

    @staticmethod
    def health_operation(operation):
        return operation.get("kind") in ("instance.inspect", "rcon.status-batch") or (operation.get("kind") == "rcon.execute" and operation.get("payload", {}).get("command") in ("status", "getstatus", "version"))

    def read_loop(self, lane="read", lane_name=None):
        # Each lane owns its SQLite connection and never writes executing_id,
        # checkpoints, or current_operation used by the mutation state machine.
        lane_name = lane_name or ("health-1" if lane == "health" else "read")
        self.lane_started(lane_name)
        journal = Journal(self.root / "agent" / "operations.sqlite3", recover=False)
        next_prune = time.monotonic()
        try:
            while not self.stopping.is_set():
                try:
                    if self.update_pending():
                        self.stopping.wait(3)
                        continue
                    if lane == "read" and time.monotonic() >= next_prune:
                        next_prune = time.monotonic() + 60
                        with self.activity(read_only=True) as acquired:
                            if acquired and not self.update_pending():
                                if journal.prune_observations(self.observation_keep_days) == 500:
                                    next_prune = time.monotonic() + 15
                    response = self.request("GET", f"/api/agent/operations?lane={lane}&wait=20")
                    self.lane_polled(lane_name)
                    handled = False
                    self.lane_busy(lane_name, bool(response.get("operations")))
                    for operation in response.get("operations", []):
                        if self.stopping.is_set() or not self.read_operation(operation):
                            continue
                        if lane == "health" and not self.health_operation(operation):
                            continue
                        with self.read_claim_lock:
                            if operation["id"] in self.read_claims:
                                continue
                            self.read_claims.add(operation["id"])
                        try:
                            with self.activity(read_only=True) as acquired:
                                if not acquired or self.update_pending():
                                    continue
                                row, _ = journal.accept(operation)
                                if row["state"] in ("accepted", "unknown", "running"):
                                    self.report(journal.update(operation["id"], "running"), journal=journal)
                                    try:
                                        with self.read_gate.read(), self.observation_lock(operation, health=lane == "health"):
                                            result = self.dispatch(operation)
                                            row = journal.update(operation["id"], "completed", result)
                                    except Exception as err:
                                        row = journal.update(operation["id"], "failed", error=str(err) if isinstance(err, AgentError) else "Observation unavailable")
                                # Receipt delivery can take seconds; it must not
                                # hold the instance lock needed by health checks.
                                self.report(row, journal=journal)
                                handled = True
                        finally:
                            with self.read_claim_lock:
                                self.read_claims.discard(operation["id"])
                    if response.get("operations") and not handled:
                        # A legacy controller may return another lane's work or
                        # a read already claimed locally. Avoid a busy poll loop.
                        self.stopping.wait(0.25)
                except (AgentError, OSError, ValueError, KeyError, sqlite3.Error):
                    self.stopping.wait(3)
                except Exception as err:
                    print(f"{lane} lane recovered from {type(err).__name__}", file=sys.stderr)
                    self.stopping.wait(3)
                finally:
                    self.lane_busy(lane_name, False)
        finally:
            journal.close()

    def poll_operations(self):
        if self.stopping.is_set() or self.update_pending():
            return
        # Check both update and mutation locks before delivery, release them
        # during the long poll, and recheck immediately before accepting work.
        with self.activity() as acquired:
            if not acquired:
                return
        # Never hold the updater lock while waiting for delivery.
        response = self.request("GET", "/api/agent/operations?lane=write&wait=20")
        self.lane_polled("write")
        self.lane_busy("write", bool(response.get("operations")))
        with self.activity() as acquired:
            if not acquired or self.stopping.is_set() or self.update_pending():
                return
            if self.config.get("pendingCredential"):
                self.recover_credential()
            completed = True
            for operation in response.get("operations", []):
                if self.stopping.is_set():
                    break
                if self.read_operation(operation):
                    continue
                self.current_operation = operation["id"]
                # Global changes wait for observations; instance changes only
                # block observations for their own immutable instance identity.
                guard = self.read_gate.write() if operation["kind"] in ("host.decommission", "credentials.rotate") else contextlib.nullcontext()
                try:
                    with guard, self.instance_lock(operation):
                        accepted, _ = self.journal.accept(operation)
                        self.report(accepted)
                        result = self.execute(operation, report_running=True)
                        self.report(result)
                        completed = completed and result["state"] in ("completed", "failed")
                finally:
                    self.current_operation = None
            return completed

    def run(self):
        self.lane_started("write")
        # Obtain the configured host port range before accepting any deployment.
        # Enrollment's default values must never race the first heartbeat.
        while not self.stopping.is_set():
            try:
                with self.activity() as acquired:
                    if not acquired:
                        self.stopping.wait(3)
                        continue
                    if self.config.get("pendingCredential"):
                        self.recover_credential()
                self.heartbeat()
                break
            except (AgentError, OSError, ValueError):
                self.stopping.wait(3)
            except Exception as err:
                print(f"Initial heartbeat recovered from {type(err).__name__}", file=sys.stderr)
                self.stopping.wait(3)
        heartbeats = threading.Thread(target=self.heartbeat_loop, daemon=True)
        heartbeats.start()
        # Preserve two concurrent health checks even while files are being read.
        readers = [threading.Thread(target=self.read_loop, args=(lane, name), daemon=True)
                   for name, lane in (("health-1", "health"), ("health-2", "health"), ("read", "read"))]
        for reader in readers:
            reader.start()
        while not self.stopping.is_set():
            completed = False
            try:
                completed = self.poll_operations()
            except (AgentError, OSError, ValueError, KeyError):
                # Logs deliberately omit URLs, tokens, payloads and raw errors.
                print("Host controller contact unavailable; existing containers continue running", file=sys.stderr)
            except Exception as err:
                print(f"Write lane recovered from {type(err).__name__}", file=sys.stderr)
            finally:
                self.lane_busy("write", False)
            self.stopping.wait(0.25 if completed else 3)
        for reader in readers:
            reader.join(timeout=2)
        heartbeats.join(timeout=2)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config", default="/etc/electric-cherry/agent.json")
    parser.add_argument("--enroll", action="store_true")
    parser.add_argument("--controller")
    parser.add_argument("--token-file")
    parser.add_argument("--root", default="/var/lib/electric-cherry")
    parser.add_argument("--force-enroll", action="store_true")
    parser.add_argument("--uninstall", metavar="REQUEST", help="Run the installed host removal service")
    parser.add_argument("--version", action="version", version=VERSION)
    args = parser.parse_args()
    if args.uninstall:
        if os.geteuid() != 0 or args.enroll:
            raise AgentError("Host uninstall must run as the dedicated root service")
        Uninstaller(args.uninstall).run()
        return
    if args.enroll:
        if not args.controller or not args.token_file:
            parser.error("Enrollment needs --controller and --token-file")
        if Path(args.config).exists() and not args.force_enroll:
            existing = json.loads(Path(args.config).read_text())
            if existing.get("credential"):
                if existing.get("controller", "").rstrip("/") != args.controller.rstrip("/") or Path(existing.get("root", "")).resolve() != Path(args.root).resolve():
                    raise AgentError("Existing enrollment uses a different controller or data directory")
                print("Host is already enrolled; retaining existing credentials")
                return
        agent = Agent({"controller": args.controller.rstrip("/"), "root": args.root})
        token = Path(args.token_file).read_text().strip()
        enrolled = agent.request("POST", "/api/agent/enroll", {"token": token, "inventory": agent.inventory()}, credential=False)
        if not enrolled.get("hostId") or not enrolled.get("credential"):
            raise AgentError("Controller returned invalid enrollment credentials")
        atomic_json(args.config, {"controller": args.controller.rstrip("/"), "root": args.root, "hostId": enrolled["hostId"], "credential": enrolled["credential"], "portRange": enrolled.get("portRange", {"min": 28960, "max": 29060})})
        agent.journal.close()
        print("Host enrollment completed")
        return
    config = json.loads(Path(args.config).read_text())
    config["configPath"] = args.config
    lock_root = Path(config.get("root", "/var/lib/electric-cherry")) / "agent"
    lock_root.mkdir(parents=True, exist_ok=True)
    lock = open(lock_root / "process.lock", "a")
    try:
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        raise AgentError("Another host agent already owns this data directory") from None
    agent = Agent(config)
    for sig in (signal.SIGINT, signal.SIGTERM):
        signal.signal(sig, lambda *_: agent.stopping.set())
    try:
        agent.run()
    finally:
        agent.journal.close()
        lock.close()


if __name__ == "__main__":
    try:
        main()
    except (AgentError, OSError, ValueError, KeyError):
        print("Electric Cherry host agent failed; check configuration and controller availability", file=sys.stderr)
        sys.exit(1)