#!/usr/bin/env python3
"""Session-owned Sunshine supervisor. No network or service configuration writes.

Requires Linux, Python 3.10+, iproute2, and the patched native Sunshine binary.
Only the Gaming L3 interface may have a directly assigned public IPv4. Discovery
is repeated at runtime; interface names, MACs and addresses are never selectors.
"""

import argparse
import datetime
import errno
import fcntl
import hashlib
import ipaddress
import json
import math
import os
from pathlib import Path
import selectors
import signal
import socket
import stat
import struct
import subprocess
import sys
import time


OFFSETS = (-5, 0, 1, 21)
PACKET_LIMIT = 16384
HEARTBEAT_TIMEOUT = 3.0


def utc():
    return datetime.datetime.now(datetime.timezone.utc).isoformat()


def log(event, **fields):
    print(json.dumps({"timestamp": utc(), "monotonic": time.monotonic(),
                      "event": event, **fields}, allow_nan=False), flush=True)


def public_ipv4(value):
    address = ipaddress.IPv4Address(value)
    # Python 3.10/3.11 predate the corrected 192.0.0/24 classification.
    if address in ipaddress.IPv4Network("192.0.0.0/24"):
        return str(address) in ("192.0.0.9", "192.0.0.10")
    if address in ipaddress.IPv4Network("192.88.99.0/24"):
        return False  # Deprecated 6to4 relay space, not a Gaming host address.
    return address.is_global and not (address.is_multicast or address.is_reserved)


def discover_network():
    """Read-only local snapshots; no DHCP, DNS, ping, route mutation or WAN query."""
    result = {"ip_present": False, "interface": None, "public_ipv4": [],
              "link_usable": False, "reason": "no_public_ipv4", "candidates": [],
              "observed_at": utc(), "error": None}
    try:
        snapshots = []
        for arguments in (("-j", "-4", "address", "show"), ("-j", "-d", "link", "show")):
            command = subprocess.run(["ip", *arguments], check=True, capture_output=True,
                                     text=True, timeout=1)
            if len(command.stdout) > 1024 * 1024:
                raise ValueError("oversized iproute2 snapshot")
            snapshot = json.loads(command.stdout)
            if not isinstance(snapshot, list):
                raise ValueError("iproute2 did not return an array")
            snapshots.append(snapshot)
        links = {item["ifindex"]: item for item in snapshots[1]}
        for item in snapshots[0]:
            link = links.get(item["ifindex"])
            if not link or link["ifname"] != item["ifname"]:
                continue  # Interface was replaced during the snapshot.
            flags = set(link.get("flags", []))
            if "LOOPBACK" in flags:
                continue
            addresses = set()
            for entry in item.get("addr_info", []):
                if (entry.get("family") != "inet" or entry.get("scope") != "global" or
                        any(entry.get(flag, False) for flag in ("tentative", "dadfailed", "deprecated")) or
                        set(entry.get("flags", [])) & {"tentative", "dadfailed", "deprecated"} or
                        entry.get("valid_life_time") == 0 or entry.get("preferred_life_time") == 0):
                    continue
                if public_ipv4(entry["local"]):
                    addresses.add(entry["local"])
            if addresses:
                usable = ("UP" in flags and link.get("operstate") not in
                          ("DOWN", "LOWERLAYERDOWN", "NOTPRESENT") and
                          "NO-CARRIER" not in flags)
                result["candidates"].append({"interface": link["ifname"],
                    "ifindex": link["ifindex"], "public_ipv4": sorted(addresses),
                    "link_usable": usable})
        candidates = result["candidates"]
        if len(candidates) == 1:
            result.update(candidates[0])
            result["ip_present"] = True
            result["reason"] = "ready" if result["link_usable"] else "link_unusable"
        elif candidates:
            result["reason"] = "ambiguous_public_interfaces"
    except (OSError, ValueError, KeyError, TypeError, subprocess.SubprocessError) as error:
        # Inspection failure is unknown, not proof of a dead listener.
        result["error"] = type(error).__name__
        result["reason"] = "network_observation_failed"
    return result


def process_identity(pid, include_zombie=False):
    try:
        proc = Path("/proc") / str(pid)
        fields = (proc / "stat").read_text().rsplit(")", 1)[1].split()
        if fields[0] == "Z" and not include_zombie:
            return None
        uid = next(line.split()[1] for line in (proc / "status").read_text().splitlines()
                   if line.startswith("Uid:"))
        return {"pid": pid, "start_ticks": int(fields[19]), "uid": int(uid),
                "boot_id": Path("/proc/sys/kernel/random/boot_id").read_text().strip()}
    except (FileNotFoundError, ProcessLookupError):
        return None


def kernel_listeners():
    rows = []
    for family, name in ((socket.AF_INET, "tcp"), (socket.AF_INET6, "tcp6")):
        path = Path("/proc/self/net") / name
        try:
            lines = path.read_text().splitlines()[1:]
        except FileNotFoundError:
            if name == "tcp6":
                continue
            raise
        for line in lines:
            fields = line.split()
            if fields[3] != "0A":
                continue
            address, port = fields[1].split(":")
            raw = b"".join(int(address[i:i + 8], 16).to_bytes(4, sys.byteorder)
                           for i in range(0, len(address), 8))
            rows.append({"port": int(port, 16), "address": socket.inet_ntop(family, raw),
                         "inode": fields[9]})
    return rows


def socket_inodes(pid):
    inodes = set()
    for descriptor in (Path("/proc") / str(pid) / "fd").iterdir():
        try:
            target = os.readlink(descriptor)
        except FileNotFoundError:
            continue
        if target.startswith("socket:["):
            inodes.add(target[8:-1])
    return inodes


def read_json(path):
    fd = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
    with os.fdopen(fd) as stream:
        info = os.fstat(stream.fileno())
        if (not stat.S_ISREG(info.st_mode) or info.st_uid not in (0, os.geteuid()) or
                info.st_mode & 0o022 or info.st_size > 1024 * 1024):
            raise ValueError("JSON configuration must be an owned, non-writable regular file <=1MiB")
        return json.load(stream)


def atomic_json(path, value):
    temporary = path.with_name(path.name + ".new")
    fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_CLOEXEC | os.O_NOFOLLOW, 0o600)
    with os.fdopen(fd, "w") as stream:
        json.dump(value, stream, allow_nan=False)
        stream.flush()
        os.fsync(stream.fileno())
    os.replace(temporary, path)


def runtime_directory(path, create=False):
    path = Path(path)
    if not path.is_absolute() or path != path.resolve(strict=False):
        raise ValueError("runtime directory must be absolute, without symlinks")
    if create:
        try:
            path.mkdir(mode=0o700)
        except FileExistsError:
            pass
    info = path.lstat()
    permitted_owner = info.st_uid == os.geteuid() or not create and os.geteuid() == 0
    if not stat.S_ISDIR(info.st_mode) or not permitted_owner or info.st_mode & 0o077:
        raise ValueError("runtime directory must be owned by this UID and mode 0700")
    return path


def unavailable(reason):
    return {"schema_version": 1, "state": "unknown", "ready": False, "reason": reason,
            "child": {"pid": None, "generation": 0},
            "network": {"ip_present": False, "interface": None, "public_ipv4": []},
            "ports": [], "recovery_count": 0, "last_recovery_action": None,
            "observed_at": utc(), "next_retry_seconds": None}


class Supervisor:
    def __init__(self, args, runtime, launch):
        self.args, self.runtime, self.launch = args, runtime, launch
        self.selector = selectors.DefaultSelector()
        self.control = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.child = self.channel = self.pidfd = None
        self.channel_closed_at = None
        self.identity = None
        self.generation = self.recovery_count = 0
        self.last_recovery = None
        self.ports = {}
        self.configured = self.gate_open = self.application_ready = False
        self.address_family = None
        self.generation_healthy = False
        self.desired = True
        self.exiting = False
        self.stop_deadline = None
        self.kill_sent = False
        self.failure_pending = None
        self.next_launch = self.next_network = self.next_check = 0.0
        self.launch_delay = self.network_delay = min(1.0, args.poll_cap)
        self.gate_time = self.spawn_time = 0.0
        self.absent_since = time.monotonic()
        self.next_warning = self.absent_since + args.address_budget
        self.network = {"ip_present": False, "interface": None, "public_ipv4": [],
                        "link_usable": False, "reason": "initializing", "error": None}
        self.fingerprint = None
        self.observation_error = None
        self.conflict = None
        self.checks = 0
        self.digest = hashlib.sha256(json.dumps(launch, sort_keys=True).encode()).hexdigest()

    def save(self):
        atomic_json(self.runtime / "history.json", {"generation": self.generation,
                    "recovery_count": self.recovery_count, "last_recovery_action": self.last_recovery})

    def setup(self):
        try:
            history = read_json(self.runtime / "history.json")
            self.generation = int(history["generation"])
            self.recovery_count = int(history["recovery_count"])
            self.last_recovery = history["last_recovery_action"]
        except FileNotFoundError:
            pass
        path = self.runtime / "control.sock"
        if path.exists():
            if not stat.S_ISSOCK(path.lstat().st_mode):
                raise ValueError("refusing to replace a non-socket control path")
            path.unlink()
        self.control.bind(str(path))
        os.chmod(path, 0o600)
        self.control.listen(16)
        self.control.setblocking(False)
        self.selector.register(self.control, selectors.EVENT_READ, "control")
        # Wake early on link/address/route changes. Lost notifications are harmless:
        # the independent periodic snapshots remain authoritative.
        try:
            monitor = socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, socket.NETLINK_ROUTE)
            monitor.bind((0, 1 | 0x10 | 0x40))
            monitor.setblocking(False)
            self.selector.register(monitor, selectors.EVENT_READ, "network")
        except OSError as error:
            if "monitor" in locals():
                monitor.close()
            log("network_monitor_polling_fallback", error=error.errno)
        atomic_json(self.runtime / "owner.json", {"pid": os.getpid(), "launch_digest": self.digest})
        # A previous owner may have died outside systemd. Never overlap its child.
        try:
            previous = read_json(self.runtime / "child.json")
        except FileNotFoundError:
            return
        saved = previous["identity"]
        if saved and process_identity(saved["pid"]) == saved:
            if previous["launch_digest"] != self.digest:
                raise ValueError("live previous child has a different launch configuration")
            self.pidfd = os.pidfd_open(saved["pid"])
            if process_identity(saved["pid"]) != saved:
                os.close(self.pidfd)
                self.pidfd = None
                return
            self.identity = saved
            self.record_recovery("previous_supervisor_exited")
            self.stop_child()

    def record_recovery(self, reason):
        self.recovery_count += 1
        self.last_recovery = {"action": "restart_child", "reason": reason,
                              "timestamp": utc(), "generation": self.generation}
        self.save()
        log("recovery_intervention", **self.last_recovery, recovery_count=self.recovery_count)

    def signal_child(self, sig):
        if self.pidfd is not None:
            try:
                signal.pidfd_send_signal(self.pidfd, sig)
            except ProcessLookupError:
                pass
        elif self.child is not None:
            # Popen retains an unreaped direct child, so its PID cannot be reused.
            # This covers failures between exec() and installing the pidfd.
            self.child.send_signal(sig)

    def stop_child(self):
        if self.identity and self.stop_deadline is None:
            self.signal_child(signal.SIGTERM)
            self.stop_deadline = time.monotonic() + self.args.stop_timeout
            self.kill_sent = False
            log("child_stop_requested", pid=self.identity["pid"], generation=self.generation)

    def recover(self, reason):
        if self.stop_deadline is not None or not self.desired or self.exiting:
            return
        self.record_recovery(reason)
        self.stop_child()
        self.next_launch = time.monotonic() + self.launch_delay
        self.launch_delay = min(self.launch_delay * 2, self.args.poll_cap)

    def spawn(self):
        parent, child = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET)
        try:
            environment = {**os.environ, **self.launch.get("environment", {})}
            # Never reuse an inherited/launch-config descriptor from an older process.
            environment["MAXIMIZER_SUPERVISOR_FD"] = str(child.fileno())
            self.child = subprocess.Popen(self.launch["command"], cwd=self.launch["cwd"],
                                          env=environment, pass_fds=(child.fileno(),))
        except OSError as error:
            parent.close()
            self.record_recovery("exec_failed_errno_" + str(error.errno))
            self.next_launch = time.monotonic() + self.launch_delay
            self.launch_delay = min(self.launch_delay * 2, self.args.poll_cap)
            return
        finally:
            child.close()
        self.channel = parent
        self.channel_closed_at = None
        try:
            self.identity = process_identity(self.child.pid, include_zombie=True)
            self.pidfd = os.pidfd_open(self.child.pid)
        except BaseException:
            # Cover a successful exec followed by failed pidfd/identity setup.
            self.child.kill()
            parent.close()
            self.channel = None
            try:
                self.child.wait(timeout=2)
            except subprocess.TimeoutExpired:
                log("child_reap_timeout", pid=self.child.pid)
            raise
        self.channel.setblocking(False)
        self.selector.register(parent, selectors.EVENT_READ, "child")
        self.generation += 1
        self.configured = self.gate_open = self.generation_healthy = self.application_ready = False
        self.conflict = self.failure_pending = None
        self.spawn_time = time.monotonic()
        self.next_check = self.spawn_time
        for port in self.ports.values():
            port["listening_reported"] = False
            port["heartbeat"] = None
            port["inode"] = None
        atomic_json(self.runtime / "child.json", {"identity": self.identity, "launch_digest": self.digest})
        self.save()
        log("child_bootstrap", pid=self.child.pid, generation=self.generation)

    def packet(self, packet):
        if not isinstance(packet, dict):
            raise ValueError("IPC packet is not an object")
        event = packet.get("event")
        if event == "configuration":
            if self.configured or packet.get("schema") != 1:
                raise ValueError("unexpected configuration/schema")
            base = packet["base_port"]
            ports = packet["ports"]
            if (type(base) is not int or not 6 <= base <= 65514 or
                    packet["address_family"] not in ("ipv4", "both") or len(ports) != 4 or
                    {p["port"] for p in ports} != {base + offset for offset in OFFSETS} or
                    any(p["protocol"] != "tcp" or not isinstance(p["name"], str) for p in ports)):
                raise ValueError("invalid effective TCP port family")
            previous = self.ports
            self.ports = {}
            for item in ports:
                old = previous.get(item["port"], {})
                self.ports[item["port"]] = {**item, "last_bind_at": old.get("last_bind_at"),
                    "bind_monotonic": old.get("bind_monotonic"), "address": "", "inode": None,
                    "ipv6_only": False, "listening_reported": False, "heartbeat": None}
            self.configured = True
            self.address_family = packet["address_family"]
            log("effective_configuration", base_port=base, ports=sorted(self.ports),
                address_family=packet["address_family"], generation=self.generation)
        elif event == "intent":
            if not self.configured:
                raise ValueError("intent before configuration")
            action = packet["action"]
            if action not in ("stop", "restart"):
                raise ValueError("unknown child intent")
            self.control_action(action)
        elif event == "application_ready":
            if type(packet["ready"]) is not bool:
                raise ValueError("invalid application readiness")
            self.application_ready = packet["ready"]
        elif event in ("bound", "listening", "heartbeat", "listener_failed"):
            if not self.configured or packet["port"] not in self.ports:
                raise ValueError("unconfigured listener")
            item = self.ports[packet["port"]]
            if event == "listener_failed":
                self.recover("listener_failed_" + str(item["port"]) + ": " + str(packet["error"])[:2048])
                return
            if not self.gate_open:
                raise ValueError("listener activity before address gate")
            instant = packet["monotonic"]
            if (type(instant) not in (int, float) or not math.isfinite(instant) or
                    not self.spawn_time - 1 <= instant <= time.monotonic() + 1):
                raise ValueError("invalid event monotonic timestamp")
            if event == "heartbeat":
                item["heartbeat"] = instant
                return
            address = str(ipaddress.ip_address(packet["address"]))
            if type(packet["ipv6_only"]) is not bool:
                raise ValueError("invalid address family evidence")
            item.update(address=address, ipv6_only=packet["ipv6_only"], inode=str(packet.get("inode", "")))
            if event == "bound":
                timestamp = datetime.datetime.fromisoformat(packet["timestamp"].replace("Z", "+00:00"))
                if timestamp.utcoffset() != datetime.timedelta(0):
                    raise ValueError("bind timestamp is not UTC")
                item.update(last_bind_at=packet["timestamp"], bind_monotonic=instant)
            else:
                if item["last_bind_at"] is None:
                    raise ValueError("listening without a bind event")
                item.update(listening_reported=True, heartbeat=instant)
            log("listener_" + event, port=item["port"], address=address,
                bind_timestamp=item["last_bind_at"], generation=self.generation)
            self.next_check = min(self.next_check, time.monotonic())
        else:
            raise ValueError("unknown IPC event")

    def receive(self):
        if self.channel is None:
            return True
        for _ in range(256):
            try:
                data, _, flags, _ = self.channel.recvmsg(PACKET_LIMIT)
                if not data:
                    self.selector.unregister(self.channel)
                    self.channel.close()
                    self.channel = None
                    # Drain-before-wait preserves explicit quit/restart intent even on exit 0.
                    self.channel_closed_at = time.monotonic()
                    return True
                if flags & (socket.MSG_TRUNC | socket.MSG_CTRUNC):
                    raise ValueError("truncated IPC packet")
                self.packet(json.loads(data))
            except BlockingIOError:
                return True
            except (ValueError, KeyError, TypeError, OSError) as error:
                self.recover("native_protocol_error_" + type(error).__name__)
                return False
        return False  # Fairness budget exhausted, not evidence that the queue is drained.

    def health(self):
        now = time.monotonic()
        observation_error = None
        try:
            live = self.identity is not None and process_identity(self.identity["pid"]) == self.identity
        except OSError as error:
            live = False
            observation_error = type(error).__name__
        evidence = "procfs"
        rows, inodes = [], set()
        try:
            rows = kernel_listeners()
            if live:
                try:
                    inodes = socket_inodes(self.identity["pid"])
                except PermissionError:
                    # File capabilities may disable ptrace-style fd inspection. Inodes
                    # come from fstat() in the owned child over the private socketpair.
                    evidence = "native_inode_and_kernel"
                    inodes = {p["inode"] for p in self.ports.values() if p.get("inode")}
                except (FileNotFoundError, ProcessLookupError):
                    live = False
        except (OSError, ValueError, IndexError) as error:
            observation_error = type(error).__name__
        ports = []
        for item in self.ports.values():
            matches = [r for r in rows if r["port"] == item["port"]]
            owned = [r for r in matches if live and r["inode"] in inodes and r["address"] == item["address"]]
            heartbeat_age = now - item["heartbeat"] if item.get("heartbeat") is not None else None
            loop_ok = heartbeat_age is not None and heartbeat_age <= HEARTBEAT_TIMEOUT
            address_ok = (item["address"] == "0.0.0.0" or
                          item["address"] == "::" and not item["ipv6_only"] or
                          item["address"] in self.network["public_ipv4"])
            healthy = bool(owned and item["listening_reported"] and item["last_bind_at"] and
                           loop_ok and address_ok and not observation_error)
            endpoint = item["address"]
            endpoint = f"[{endpoint}]:{item['port']}" if ":" in endpoint else f"{endpoint}:{item['port']}"
            ports.append({"port": item["port"], "name": item["name"], "protocol": "tcp",
                "ip_present": self.network["ip_present"], "listening": bool(matches),
                "owned_by_host": bool(owned), "local_endpoint": endpoint,
                "last_bind_at": item["last_bind_at"], "bind_age_seconds":
                    max(0, now - item["bind_monotonic"]) if item.get("bind_monotonic") is not None else None,
                "last_recovery_action": self.last_recovery, "healthy": healthy,
                "address_covers_public_ipv4": address_ok, "heartbeat_age_seconds": heartbeat_age,
                "ownership_evidence": evidence, "kernel_endpoints": matches})
        transport_healthy = bool(len(ports) == 4 and all(p["healthy"] for p in ports))
        network_ready = self.network["reason"] == "ready"
        if self.exiting or not self.desired:
            state = "stopped"
        elif self.stop_deadline is not None or self.channel_closed_at is not None:
            state = "recovering"
        elif observation_error or self.network.get("error"):
            state = "unknown"
        elif not live and self.generation:
            state = "recovering"
        elif self.conflict or self.network["reason"] == "ambiguous_public_interfaces":
            state = "conflict"
        elif not network_ready:
            state = "waiting_ipv4"
        elif transport_healthy:
            state = "ready"
        else:
            state = "starting"
        self.observation_error = observation_error
        return {"schema_version": 1, "state": state, "ready": state == "ready",
            "transport_healthy": transport_healthy, "application_ready": self.application_ready,
            "reason": self.conflict or observation_error or self.network["reason"],
            "child": {"pid": self.identity["pid"] if live else None, "generation": self.generation},
            "network": self.network, "ports": ports, "recovery_count": self.recovery_count,
            "last_recovery_action": self.last_recovery, "observed_at": utc(),
            "watchdog_checks": self.checks, "effective_ports_known": self.configured,
            "next_retry_seconds": max(0, (self.next_launch if not live else self.next_network) - now)
                if self.desired else None}

    def control_action(self, action):
        log("control_request", action=action)
        if action == "stop":
            self.desired = False
            self.stop_child()
        elif action == "start":
            self.desired = True
            self.next_launch = 0
        elif action == "restart":
            self.desired = True
            self.stop_child()
            self.next_launch = 0

    def control_request(self):
        client, _ = self.control.accept()
        with client:
            client.settimeout(0.2)
            try:
                _, uid, _ = struct.unpack("3i", client.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))
                if uid not in (0, os.geteuid()):
                    return
                request = json.loads(client.recv(4096))
                if not isinstance(request, dict):
                    return
                action = request.get("command")
                if action not in ("health", "start", "stop", "restart"):
                    return
                if action != "health":
                    self.control_action(action)
                client.sendall(json.dumps(self.health(), allow_nan=False).encode() + b"\n")
            except (OSError, ValueError, TypeError):
                return

    def tick(self):
        now = time.monotonic()
        if self.identity:
            exited = self.child.poll() is not None if self.child else process_identity(self.identity["pid"]) != self.identity
            if exited:
                if not self.receive():
                    return  # A quit intent may still be behind queued heartbeat packets.
                if self.child and self.configured and self.child.returncode in (90, 91):
                    self.control_action("stop" if self.child.returncode == 90 else "restart")
                if self.stop_deadline is None and self.desired and not self.exiting:
                    self.recover("unexpected_child_exit")
                log("child_exited", pid=self.identity["pid"], generation=self.generation,
                    returncode=self.child.returncode if self.child else None)
                if self.channel:
                    self.selector.unregister(self.channel)
                    self.channel.close()
                    self.channel = None
                if self.pidfd is not None:
                    os.close(self.pidfd)
                self.child = self.identity = self.pidfd = self.stop_deadline = None
                self.channel_closed_at = None
                self.configured = self.gate_open = self.application_ready = False
                atomic_json(self.runtime / "child.json", {"identity": None, "launch_digest": self.digest})
            elif self.stop_deadline is not None and now >= self.stop_deadline and not self.kill_sent:
                self.signal_child(signal.SIGKILL)
                self.kill_sent = True
                log("child_stop_escalated", pid=self.identity["pid"], signal="SIGKILL")
        if self.exiting:
            self.stop_child()
            return
        if self.identity and self.channel_closed_at is not None and now - self.channel_closed_at >= 1:
            # Native RAII closes IPC just before exit. Allow the reserved intent
            # exit status to arrive rather than racing normal process teardown.
            self.recover("native_channel_closed")
        if not self.identity and not self.child and self.desired and now >= self.next_launch:
            self.spawn()
        if now >= self.next_network:
            network = discover_network()
            fingerprint = json.dumps({k: v for k, v in network.items() if k != "observed_at"}, sort_keys=True)
            if fingerprint != self.fingerprint:
                log("network_revalidation", **network, generation=self.generation)
                self.network_delay = min(1.0, self.args.poll_cap)
                self.next_check = min(self.next_check, now)
                self.fingerprint = fingerprint
            self.network = network
            if network["reason"] == "ready":
                self.absent_since = None
                self.next_warning = now + self.args.address_budget
                self.next_network = now + min(5.0, self.args.poll_cap)
            else:
                if self.absent_since is None:
                    self.absent_since = now
                    self.next_warning = now + self.args.address_budget
                self.next_network = now + self.network_delay
                self.network_delay = min(self.network_delay * 2, self.args.poll_cap)
                if now >= self.next_warning:
                    log("address_budget_exceeded_still_waiting", elapsed_seconds=now - self.absent_since,
                        reason=network["reason"], next_poll_seconds=self.network_delay)
                    self.next_warning = now + self.args.address_budget
        if not self.identity or self.stop_deadline is not None or not self.desired:
            return
        if not self.configured:
            if now - self.spawn_time > self.args.startup_timeout:
                self.recover("configuration_bootstrap_timeout")
            return
        if not self.gate_open and self.network["reason"] == "ready" and now >= self.next_check:
            # The raw binary must never compete with an unrelated listener. Keep
            # waiting; freeing a conflicting port wakes the periodic retry.
            try:
                occupied = sorted({r["port"] for r in kernel_listeners() if r["port"] in self.ports and
                    (self.address_family == "both" or ":" not in r["address"] or
                     r["address"] == "::" or ipaddress.IPv6Address(r["address"]).ipv4_mapped is not None)})
                self.conflict = "foreign_listener_ports:" + str(occupied) if occupied else None
                if occupied:
                    if self.failure_pending != self.conflict:
                        log("listener_conflict_waiting", ports=occupied)
                        self.failure_pending = self.conflict
                    self.next_check = now + self.launch_delay
                    self.launch_delay = min(self.launch_delay * 2, self.args.poll_cap)
                else:
                    self.channel.send(b'{"command":"start"}')
                    self.gate_open = True
                    self.gate_time = now
                    self.next_check = now
                    log("address_gate_opened", interface=self.network["interface"],
                        public_ipv4=self.network["public_ipv4"], generation=self.generation)
            except (OSError, ValueError, IndexError) as error:
                log("gate_observation_retry", error=type(error).__name__)
                self.next_check = now + min(1.0, self.args.poll_cap)
            return
        if self.gate_open and now >= self.next_check:
            health = self.health()
            if health["transport_healthy"]:
                self.generation_healthy = True
                self.failure_pending = None
                self.launch_delay = min(1.0, self.args.poll_cap)
                self.checks += 1
                log("watchdog_check_complete_healthy", generation=self.generation,
                    ready=health["ready"], check=self.checks)
                self.next_check = now + self.args.watchdog_interval
            elif self.observation_error:
                log("watchdog_observation_unknown", reason=self.observation_error)
                self.next_check = now + min(1.0, self.args.watchdog_interval)
            elif not self.generation_healthy:
                if now - self.gate_time > self.args.startup_timeout:
                    self.recover("listener_startup_timeout")
                self.next_check = now + min(0.2, self.args.watchdog_interval)
            elif self.failure_pending is None:
                self.failure_pending = "listener_health_failed"
                self.next_check = now + min(0.25, self.args.watchdog_interval)
            else:
                self.recover("watchdog_listener_health_failed")

    def run(self):
        try:
            self.setup()
            for sig in (signal.SIGTERM, signal.SIGINT):
                signal.signal(sig, lambda _signum, _frame: setattr(self, "exiting", True))
            log("supervisor_started", pid=os.getpid(), watchdog_seconds=self.args.watchdog_interval)
            while not (self.exiting and self.identity is None and self.child is None):
                self.tick()
                for key, _ in self.selector.select(0.05):
                    if key.data == "control":
                        self.control_request()
                    elif key.data == "child":
                        self.receive()
                    else:
                        try:
                            key.fileobj.recv(65536)
                        except OSError:
                            pass
                        self.next_network = min(self.next_network, time.monotonic() + 0.1)
        finally:
            # Exception cleanup never leaves a managed child serving without an owner.
            if self.identity or self.child:
                self.signal_child(signal.SIGKILL)
            if self.child:
                try:
                    self.child.wait(timeout=2)
                except subprocess.TimeoutExpired:
                    log("child_reap_timeout", pid=self.child.pid)
            for key in list(self.selector.get_map().values()):
                self.selector.unregister(key.fileobj)
                key.fileobj.close()
            if self.pidfd is not None:
                os.close(self.pidfd)
            self.selector.close()
            self.control.close()
            if self.channel is not None:
                self.channel.close()
            (self.runtime / "control.sock").unlink(missing_ok=True)
        return 0


def control(args):
    received = False
    try:
        runtime = runtime_directory(args.runtime_dir)
        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client:
            client.settimeout(3)
            client.connect(str(runtime / "control.sock"))
            _, uid, _ = struct.unpack("3i", client.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))
            if uid != runtime.stat().st_uid:
                raise ValueError("unexpected control socket owner")
            client.sendall(json.dumps({"command": args.action}).encode())
            data = bytearray()
            while not data.endswith(b"\n"):
                part = client.recv(65536)
                if not part or len(data) + len(part) > 1024 * 1024:
                    raise ValueError("invalid health response")
                data.extend(part)
            health = json.loads(data)
            received = True
    except (OSError, ValueError) as error:
        health = unavailable(type(error).__name__)
    print(json.dumps(health, allow_nan=False), flush=True)
    if args.action != "health":
        return 0 if received else 2
    return 0 if health["ready"] else 2 if health["state"] == "unknown" else 1


def positive(value):
    number = float(value)
    if not math.isfinite(number) or number <= 0:
        raise argparse.ArgumentTypeError("must be finite and positive")
    return number


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("action", choices=("run", "health", "start", "stop", "restart"))
    parser.add_argument("--runtime-dir", default=os.path.join(
        os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.geteuid()}"), "maximizer-host"))
    parser.add_argument("--launch-config", help="owned JSON containing command array, cwd, environment")
    parser.add_argument("--json", action="store_true", help="health is always JSON")
    parser.add_argument("--poll-cap", type=positive, default=10.0)
    parser.add_argument("--watchdog-interval", type=positive, default=30.0)
    parser.add_argument("--startup-timeout", type=positive, default=15.0)
    parser.add_argument("--stop-timeout", type=positive, default=8.0)
    parser.add_argument("--address-budget", type=positive, default=600.0)
    args = parser.parse_args(argv)
    if args.action != "run":
        return control(args)
    if not args.launch_config:
        parser.error("run requires --launch-config; no executable or configuration is guessed")
    if not hasattr(os, "pidfd_open") or not hasattr(signal, "pidfd_send_signal"):
        parser.error("Linux pidfds and Python 3.10+ are required")
    try:
        # Attribute presence alone does not establish kernel/seccomp support.
        descriptor = os.pidfd_open(os.getpid())
        os.close(descriptor)
        launch = read_json(args.launch_config)
        command = launch["command"]
        environment = launch.get("environment", {})
        if (not isinstance(command, list) or not command or
                any(not isinstance(arg, str) or "\0" in arg for arg in command) or
                not os.path.isabs(command[0]) or not os.path.isabs(launch["cwd"]) or
                not isinstance(environment, dict) or any(not isinstance(k, str) or
                    not isinstance(v, str) or "\0" in k + v or "=" in k for k, v in environment.items())):
            raise ValueError("invalid launch configuration; command/cwd must be absolute, argv is an array")
        runtime = runtime_directory(args.runtime_dir, create=True)
        lock = os.open(runtime / "owner.lock", os.O_CREAT | os.O_RDWR | os.O_CLOEXEC | os.O_NOFOLLOW, 0o600)
        with os.fdopen(lock, "w"):
            try:
                fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                deadline = time.monotonic() + 1
                while True:
                    try:
                        owner = read_json(runtime / "owner.json")
                        break
                    except FileNotFoundError:
                        if time.monotonic() >= deadline:
                            raise ValueError("owner is starting; retry activation")
                        time.sleep(0.01)
                digest = hashlib.sha256(json.dumps(launch, sort_keys=True).encode()).hexdigest()
                if owner["launch_digest"] != digest:
                    raise ValueError("existing owner has a different launch configuration")
                log("duplicate_activation_no_spawn", owner_pid=owner["pid"])
                return 0
            return Supervisor(args, runtime, launch).run()
    except (OSError, ValueError, KeyError, TypeError, subprocess.SubprocessError) as error:
        log("supervisor_error", error=type(error).__name__, reason=str(error))
        return 1


if __name__ == "__main__":
    raise SystemExit(main())
