#!/usr/bin/python3
"""Opt-in, inventory-bound deployment of the Maximizer host supervisor."""

import argparse
import base64
import configparser
import contextlib
import fcntl
import hashlib
import json
import os
from pathlib import Path
import pwd
import re
import signal
import stat
import subprocess
import sys
import tempfile
import time


DROPIN = "90-maximizer-host-supervision.conf"
SYSTEMCTL = "/usr/bin/systemctl"
HOOKS = ("ExecCondition", "ExecStartPre", "ExecStartPost", "ExecStop", "ExecStopPost", "ExecReload")
DEPENDENCY_PROPERTIES = (
    "Id", "LoadState", "ActiveState", "NeedDaemonReload", "Job", "StopWhenUnneeded",
    "Wants", "Requires", "Requisite", "BindsTo", "Upholds", "UpheldBy", "Conflicts", "ConflictedBy",
    "RequiredBy", "RequisiteOf", "BoundBy", "ConsistsOf", "PropagatesStopTo", "OnFailure", "OnSuccess",
    "FailureAction", "SuccessAction", "StartLimitAction",
)
PROPERTIES = (
    "Id", "LoadState", "FragmentPath", "DropInPaths", "NeedDaemonReload",
    "ActiveState", "SubState", "MainPID", "ControlGroup", "User", "DynamicUser",
    "WorkingDirectory", "Type", "PIDFile", "RemainAfterExit", "KillMode",
    "KillSignal", "SendSIGKILL", "WatchdogUSec", "RuntimeDirectory",
    "RootDirectory", "RootImage", "RestartPreventExitStatus", "RestartForceExitStatus",
    "FailureAction", "SuccessAction", "StartLimitAction", "OnFailure", "OnSuccess",
    "PropagatesStopTo", "BoundBy", "ConsistsOf", "RequiredBy", "TriggeredBy",
    "Restart", "RestartUSec", "TimeoutStopUSec", "StartLimitIntervalUSec", "ExecStart",
    "Conflicts", "ConflictedBy", "Wants", "Requires", "Requisite", "RequisiteOf", "BindsTo",
    "Upholds", "UpheldBy", "Job", "DefaultDependencies", "StopWhenUnneeded", "Slice",
) + HOOKS
LIMITS = """Operational limits:
  audit is the default. install, uninstall and recover are dry-run unless --apply.
  Supply an exact, private JSON deployment inventory and an explicit --state-dir.
  Only the current nonroot user's manager, or root accessing the system manager
  with an existing nonroot User=, is supported. No sudo, guessed UID, remote bus,
  cross-manager migration, enable/disable/mask, linger, targets or ordering edits.
  Direct native ELF Sunshine simple/exec units only; all fragments/drop-ins must be
  inventoried by SHA-256. Shell wrappers, forking/notify units, stop/post hooks,
  socket activation, stop propagation and mixed session scripts require manual
  integration. Only the exact /bin/sleep 5 pre-hook is supported; it is retained
  unless explicitly approved for removal. Original literal argv is an array;
  systemd variable/specifier expansion in the old ExecStart is not supported.
  The canonical binary must contain the NUL-terminated MAXIMIZER_SUPERVISOR_FD
  native bootstrap-gate marker. This read-only compatibility screen is not binary
  authentication; duplicate legacy binaries need not contain it. No probe is run.
  Direct application-service dependencies/conflicts are refused, even between
  inventory units. Only settled basic/sysinit targets, required mounts and the
  reported slice are admitted, with their transitive pull-ins already active and
  job-free; shutdown/umount conflicts and propagated stop recipients must be inactive.
  Dependency effects are rechecked immediately before every unit start/stop.
  Simple direct-launch .desktop entries only. Autostarts become Hidden=true;
  menu/terminal entries delegate start to the canonical manager and supervisor.
  Global desktop files are never modified: use same-basename per-user overrides.
  Inventory completeness and all configured TCP/UDP listen ports must be reviewed
  by the operator. Unknown socket/process owners are refused, never killed.
  No auth, certificates, apps, databases, boot, NIC, LightDM, NetworkManager,
  DHCP or udev edits. No package maintainer scripts are invoked.
  Apply interrupts streaming. Files and active/inactive states are journaled
  before stopping exact inventory units. No start/stop outside those units.
  Uninstall stops the supervisor owner first and restores only unchanged managed
  files. Drift is refused, including changed source units after package upgrades.
  recover --apply reverses an interrupted transaction; it never forces drift.
  Backups remain private and are never deleted or rebased. A changed inventory
  requires uninstall and a fresh state directory. Keep old recovery records.
  Nonroot invokers must own every managed preimage and be able to restore its GID
  using their current effective/supplementary groups; no elevation is attempted.
  Directory creation intent is journaled before mkdir. A surviving path without
  a checkpointed identity blocks recovery until operator review; keep the journal.
  Linux /proc, Python 3 and a reachable local systemd manager are required.
  Run in a maintenance window: no concurrent package, session or unit changes.
  Audit proves only the declared scope at that instant, not all future launchers.
"""


class Refusal(Exception):
    pass


def require(condition, message):
    if not condition:
        raise Refusal(message)


def keys(value, required, optional=()):
    require(isinstance(value, dict), "Expected an inventory object")
    require(set(required) <= value.keys() and value.keys() <= set(required) | set(optional),
            "Missing or unsupported inventory fields; manual integration required")


def absolute(value):
    require(isinstance(value, str) and value.startswith("/") and
            not any(ord(c) < 32 for c in value) and
            str(Path(value)) == value and ".." not in Path(value).parts,
            "Expected a normalized absolute path")
    return Path(value)


def unit_name(value):
    require(isinstance(value, str) and re.fullmatch(r"[A-Za-z0-9_][A-Za-z0-9_.-]*\.service", value),
            "An exact non-template .service name is required (no aliases or patterns)")
    return value


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


def encoded(value):
    return (json.dumps(value, sort_keys=True, indent=2, ensure_ascii=True) + "\n").encode()


def safe_ancestors(path):
    for parent in reversed(path.parents):
        try:
            info = parent.lstat()
        except FileNotFoundError:
            continue
        require(stat.S_ISDIR(info.st_mode), "Symlink/non-directory ancestor refused: " + str(parent))
        require(not info.st_mode & 0o022, "Writable shared ancestor refused: " + str(parent))


def snapshot(path):
    path = absolute(str(path))
    safe_ancestors(path)
    try:
        fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
    except FileNotFoundError:
        return None
    with os.fdopen(fd, "rb") as source:
        info = os.fstat(source.fileno())
        require(stat.S_ISREG(info.st_mode) and info.st_nlink == 1,
                "Only regular, unlinked files are supported: " + str(path))
        require(info.st_size <= 4 * 1024 * 1024, "Oversized configuration refused: " + str(path))
        data = source.read()
    return image(data, stat.S_IMODE(info.st_mode), info.st_uid, info.st_gid)


def image(data, mode, uid, gid):
    return {"data": base64.b64encode(data).decode("ascii"), "sha256": digest(data),
            "mode": mode, "uid": uid, "gid": gid}


def contents(value):
    data = base64.b64decode(value["data"], validate=True)
    require(digest(data) == value["sha256"], "Corrupt recovery image; refusing restoration")
    return data


def check_restorable(images):
    uid = os.geteuid()
    if uid == 0:
        return
    groups = set(os.getgroups()) | {os.getegid()}
    for path, value in images.items():
        require(value is None or (value["uid"] == uid and value["gid"] in groups),
                "Managed file has non-restorable UID/GID; refusing without elevation: " + str(path))


def fsync_directory(path):
    fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
    try:
        os.fsync(fd)
    finally:
        os.close(fd)


def replace_file(path, expected, desired):
    require(snapshot(path) == expected, "File drift; refusing write: " + str(path))
    if desired == expected:
        return
    if desired is None:
        path.unlink()
    else:
        fd, temporary = tempfile.mkstemp(prefix=".maximizer-", dir=path.parent)
        try:
            with os.fdopen(fd, "wb") as output:
                os.fchown(output.fileno(), desired["uid"], desired["gid"])
                os.fchmod(output.fileno(), desired["mode"])
                output.write(contents(desired))
                output.flush()
                os.fsync(output.fileno())
            require(snapshot(path) == expected, "Concurrent file drift: " + str(path))
            os.replace(temporary, path)
        finally:
            if os.path.exists(temporary):
                os.unlink(temporary)
    fsync_directory(path.parent)


def read_json(path, private=False):
    value = snapshot(path)
    require(value is not None, "Required file is absent: " + str(path))
    if private:
        require(value["mode"] == 0o600 and value["uid"] == os.geteuid(),
                "Inventory/journal must be mode 0600 and owned by the invoking UID")
    try:
        def unique(pairs):
            result = {}
            for key, item in pairs:
                require(key not in result, "Duplicate JSON field refused")
                result[key] = item
            return result
        return json.loads(contents(value), object_pairs_hook=unique)
    except (ValueError, UnicodeError):
        raise Refusal("Invalid JSON (contents suppressed)") from None


def literal_words(text):
    """A deliberately small systemd/desktop literal subset, not shell syntax."""
    require(not any(c in text for c in "\\$%\n\r"),
            "Escapes/expansions in launch directives require manual integration")
    words = []
    position = 0
    while position < len(text):
        if text[position].isspace():
            position += 1
            continue
        if text[position] in "\"'":
            quote = text[position]
            end = text.find(quote, position + 1)
            require(end >= 0 and (end + 1 == len(text) or text[end + 1].isspace()),
                    "Unsupported quoting in launch directive; manual integration required")
            words.append(text[position + 1:end])
            position = end + 1
        else:
            end = position
            while end < len(text) and not text[end].isspace():
                end += 1
            word = text[position:end]
            require(not any(c in word for c in "\"'"), "Unsupported partial quoting in launch directive")
            words.append(word)
            position = end
    require(";" not in words, "Multiple launch commands require manual integration")
    return words


def quote_systemd(value, specifier=False):
    require(not any(ord(c) < 32 for c in value), "Control characters in generated directive refused")
    value = value.replace("\\", "\\\\").replace('"', '\\"').replace("$", "$$")
    if not specifier:
        value = value.replace("%", "%%")
    return '"' + value + '"'


def quote_desktop(value):
    require(not any(ord(c) < 32 for c in value), "Control characters in desktop directive refused")
    # Desktop entries have an outer string escape layer and an Exec quoting layer.
    value = value.replace("%", "%%")
    for character in ("\\", '"', "`", "$"):
        value = value.replace(character, "\\" + character)
    return '"' + value.replace("\\", "\\\\") + '"'


class Manager:
    def __init__(self, manager, apply=False):
        require(manager in ("user", "system"), "Explicit manager user|system is required")
        self.manager = manager
        self.apply = apply

    def call(self, action, unit=None):
        require(action in ("show", "show-dependencies", "daemon-reload", "start", "stop"), "Unsupported manager operation")
        require(action in ("show", "show-dependencies") or self.apply, "Activation requires --apply")
        command = [SYSTEMCTL, "--" + self.manager, "--no-pager", "--no-ask-password"]
        if action == "show":
            command += ["show", "--all", "--property=" + ",".join(PROPERTIES), unit_name(unit)]
        elif action == "show-dependencies":
            require(isinstance(unit, str) and re.fullmatch(
                r"(?:[A-Za-z0-9_.:@-]|\\x[0-9A-Fa-f]{2})+\.(?:service|target|mount|automount|swap|socket|device|path|timer|scope|slice)", unit),
                "Dependency inspection requires an exact unit name")
            command += ["show", "--all", "--property=" + ",".join(DEPENDENCY_PROPERTIES), "--", unit]
        elif action == "daemon-reload":
            command += [action]
        else:
            unit_name(unit)
            check_dependencies(self, self.show(unit))
            command += ["--job-mode=fail", action, unit_name(unit)]
        try:
            result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                    timeout=20, check=False, env={**os.environ, "LC_ALL": "C"})
        except subprocess.TimeoutExpired:
            raise Refusal("Bounded systemctl operation timed out; pending jobs may require recovery") from None
        require(result.returncode == 0, "systemctl " + action + " failed (output suppressed)")
        return result.stdout.decode("utf-8")

    def show(self, unit):
        values = dict(line.split("=", 1) for line in self.call("show", unit).splitlines() if "=" in line)
        require(set(PROPERTIES) <= values.keys(), "Manager is missing required inspection properties")
        require(values["Id"] == unit and values["LoadState"] == "loaded",
                "Original exact unit absent, aliased, masked or not loaded: " + unit)
        return values

    def state(self, unit):
        values = self.show(unit)
        require((values["ActiveState"], values["SubState"]) in
                (("active", "running"), ("active", "exited"), ("inactive", "dead")),
                "Unit is failed or transitioning; manual stabilization required: " + unit)
        return values["ActiveState"]

    def dependency_status(self, unit):
        values = dict(line.split("=", 1) for line in self.call("show-dependencies", unit).splitlines() if "=" in line)
        require(set(DEPENDENCY_PROPERTIES) <= values.keys(), "Manager is missing dependency inspection properties")
        require(values["Id"] == unit and values["LoadState"] == "loaded",
                "Dependency is absent, aliased or not loaded; manual integration required: " + unit)
        return values

    def stop(self, unit):
        self.call("stop", unit)
        values = self.show(unit)
        require(values["ActiveState"] == "inactive" and values["MainPID"] == "0",
                "Unit did not stop cleanly: " + unit)
        group = values["ControlGroup"]
        if group:
            members = Path("/sys/fs/cgroup") / group.lstrip("/") / "cgroup.procs"
            require(not members.exists() or not members.read_text().strip(),
                    "Unit cgroup still has processes; manual integration required: " + unit)


def check_dependencies(manager, values):
    unit = values["Id"]
    require(values["Job"] in ("", "0"), "Pending unit job; manual stabilization required: " + unit)
    for prop in ("OnFailure", "OnSuccess", "Requisite", "BindsTo", "Upholds", "UpheldBy",
                 "RequiredBy", "RequisiteOf", "BoundBy", "ConsistsOf", "PropagatesStopTo"):
        require(not values[prop], "Unsupported dependency effect " + prop + "; manual integration required: " + unit)
    for prop in ("FailureAction", "SuccessAction", "StartLimitAction"):
        require(values[prop] in ("", "none"), "System action policy refused: " + unit)
    defaults = {"basic.target", "sysinit.target"}
    wants, requires = set(values["Wants"].split()), set(values["Requires"].split())
    slice_unit = values["Slice"]
    require(not slice_unit or slice_unit.endswith(".slice"), "Unidentified service slice: " + unit)
    allowed_requires = defaults | {name for name in requires if name.endswith(".mount")}
    if slice_unit:
        allowed_requires.add(slice_unit)
    require(wants <= defaults and requires <= allowed_requires,
            "Application-service/nondefault Wants/Requires are unsupported; manual integration required: " + unit)
    conflicts = set(values["Conflicts"].split()) | set(values["ConflictedBy"].split())
    require(conflicts <= {"shutdown.target"}, "Nondefault Conflicts/ConflictedBy refused; manual integration required: " + unit)
    require(not ((wants | requires) & defaults or conflicts) or values["DefaultDependencies"] == "yes",
            "Only normal implicit target dependencies are supported: " + unit)
    pending = [(name, "active") for name in wants | requires | ({slice_unit} if slice_unit else set())]
    pending += [(name, "inactive") for name in conflicts]
    checked = {}
    # Starting an already-active target can still pull in its inactive Wants.
    # Verify the whole job graph, not just the direct target's active bit.
    while pending:
        name, expected = pending.pop()
        if name in checked:
            require(checked[name] == expected, "Conflicting transitive dependency effects; manual integration required")
            continue
        require(len(checked) < 256, "Dependency graph exceeds the supported inspection boundary")
        checked[name] = expected
        dependency = manager.dependency_status(name)
        require(dependency["ActiveState"] == expected and dependency["Job"] in ("", "0") and
                dependency["NeedDaemonReload"] == "no",
                "Dependency would change state or has a pending job/edit; manual integration required: " + name)
        if expected == "active":
            require(dependency["StopWhenUnneeded"] == "no" and not any(
                dependency[prop] for prop in ("Upholds", "OnFailure", "OnSuccess")),
                "Transitive automatic dependency effects require manual integration: " + name)
            for prop in ("FailureAction", "SuccessAction", "StartLimitAction"):
                require(dependency[prop] in ("", "none"), "Transitive system action refused: " + name)
            conflicts = set(dependency["Conflicts"].split()) | set(dependency["ConflictedBy"].split())
            allowed_conflicts = {"shutdown.target"}
            if name.endswith((".mount", ".automount", ".swap")):
                allowed_conflicts.add("umount.target")
            require(conflicts <= allowed_conflicts, "Transitive nondefault conflict refused: " + name)
            pending += [(other, "inactive") for other in conflicts]
            for prop in ("Wants", "Requires", "Requisite", "BindsTo"):
                pending += [(other, "active") for other in dependency[prop].split()]
        else:
            # Even an inactive conflict target can propagate a stop job.
            for prop in ("RequiredBy", "RequisiteOf", "BoundBy", "ConsistsOf", "PropagatesStopTo", "UpheldBy"):
                pending += [(other, "inactive") for other in dependency[prop].split()]


def validate_inventory(inventory, tools_required=True):
    keys(inventory, ("version", "inventory_complete", "canonical", "context", "supervisor",
                     "controller", "approved_binaries", "duplicates", "listen_ports"))
    require(inventory["version"] == 1 and inventory["inventory_complete"] is True,
            "Version 1 and an explicitly reviewed complete inventory are required")
    context = inventory["context"]
    keys(context, ("uid", "home", "config_home", "data_home", "runtime_dir"))
    require(type(context["uid"]) is int and context["uid"] > 0, "Explicit nonroot service UID required")
    account = pwd.getpwuid(context["uid"])
    require(context["home"] == account.pw_dir, "Inventory HOME does not match the identified account")
    for field in ("home", "config_home", "data_home", "runtime_dir"):
        absolute(context[field])
    canonical = inventory["canonical"]
    require(isinstance(canonical, dict), "canonical must be an explicit unit object")
    require(canonical.get("manager") in ("user", "system"), "Canonical manager must be explicit")
    if canonical["manager"] == "user":
        require("runtime_directory" not in canonical, "User units use %t/maximizer-host, not a system RuntimeDirectory override")
        require(os.geteuid() == context["uid"], "Run as the exact user, not root/sudo or a guessed user bus")
        require(context["runtime_dir"] == os.environ.get("XDG_RUNTIME_DIR") and
                context["config_home"] == os.environ.get("XDG_CONFIG_HOME", account.pw_dir + "/.config") and
                context["data_home"] == os.environ.get("XDG_DATA_HOME", account.pw_dir + "/.local/share"),
                "Inventory must match the invoking user's XDG manager/session context")
    else:
        require(os.geteuid() == 0, "System-manager installation requires root")
        require(canonical.get("runtime_directory") == "maximizer-host",
                "System service requires explicit approval: runtime_directory=maximizer-host")
    for field in ("supervisor", "controller"):
        path = absolute(inventory[field])
        require(path.name == ("maximizer-host-supervisor" if field == "supervisor" else "maximizer-host-supervision"),
                "Inventory must identify the installed supervisor/controller command name")
        require(not tools_required or (path.is_file() and os.access(path, os.X_OK)),
                "Required installed executable is absent: " + str(path))
    require(isinstance(inventory["approved_binaries"], list) and inventory["approved_binaries"],
            "Explicit approved Sunshine binary paths required")
    for binary in inventory["approved_binaries"]:
        path = absolute(binary)
        require(path.is_file() and os.access(path, os.X_OK), "Approved Sunshine executable is absent: " + str(path))
        require(path.resolve().name not in {"sh", "bash", "dash", "zsh", "fish", "ksh", "env", "nohup", "setsid",
                "systemctl", "sudo", "su", "runuser", "flatpak", "gio", "python", "python3", "perl", "node", "true"},
                "Interpreter/wrapper is not a native Sunshine executable; manual integration required")
        with path.open("rb") as executable:
            require(executable.read(4) == b"\x7fELF", "Script/non-ELF Sunshine launcher requires manual integration")
    require(isinstance(inventory["duplicates"], list), "duplicates must be an explicit array (possibly empty)")
    require(isinstance(inventory["listen_ports"], list) and inventory["listen_ports"],
            "All configured Sunshine listen ports must be explicit")
    ports = set()
    for port in inventory["listen_ports"]:
        keys(port, ("protocol", "port"))
        require(port["protocol"] in ("tcp", "udp") and type(port["port"]) is int and
                0 < port["port"] < 65536, "Invalid listen-port inventory")
        pair = (port["protocol"], port["port"])
        require(pair not in ports, "Duplicate listen-port inventory")
        ports.add(pair)
    return account


def unit_commands(files):
    commands = {key: [] for key in ("ExecStart",) + HOOKS}
    for value in files:
        section = ""
        for line in contents(value).decode("utf-8").splitlines():
            line = line.strip()
            if not line or line.startswith(("#", ";")):
                continue
            require(not line.endswith("\\"), "Continued unit directives require manual integration")
            if line.startswith("[") and line.endswith("]"):
                section = line[1:-1]
            elif section == "Service" and "=" in line:
                key, value = (part.strip() for part in line.split("=", 1))
                if key in commands:
                    if not value:
                        commands[key] = []
                    else:
                        commands[key].append(literal_words(value))
    return commands


def inspect_unit(spec, inventory, manager, target, installed, guards, pending=False):
    keys(spec, ("manager", "unit", "fragment_path", "unit_files", "command", "working_directory"),
         ("kind", "remove_package_sleep", "environment", "runtime_directory"))
    if spec is not inventory["canonical"]:
        require(spec.get("kind") == "unit", "Duplicate service requires kind=unit")
        require(not (set(spec) & {"environment", "runtime_directory"}),
                "Duplicate units cannot carry canonical environment/runtime overrides")
    else:
        require("kind" not in spec, "canonical must be a unit object, not a duplicate path")
    unit = unit_name(spec["unit"])
    require(spec["manager"] == manager.manager, "Mixed managers require manual integration, not implicit bus switching")
    command = spec["command"]
    require(isinstance(command, list) and command and all(isinstance(arg, str) and "\0" not in arg for arg in command),
            "command must be the original argument ARRAY, never a shell string")
    require(command[0] in inventory["approved_binaries"], "Unit does not directly launch an approved Sunshine binary")
    if spec is inventory["canonical"]:
        marker = b"MAXIMIZER_SUPERVISOR_FD\0"
        overlap = b""
        # Never execute a compatibility probe: an old host could start listening.
        with absolute(command[0]).open("rb") as executable:
            while True:
                chunk = executable.read(64 * 1024)
                require(chunk, "Canonical executable lacks the MAXIMIZER_SUPERVISOR_FD native bootstrap-gate marker; patched Sunshine is required")
                window = overlap + chunk
                if marker in window:
                    break
                overlap = window[-(len(marker) - 1):]
    cwd = absolute(spec["working_directory"])
    require(cwd.is_dir(), "Original working_directory is absent")
    environment = spec.get("environment", {})
    require(isinstance(environment, dict) and all(re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", key) and
            isinstance(value, str) and "\0" not in value for key, value in environment.items()),
            "environment must contain explicit string overrides only")
    require(type(spec.get("remove_package_sleep", False)) is bool, "remove_package_sleep must be boolean")
    values = manager.show(unit)
    check_dependencies(manager, values)
    require(pending or values["NeedDaemonReload"] == "no", "Pending unit edits require manual review/reload before installation")
    require(values["FragmentPath"] == str(absolute(spec["fragment_path"])), "Unit fragment differs from inventory: " + unit)
    paths = literal_words(values["DropInPaths"])
    if installed:
        require(str(target) in paths, "Managed drop-in is not loaded: " + unit)
        paths.remove(str(target))
    else:
        require(str(target) not in paths, "Unidentified existing supervision drop-in: " + unit)
    paths.insert(0, values["FragmentPath"])
    require(isinstance(spec["unit_files"], dict) and set(paths) == set(spec["unit_files"]),
            "Unidentified/missing unit fragment or drop-in: " + unit)
    files = []
    for path in paths:
        value = snapshot(absolute(path))
        require(value is not None and value["sha256"] == spec["unit_files"][path],
                "Unit source hash mismatch or absent original: " + path)
        guards[path] = value
        files.append(value)
    commands = unit_commands(files)
    require(commands["ExecStart"] == [command], "Original ExecStart does not exactly match command ARRAY: " + unit)
    for hook in HOOKS:
        allowed = hook == "ExecStartPre" and commands[hook] == [["/bin/sleep", "5"]]
        require(not commands[hook] or allowed, "Unreviewed launch/stop hook; manual integration required: " + unit + " " + hook)
        if hook != "ExecStartPre":
            require(not values[hook], "Loaded manager has unsupported hook: " + unit + " " + hook)
    require(not values["ExecStartPre"] or (values["ExecStartPre"].startswith(
        "{ path=/bin/sleep ; argv[]=/bin/sleep 5 ;") and values["ExecStartPre"].count("{ path=") == 1),
        "Loaded manager has an unreviewed pre-hook; manual integration required: " + unit)
    if spec.get("remove_package_sleep"):
        require(commands["ExecStartPre"] == [["/bin/sleep", "5"]], "Sleep removal approval does not match exact package hook")
    require(values["Type"] in (("simple", "exec", "oneshot") if installed else ("simple", "exec")),
            "Forking/notify/oneshot original units require manual integration: " + unit)
    require(values["DynamicUser"] == "no" and values["RemainAfterExit"] == "no" and
            values["KillMode"] == "control-group" and values["KillSignal"] in ("15", "SIGTERM") and
            values["SendSIGKILL"] == "yes" and values["WatchdogUSec"] == "0",
            "Unsupported process identity/lifecycle policy: " + unit)
    for prop in ("PIDFile", "RootDirectory", "RootImage", "RestartPreventExitStatus", "RestartForceExitStatus",
                 "OnFailure", "OnSuccess", "PropagatesStopTo", "BoundBy", "ConsistsOf", "RequiredBy", "TriggeredBy"):
        require(not values[prop], "Unsafe or unsupported unit policy " + prop + "; manual integration required: " + unit)
    for prop in ("FailureAction", "SuccessAction", "StartLimitAction"):
        require(values[prop] in ("", "none"), "System action policy refused: " + unit)
    uid = inventory["context"]["uid"]
    if manager.manager == "system" or values["User"]:
        user = values["User"]
        require(bool(user), "Existing system service must have an explicit nonroot User=")
        account = pwd.getpwuid(int(user)) if user.isdecimal() else pwd.getpwnam(user)
        require(account.pw_uid == uid, "Existing service UID differs from explicit inventory: " + unit)
    inherited_cwd = values["WorkingDirectory"] or (inventory["context"]["home"] if manager.manager == "user" else "/")
    require(inherited_cwd == str(cwd), "WorkingDirectory differs; refusing to change service CWD: " + unit)
    allowed_runtime = "maximizer-host" if installed and spec is inventory["canonical"] and manager.manager == "system" else ""
    require(values["RuntimeDirectory"] == allowed_runtime, "Existing RuntimeDirectory policy requires manual integration")
    return values


def check_listeners(inventory, values):
    ports = {(entry["protocol"], entry["port"]) for entry in inventory["listen_ports"]}
    sockets = set()
    for protocol in ("tcp", "udp"):
        for suffix in ("", "6"):
            table = Path("/proc/net/" + protocol + suffix)
            for line in table.read_text().splitlines()[1:]:
                fields = line.split()
                if (protocol, int(fields[1].rsplit(":", 1)[1], 16)) in ports and (protocol == "udp" or fields[3] == "0A"):
                    require(fields[9] != "0", "Unattributable listening socket; manual integration required")
                    sockets.add(fields[9])
    approved = {str(Path(path).resolve()) for path in inventory["approved_binaries"]}
    groups = [value["ControlGroup"] for value in values if value["ControlGroup"]]
    attributed = set()
    for process in Path("/proc").iterdir():
        if not process.name.isdecimal():
            continue
        owned = set()
        try:
            for fd in (process / "fd").iterdir():
                try:
                    link = os.readlink(fd)
                except FileNotFoundError:
                    continue
                if link.startswith("socket:[") and link[8:-1] in sockets:
                    owned.add(link[8:-1])
            executable = os.readlink(process / "exe")
            if not owned and executable not in approved:
                continue
            membership = [line.split(":", 2)[2] for line in (process / "cgroup").read_text().splitlines()
                          if line.startswith("0::") or ":name=systemd:" in line]
            require(process.stat().st_uid == inventory["context"]["uid"] and any(
                member == group or member.startswith(group + "/") for member in membership for group in groups),
                "Unmanaged Sunshine process or socket owner; manual integration required (no processes killed)")
            attributed.update(owned)
        except (FileNotFoundError, ProcessLookupError):
            continue
        except PermissionError:
            continue
    require(attributed == sockets, "Socket owner cannot be verified via /proc; manual integration required")


def desktop_image(spec, inventory, uid, gid, runtime, guards, installed):
    keys(spec, ("kind", "role", "source", "sha256", "name", "command"))
    require(spec["role"] in ("autostart", "application"), "Unsupported desktop role; manual integration required")
    source = absolute(spec["source"])
    require(re.fullmatch(r"[A-Za-z0-9_.-]*(?:sunshine|maximizer-host)[A-Za-z0-9_.-]*\.desktop", source.name, re.I),
            "Desktop basename is not an identified Sunshine/Maximizer application entry")
    context = inventory["context"]
    destination = Path(context["config_home"] if spec["role"] == "autostart" else context["data_home"])
    destination /= "autostart" if spec["role"] == "autostart" else "applications"
    target = destination / source.name
    allowed = {destination, Path("/etc/xdg/autostart")} if spec["role"] == "autostart" else {
        destination, Path("/usr/share/applications"), Path("/usr/local/share/applications")}
    require(source.parent in allowed, "Mixed shell/session or unidentified desktop location; manual integration required")
    value = installed["originals"].get(str(source)) if installed and source == target else snapshot(source)
    require(value is not None and value["sha256"] == spec["sha256"], "Desktop source is absent or hash differs: " + str(source))
    if source != target:
        guards[str(source)] = value
        require(installed or snapshot(target) is None, "Existing user override must itself be inventoried, never overwritten")
    text = contents(value).decode("utf-8")
    parser = configparser.ConfigParser(interpolation=None, delimiters=("=",), comment_prefixes=("#",))
    parser.optionxform = str
    try:
        parser.read_string(text)
    except configparser.Error:
        raise Refusal("Ambiguous desktop entry; manual integration required") from None
    require(parser.sections() == ["Desktop Entry"] and not parser.defaults(),
            "Desktop actions/mixed entries require manual integration")
    entry = parser["Desktop Entry"]
    require(entry.get("Type") == "Application" and entry.get("Name") == spec["name"] and
            entry.get("Hidden", "false") == "false" and entry.get("DBusActivatable", "false") == "false" and
            not entry.get("Actions"), "Desktop identity or launch semantics are unsupported")
    command = literal_words(entry.get("Exec", ""))
    require(command == spec["command"] and command and
            command[0] in ["sunshine"] + inventory["approved_binaries"],
            "Desktop Exec is not the exact approved direct Sunshine launch; manual integration required")
    canonical = inventory["canonical"]
    require(command[1:] == canonical["command"][1:], "Desktop arguments differ from canonical service; manual integration required")
    require(not entry.get("Path") or entry["Path"] == canonical["working_directory"],
            "Desktop CWD differs from canonical service; manual integration required")
    if spec["role"] == "autostart":
        field, replacement = "Hidden", "true"
    else:
        field = "Exec"
        replacement = " ".join(quote_desktop(word) for word in (
            inventory["controller"], "control-start", "--manager", canonical["manager"],
            "--unit", canonical["unit"], "--runtime-dir", runtime,
            "--supervisor", inventory["supervisor"], "--apply"))
    lines = text.splitlines(keepends=True)
    matches = [i for i, line in enumerate(lines) if re.match(r"^\s*" + field + r"\s*=", line)]
    require(len(matches) <= 1, "Ambiguous desktop launch field")
    if matches:
        lines[matches[0]] = field + "=" + replacement + "\n"
    else:
        lines.append(("" if text.endswith("\n") else "\n") + field + "=" + replacement + "\n")
    return target, image("".join(lines).encode(), 0o644, uid, gid)


def plan_install(inventory, manager, state=None, tools_required=True):
    account = validate_inventory(inventory, tools_required=tools_required)
    canonical = inventory["canonical"]
    context = inventory["context"]
    uid, gid = context["uid"], account.pw_gid
    system = canonical["manager"] == "system"
    root = Path("/etc/systemd/system") if system else Path(context["config_home"]) / "systemd/user"
    config_dir = Path("/etc/maximizer-host-supervision") if system else Path(context["config_home"]) / "maximizer-host-supervision"
    runtime = "/run/maximizer-host" if system else context["runtime_dir"] + "/maximizer-host"
    runtime_argument = runtime if system else "%t/maximizer-host"
    launch = config_dir / "launch.json"
    unit_specs = [canonical]
    desktop_specs = []
    for spec in inventory["duplicates"]:
        require(isinstance(spec, dict), "Invalid duplicate inventory")
        if spec.get("kind") == "unit":
            unit_specs.append(spec)
        elif spec.get("kind") == "desktop":
            desktop_specs.append(spec)
        else:
            raise Refusal("Unknown mixed shell/session startup path; manual integration required (no surrounding scripts changed)")
    require(len({spec["unit"] for spec in unit_specs}) == len(unit_specs), "Canonical/duplicate unit identities overlap")
    guards, files, directories, values, preimages = {}, {}, {}, [], {}

    def add(path, value, directory_uid, directory_gid, mode=0o755):
        require(str(path) not in files, "Duplicate managed path in inventory")
        preimages[str(path)] = snapshot(path)
        parent = path.parent
        missing = []
        while not parent.exists():
            missing.append(parent)
            parent = parent.parent
        safe_ancestors(parent / "placeholder")
        for parent in reversed(missing):
            directories[str(parent)] = {"uid": directory_uid, "gid": directory_gid, "mode": mode}
        files[str(path)] = value

    for spec in unit_specs:
        target = root / (unit_name(spec["unit"]) + ".d") / DROPIN
        require(not any(character.isspace() or character in "\\$%\"'" for character in str(target)),
                "Escaped/whitespace unit drop-in paths require manual integration")
        # Do not create missing parent directories during audit.
        old = snapshot(target) if target.parent.exists() else None
        require(state or old is None, "Unidentified existing managed drop-in: " + str(target))
        values.append(inspect_unit(spec, inventory, manager, target, state, guards))
        lines = ["# Managed by maximizer-host-supervision; uninstall through its journal.",
                 "[Unit]", "StartLimitIntervalSec=0", "[Service]", "ExecStart="]
        if spec is canonical:
            lines += ["Type=simple", "Restart=always", "RestartSec=1s", "KillMode=control-group", "TimeoutStopSec=8s",
                      "ExecStart=" + quote_systemd(inventory["supervisor"]) + " run --runtime-dir " +
                      quote_systemd(runtime_argument, specifier=not system) + " --launch-config " + quote_systemd(str(launch))]
            if system:
                lines += ["RuntimeDirectory=maximizer-host", "RuntimeDirectoryMode=0700"]
        else:
            lines += ["Type=oneshot", "Restart=no", "ExecStart=/usr/bin/true"]
        if spec.get("remove_package_sleep"):
            lines += ["ExecStartPre="]
        add(target, image(("\n".join(lines) + "\n").encode(), 0o644,
                          0 if system else uid, 0 if system else gid), 0 if system else uid, 0 if system else gid)
    launch_value = {"command": canonical["command"], "cwd": canonical["working_directory"]}
    if canonical.get("environment"):
        launch_value["environment"] = canonical["environment"]
    require(state or not config_dir.exists(), "Unidentified launch-config directory; manual integration required")
    require(state or not Path(runtime).exists(), "Existing supervisor runtime directory requires manual identification")
    add(launch, image(encoded(launch_value), 0o600, uid, gid), uid, gid, 0o700)
    for spec in desktop_specs:
        target, value = desktop_image(spec, inventory, uid, gid, runtime, guards, state)
        add(target, value, uid, gid)
        if not state and str(target) == spec["source"]:
            require(preimages[str(target)] is not None and preimages[str(target)]["sha256"] == spec["sha256"],
                    "Desktop changed during preflight; refusing installation")
    require(not set(files) & set(guards), "Managed destinations overlap protected source files")
    check_restorable(preimages)
    states = {spec["unit"]: manager.state(spec["unit"]) for spec in unit_specs}
    check_listeners(inventory, values)
    return {"files": files, "before": preimages, "guards": guards, "directories": directories, "states": states,
            "units": [spec["unit"] for spec in unit_specs], "runtime": runtime}


def check_files(expected):
    for path, value in expected.items():
        current = snapshot(Path(path)) if Path(path).parent.exists() else None
        require(current == value, "File drift; manual integration required: " + path)


def save_journal(directory, state):
    path = directory / "manifest.json"
    replace_file(path, snapshot(path), image(encoded(state), 0o600, os.geteuid(), os.getegid()))


def load_journal(directory):
    if not directory.exists():
        return None
    safe_ancestors(directory / "placeholder")
    info = directory.lstat()
    require(stat.S_IMODE(info.st_mode) == 0o700 and info.st_uid == os.geteuid(),
            "Backup directory must be private 0700 and owned by invoking UID")
    require((directory / "manifest.json").exists(), "Unidentified/incomplete backup directory; never overwriting original backup")
    state = read_json(directory / "manifest.json", private=True)
    require(state.get("version") == 1, "Unknown recovery journal version")
    for value in state["originals"].values():
        if value is not None:
            contents(value)
    return state


def make_directories(directories, created, pending, checkpoint):
    for name, policy in sorted(directories.items(), key=lambda item: len(Path(item[0]).parts)):
        path = Path(name)
        safe_ancestors(path)
        require(not path.exists(), "Concurrent directory creation; refusing ownership change: " + name)
        pending[name] = dict(policy)
        checkpoint()
        path.mkdir(mode=policy["mode"])
        os.chown(path, policy["uid"], policy["gid"])
        os.chmod(path, policy["mode"])
        fsync_directory(path.parent)
        info = path.stat()
        created[name] = {**policy, "device": info.st_dev, "inode": info.st_ino}
        del pending[name]
        checkpoint()


def check_pending_directories(pending):
    for name in pending:
        safe_ancestors(Path(name))
        require(not os.path.lexists(name),
                "Unaccounted directory creation requires operator review; keep the journal and resolve the path before recovery: " + name)


def check_directories(directories):
    for name, policy in directories.items():
        path = Path(name)
        safe_ancestors(path / "placeholder")
        info = path.lstat()
        require(stat.S_ISDIR(info.st_mode) and
                (stat.S_IMODE(info.st_mode), info.st_uid, info.st_gid, info.st_dev, info.st_ino) ==
                (policy["mode"], policy["uid"], policy["gid"], policy["device"], policy["inode"]),
                "Managed directory drift; manual integration required: " + name)


def cleanup_directories(directories):
    for name, policy in sorted(directories.items(), key=lambda item: -len(Path(item[0]).parts)):
        path = Path(name)
        if not path.exists():
            continue
        safe_ancestors(path / "placeholder")
        info = path.stat()
        if (stat.S_IMODE(info.st_mode), info.st_uid, info.st_gid, info.st_dev, info.st_ino) == (
                policy["mode"], policy["uid"], policy["gid"], policy["device"], policy["inode"]):
            try:
                path.rmdir()
                fsync_directory(path.parent)
            except OSError:
                pass  # Never remove a nonempty directory, including user data added later.


def restore_states(manager, states):
    for unit, desired in states.items():
        if desired == "active":
            manager.call("start", unit)
            require(manager.state(unit) == "active", "Service did not regain its recorded active state: " + unit)


def check_stop_scope(state, manager, pending=False):
    inventory = state["inventory"]
    specs = [inventory["canonical"]] + [spec for spec in inventory["duplicates"] if spec["kind"] == "unit"]
    for spec in specs:
        targets = [Path(path) for path in state["originals"] if Path(path).name == DROPIN and
                   Path(path).parent.name == spec["unit"] + ".d"]
        require(len(targets) == 1, "Recovery journal has ambiguous unit destinations")
        loaded = str(targets[0]) in literal_words(manager.show(spec["unit"])["DropInPaths"])
        inspect_unit(spec, inventory, manager, targets[0], loaded, {}, pending=pending)


def rollback(directory, state, manager):
    tx = state["transaction"]
    require(isinstance(tx.get("pending_directories"), dict),
            "Interrupted journal lacks directory creation intents; keep the journal for manual recovery")
    check_restorable(tx["before"])
    check_restorable(tx["after"])
    check_pending_directories(tx["pending_directories"])
    check_files(state["guards"])
    current_images = {}
    for path in tx["before"]:
        current = snapshot(Path(path))
        require(current in (tx["before"][path], tx["after"][path]), "Rollback refused due to file drift: " + path)
        current_images[path] = current
    check_stop_scope(state, manager, pending=True)
    state["stage"] = "rolling-back"
    save_journal(directory, state)
    for unit in state["units"]:  # Canonical owner is always stopped first.
        manager.stop(unit)
    for path, desired in tx["before"].items():
        path = Path(path)
        if desired is not None or path.parent.exists():
            replace_file(path, current_images[str(path)], desired)
    manager.call("daemon-reload")
    check_stop_scope(state, manager)
    restore_states(manager, tx["states_before"])
    cleanup_directories(tx["new_directories"])
    check_pending_directories(tx["pending_directories"])
    for name in tx["new_directories"]:
        require(not os.path.lexists(name), "Rollback directory cleanup incomplete; keep the journal for operator review: " + name)
    state["installed"] = tx["installed_before"]
    state["directories"] = tx["directories_before"]
    state["stage"] = tx["previous_stage"]
    state["transaction"] = None
    save_journal(directory, state)


def verify_loaded(state, manager):
    for index, unit in enumerate(state["units"]):
        values = manager.show(unit)
        expected = {"Type": "simple", "Restart": "always", "RestartUSec": "1s",
                    "KillMode": "control-group", "TimeoutStopUSec": "8s", "StartLimitIntervalUSec": "0"} if index == 0 else {
                        "Type": "oneshot", "Restart": "no"}
        require(all(values[key] == value for key, value in expected.items()),
                "Generated foreground/no-op lifecycle policy did not load: " + unit)
        executable = state["inventory"]["supervisor"] if index == 0 else "/usr/bin/true"
        command = [executable]
        if index == 0:
            inventory = state["inventory"]
            runtime = "/run/maximizer-host" if manager.manager == "system" else inventory["context"]["runtime_dir"] + "/maximizer-host"
            launch = [path for path in state["originals"] if Path(path).name == "launch.json"]
            require(len(launch) == 1, "Ambiguous launch configuration in journal")
            command += ["run", "--runtime-dir", runtime, "--launch-config", launch[0]]
        require(values["ExecStart"].startswith("{ path=" + executable + " ; argv[]=" + " ".join(command) + " ; ignore_errors=no ;") and
                values["ExecStart"].count("{ path=") == 1,
                "Generated ExecStart was not loaded exactly: " + unit)


def transact(directory, state, manager, after, states_after, new_directories, intent):
    before = state["installed"] if state["stage"] == "installed" else state["originals"]
    check_restorable(before)
    check_restorable(after)
    check_files(before)
    tx = {"before": before, "after": after, "states_before": {unit: manager.state(unit) for unit in state["units"]},
          "previous_stage": state["stage"], "new_directories": {}, "pending_directories": {}, "intent": intent,
          "installed_before": state.get("installed"), "directories_before": dict(state["directories"])}
    state["transaction"] = tx
    state["stage"] = "prepared"
    save_journal(directory, state)
    try:
        check_files(state["guards"])
        check_stop_scope(state, manager)
        state["stage"] = "stopping"
        save_journal(directory, state)
        for unit in state["units"]:
            manager.stop(unit)
        make_directories(new_directories, tx["new_directories"], tx["pending_directories"], lambda: save_journal(directory, state))
        state["stage"] = "writing"
        save_journal(directory, state)
        for path, desired in after.items():
            replace_file(Path(path), before[path], desired)
        state["stage"] = "reloading"
        save_journal(directory, state)
        manager.call("daemon-reload")
        if intent == "install":
            verify_loaded(state, manager)
        check_stop_scope(state, manager)
        # Recheck after stopping: desktop-launched processes can race preflight.
        check_listeners(state["inventory"], [manager.show(unit) for unit in state["units"]])
        state["stage"] = "starting"
        save_journal(directory, state)
        restore_states(manager, states_after)
        check_files(after)
        check_files(state["guards"])
        if intent == "install":
            state["installed"] = after
            state["directories"].update(tx["new_directories"])
            state["stage"] = "installed"
        else:
            state["stage"] = "uninstalled"
        state["transaction"] = None
        save_journal(directory, state)
    except BaseException:
        state["transaction"] = tx
        state["stage"] = "rollback-required"
        save_journal(directory, state)
        try:
            rollback(directory, state, manager)
        except BaseException:
            state["stage"] = "rollback-required"
            save_journal(directory, state)
            raise Refusal("Transaction interrupted; rollback incomplete. Keep backup and run recover --apply after resolving directory/dependency/ownership drift or manager jobs") from None
        raise Refusal("Transaction failed; pre-transaction files and active states restored (details suppressed)") from None
    if intent == "uninstall":
        cleanup_directories(state["directories"])


def operate(args):
    directory = absolute(args.state_dir)
    inventory = read_json(absolute(args.inventory), private=True) if args.inventory else None
    state = load_journal(directory)
    if inventory is None:
        require(state is not None, "An explicit --inventory is required before first installation")
        inventory = state["inventory"]
    tools_required = args.action not in ("uninstall", "recover") and not (state and state.get("transaction"))
    validate_inventory(inventory, tools_required=tools_required)
    inventory_hash = digest(encoded(inventory))
    if state:
        require(inventory_hash == state["inventory_hash"], "Inventory changed; uninstall with original inventory and use a fresh backup directory")
        check_restorable(state["originals"])
    manager = Manager(inventory["canonical"]["manager"], args.apply)
    lock_path = Path("/etc/systemd/system") if manager.manager == "system" else Path(inventory["context"]["config_home"])
    safe_ancestors(lock_path / "placeholder")
    with contextlib.ExitStack() as stack:
        if args.apply:
            fd = os.open(lock_path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
            stack.callback(os.close, fd)
            try:
                fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                raise Refusal("Another installer owns this manager's deployment lock") from None
            require(load_journal(directory) == state, "Recovery journal changed while acquiring lock")
        if state and state.get("transaction"):
            require(args.action in ("audit", "recover"), "Incomplete transaction; use recover, never start a second transaction")
            if args.action == "recover" and args.apply:
                rollback(directory, state, manager)
                return {"status": "recovered", "stage": state["stage"]}
            pending = state["transaction"].get("pending_directories")
            return {"status": "recovery-required", "stage": state["stage"], "apply": False,
                    "pending_directories": sorted(pending) if isinstance(pending, dict) else None}
        require(args.action != "recover", "No interrupted transaction to recover")
        if state and state["stage"] in ("uninstalled", "not-installed"):
            check_files(state["originals"])
            if args.action in ("audit", "uninstall"):
                return {"status": "not-installed", "apply": False}
            require(state["stage"] == "not-installed", "Original backup retained; use a fresh state directory for a new deployment")
        installed = state if state and state["stage"] == "installed" else None
        if installed:
            check_files(state["installed"])
            check_files(state["guards"])
            check_directories(state["directories"])
        plan = plan_install(inventory, manager, installed, tools_required=tools_required)
        require(not any(directory == Path(path) or directory in Path(path).parents or Path(path) in directory.parents
                        for path in plan["files"]), "Backup directory overlaps managed assets")
        if args.action == "uninstall":
            require(installed, "No installed transaction to uninstall")
            if args.apply:
                transact(directory, state, manager, state["originals"], state["original_states"], {}, "uninstall")
                return {"status": "uninstalled", "backup_retained": str(directory)}
        elif installed and plan["files"] == state["installed"]:
            verify_loaded(state, manager)
            return {"status": "installed-unchanged", "units": plan["units"], "service_states": plan["states"], "apply": False}
        elif args.action == "install" and args.apply:
            if not state:
                check_files(plan["before"])
                safe_ancestors(directory)
                directory.mkdir(mode=0o700)
                directory.chmod(0o700)
                fsync_directory(directory.parent)
                state = {"version": 1, "stage": "not-installed", "inventory": inventory, "inventory_hash": inventory_hash,
                         "originals": plan["before"], "original_states": plan["states"], "guards": plan["guards"],
                         "units": plan["units"], "directories": {}, "transaction": None}
                save_journal(directory, state)
            require(set(plan["files"]) == set(state["originals"]), "Upgrade changes managed scope; uninstall first")
            states_after = {unit: "active" if i == 0 else "inactive" for i, unit in enumerate(plan["units"])}
            transact(directory, state, manager, plan["files"], states_after, plan["directories"], "install")
            return {"status": "installed", "canonical": plan["units"][0], "backup": str(directory)}
        return {"status": "dry-run", "action": args.action, "manager": manager.manager,
                "units": plan["units"], "service_states": plan["states"],
                "managed_paths": sorted(plan["files"]), "runtime_dir": plan["runtime"], "apply": False}


def control_start(args):
    unit_name(args.unit)
    absolute(args.runtime_dir)
    require(absolute(args.supervisor).name == "maximizer-host-supervisor", "Expected the installed supervisor command")
    if not args.apply:
        return {"status": "dry-run", "action": "control-start", "unit": args.unit}
    manager = Manager(args.manager, apply=True)
    values = manager.show(args.unit)
    require(any(Path(path).name == DROPIN for path in literal_words(values["DropInPaths"])),
            "Canonical supervisor drop-in is not loaded; refusing launcher fallback")
    require(values["Type"] == "simple" and values["ExecStart"].startswith(
        "{ path=" + args.supervisor + " ; argv[]=" + args.supervisor + " run --runtime-dir " + args.runtime_dir + " --launch-config "),
        "Launcher does not identify the canonical supervisor command/runtime")
    manager.call("start", args.unit)
    deadline = time.monotonic() + 10
    while time.monotonic() < deadline:
        try:
            result = subprocess.run([args.supervisor, "start", "--runtime-dir", args.runtime_dir],
                                    stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=2, check=False)
            if result.returncode == 0:
                return {"status": "started", "unit": args.unit}
        except subprocess.TimeoutExpired:
            pass
        time.sleep(0.2)
    raise Refusal("Canonical service started but supervisor control socket is unavailable")


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__, epilog=LIMITS,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("action", nargs="?", default="audit", choices=("audit", "install", "uninstall", "recover", "control-start"))
    parser.add_argument("--inventory", help="private 0600 JSON inventory (required initially)")
    parser.add_argument("--state-dir", help="explicit private backup directory; existing parent required")
    parser.add_argument("--apply", action="store_true", help="explicitly permit file and inventory-unit activation changes")
    parser.add_argument("--manager", choices=("user", "system"), help="control-start only")
    parser.add_argument("--unit", help="control-start only; exact canonical .service")
    parser.add_argument("--runtime-dir", help="control-start only; resolved runtime path, not %%t")
    parser.add_argument("--supervisor", help="control-start only; installed absolute executable")
    args = parser.parse_args(argv)
    if args.action == "control-start":
        require_fields = (args.manager, args.unit, args.runtime_dir, args.supervisor)
        if not all(require_fields) or args.inventory or args.state_dir:
            parser.error("control-start requires only --manager, --unit, --runtime-dir, --supervisor and optional --apply")
    elif not args.state_dir or any((args.manager, args.unit, args.runtime_dir, args.supervisor)):
        parser.error("deployment operations require --state-dir; control-start arguments are not accepted")
    if args.action == "audit" and args.apply:
        parser.error("audit is read-only; choose install/uninstall/recover for --apply")
    try:
        result = control_start(args) if args.action == "control-start" else operate(args)
        print(json.dumps(result, sort_keys=True))
        return 2 if result["status"] == "recovery-required" else 0
    except Refusal as error:
        print("Refused: " + str(error), file=sys.stderr)
    except (OSError, ValueError, KeyError, TypeError, UnicodeError) as error:
        print("Refused: inspection/IO failed (" + type(error).__name__ + "); no credential-bearing details printed", file=sys.stderr)
    return 1


if __name__ == "__main__":
    def interrupted(signum, frame):
        raise Refusal("Interrupted; consult the recovery journal")
    signal.signal(signal.SIGTERM, interrupted)
    sys.exit(main())
