#!/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
import hashlib
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 urllib.error
import urllib.parse
import urllib.request
import uuid

VERSION = "1.4.1"
LABEL = "io.electric-cherry"
MAX_FILE = 32 * 1024 * 1024
LIBRARY_DOWNLOAD_WORKERS = 4
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."""


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_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)


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()


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)")
        if "checkpoint" not in {row[1] for row in self.db.execute("PRAGMA table_info(operations)")}:
            self.db.execute("ALTER TABLE operations ADD COLUMN checkpoint TEXT")
        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=? 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 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.instance_locks = {}
        self.read_gate = ReadWriteGate()
        self.read_claim_lock = threading.Lock()
        self.read_claims = set()
        self.cancelled_creations = frozenset()
        self.library_progress = {}
        self.library_transfers = {}
        self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
        self.executing_id = None

    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("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):
            raise AgentError("Controller request failed") from None

    def inventory(self):
        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())})
        for marker in (self.root / "libraries").glob("*/*/manifest.json"):
            # 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 = json.loads(marker.read_text())
                # 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 (self.root / "libraries").glob("*/.*.partial/source.json"):
            try:
                value = json.loads(marker.read_text())
                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 = []
        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):
                        images.extend(alias for alias, image_id in json.loads(aliases.read_text()).items() if image_id in image_ids)
        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 = ["instance.cancel-create", "operations.read-lane"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
        return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, "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)), "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):
        health = self.inspect(instance_id, generation)
        if health["state"] in ("missing", "stopped", "exited", "created", "dead"):
            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(self, instance_id, generation, command):
        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
        chunks = []
        with socket.socket(family, socket.SOCK_DGRAM) as sock:
            sock.settimeout(3)
            try:
                sock.connect((host, int(saved["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:
                    chunks.append(sock.recv(65535))
                    sock.settimeout(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.
        self.docker.call("image", "inspect", image)
        files = payload.get("files", [])
        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, port, bind, env

    def create(self, instance_id, generation, payload):
        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
        game, library, image, port, bind, env = self.validate_spec(payload)
        app = directory / "app"
        app.mkdir(parents=True, exist_ok=True)
        self.make_writable(app, directory=True)
        self.write_initial_files(app, payload.get("files", []))
        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())
        args = ["container", "create", "--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)
        self.check_creation_cancellation()
        self.checkpoint({"phase": "creating", "generation": generation, "specSha256": spec_sha})
        self.docker.call(*args, timeout=120)
        self.check_creation_cancellation()
        self.docker.call("container", "start", name)
        self.check_creation_cancellation()
        atomic_json(current_marker, {"generation": generation})
        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 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):
        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 lifecycle(self, kind, instance_id, generation, payload):
        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"):
            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")}
                self.validate_spec(spec)
            self.safe_to_disrupt(instance_id, generation)
            self.checkpoint({"phase": "disrupting", "containerId": health.get("containerId"), "startedAt": health.get("startedAt")})
            if payload.get("stageId") and kind == "instance.restart":
                before = self.activate(instance_id, generation, payload["stageId"])
            if kind in ("instance.remove", "instance.rebuild"):
                if health["exists"]:
                    self.docker.call("container", "stop", "--time", "30", name)
                    self.docker.call("container", "rm", name)
                if kind == "instance.rebuild":
                    result = self.create(instance_id, next_generation, spec)
                    shutil.rmtree(self.instance_dir(instance_id, generation), ignore_errors=False)
                    return result
                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"):
                    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):
        """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"))
        marker = self.instance_dir(instance_id, generation).parent / "current.json"
        if marker.exists() and json.loads(marker.read_text())["generation"] > generation:
            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 = self.inspect(instance_id, generation)
        if health["exists"] and health.get("creationOperation") != create_id:
            raise Rejected("Container belongs to a different creation operation")
        self.checkpoint({"phase": "cancelling-creation", "createOperationId": create_id})
        # Fence before deleting: a crash must never permit the old creation to run again.
        self.mark_removed(instance_id, generation)
        if health["exists"]:
            self.docker.call("container", "rm", "--force", "--volumes", self.container_name(instance_id, generation))
        after = self.inspect(instance_id, generation)
        if after["exists"]:
            raise AgentError("Cancelled creation container still exists")
        directory = self.instance_dir(instance_id, generation)
        if directory.exists():
            shutil.rmtree(directory)
        return {**after, "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
            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 activate(self, instance_id, generation, stage_id):
        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.
        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)
        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())
            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}

    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_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 == "library.download":
            return self.library_download(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 current["generation"] > generation or current["generation"] == generation and current.get("removed") and kind not in ("instance.inspect", "instance.remove", "instance.cancel-create"):
                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"})
            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"]
        try:
            result = self.dispatch(operation)
        except Held as err:
            return self.journal.update(operation["id"], "accepted", {"blocked": "occupied" if "Players" in str(err) else "unknown", "message": 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.
            message = str(err) if isinstance(err, AgentError) else "Operation could not be completed; inspect its current state"
            checkpoint = self.journal.get(operation["id"]).get("checkpoint")
            return self.journal.update(operation["id"], "failed" if isinstance(err, Rejected) or not checkpoint else "unknown", error=message)
        finally:
            self.executing_id = 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(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.cancel-create":
                return self.journal.update(operation["id"], "completed", self.cancel_creation(instance_id, generation, payload))
            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 == "runtime.pull":
                # 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 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.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 kind in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "logs.read") or kind == "rcon.execute" and payload.get("command") in ("status", "getstatus", "version"):
                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"):
                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"]:
                        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:
                    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):
        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"]
        return self.request("POST", "/api/agent/operations/" + urllib.parse.quote(row["id"], safe=""), payload)

    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))
        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

    @staticmethod
    def read_operation(operation):
        return operation.get("kind") in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "logs.read") or (operation.get("kind") == "rcon.execute" and operation.get("payload", {}).get("command") in ("status", "getstatus", "version"))

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

    def read_loop(self):
        # Each lane owns its SQLite connection and never writes executing_id,
        # checkpoints, or current_operation used by the mutation state machine.
        journal = Journal(self.root / "agent" / "operations.sqlite3", recover=False)
        try:
            while not self.stopping.is_set():
                try:
                    if self.update_pending():
                        self.stopping.wait(3)
                        continue
                    response = self.request("GET", "/api/agent/operations?lane=read&wait=20")
                    for operation in response.get("operations", []):
                        if self.stopping.is_set() or not self.read_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
                                with self.read_gate.read(), self.instance_lock(operation):
                                    row, _ = journal.accept(operation)
                                    if row["state"] in ("accepted", "unknown", "running"):
                                        self.report(journal.update(operation["id"], "running"))
                                        try:
                                            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")
                                    self.report(row)
                        finally:
                            with self.read_claim_lock:
                                self.read_claims.discard(operation["id"])
                except (AgentError, OSError, ValueError, KeyError, sqlite3.Error):
                    self.stopping.wait(3)
        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")
        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):
        # 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)
        heartbeats = threading.Thread(target=self.heartbeat_loop, daemon=True)
        heartbeats.start()
        readers = [threading.Thread(target=self.read_loop, daemon=True) for _ in range(2)]
        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)
            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)