bash <<'VPS_INSTALL'
set -euo pipefail
export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
export LC_ALL=C PYTHONIOENCODING=utf-8
umask 077
if (( EUID != 0 )); then
    echo '请先执行 sudo -i，再粘贴本段。' >&2
    exit 1
fi
if ! ( : </dev/tty ) 2>/dev/null; then
    echo '需要交互终端，请在 SSH 终端中粘贴本段。' >&2
    exit 1
fi
VPS_PACKAGES=()
command -v python3 >/dev/null || VPS_PACKAGES+=(python3)
command -v sysctl >/dev/null || VPS_PACKAGES+=(procps)
command -v ip >/dev/null || VPS_PACKAGES+=(iproute2)
command -v modprobe >/dev/null || VPS_PACKAGES+=(kmod)
if ((${#VPS_PACKAGES[@]})); then
    echo "安装所需依赖：${VPS_PACKAGES[*]}"
    command -v apt-get >/dev/null || { echo '此安装器需要 Debian / Ubuntu。' >&2; exit 1; }
    apt-get update </dev/null
    DEBIAN_FRONTEND=noninteractive apt-get install -y -- "${VPS_PACKAGES[@]}" </dev/null
fi
python3 - <<'VPS_PATH_CHECK'
import os, stat
from pathlib import Path
for target in (Path('/usr/local/sbin/vps-proxy-tune'), Path('/var/lib/vps-proxy-tune/install-backups')):
    for path in [target] + list(target.parents):
        if path.is_symlink():
            raise SystemExit('拒绝安装到符号链接：' + str(path))
        if path.exists():
            s = path.stat()
            if s.st_uid != 0 or s.st_mode & 0o022:
                raise SystemExit('安装路径须由 root 管理且不可由组/其他用户写入：' + str(path))
os.makedirs('/usr/local/sbin', mode=0o755, exist_ok=True)
VPS_PATH_CHECK
VPS_SCRIPT_TMP=$(mktemp /usr/local/sbin/.vps-proxy-tune.XXXXXX)
trap 'rm -f -- "$VPS_SCRIPT_TMP"' EXIT
cat >"$VPS_SCRIPT_TMP" <<'VPS_PROGRAM'
#!/usr/bin/env bash
# VPS proxy network tuning 3.1.0
# Debian 10+ / Ubuntu 18.04+; Python 3.6+ standard library, procps, iproute2.
# No arguments = interactive menu. Legacy command-line arguments remain supported.
# Install: install -m 0755 vps-proxy-tune-v3.1.0.sh /usr/local/sbin/vps-proxy-tune
# Preview: vps-proxy-tune plan
# Apply:   vps-proxy-tune apply
# Inspect: vps-proxy-tune status --sample 5
# Undo:    vps-proxy-tune restore
#
# 自动参数是运行当时的资源分档，不是持续学习或测速；收益须由实际业务验证。
# 默认只增大缓冲 max，保留 min/default；更高的已有 max 不会被压低。
# BBR 只作用于 TCP；default_qdisc=fq 不会替换在线接口的已有 qdisc。
# 自动应用配置编辑限已识别的原生 ssserver 服务和 root 管理的 JSON/JSONC。
# Shadowsocks-rust: no_delay=true；已有较低 nofile 与服务限制协调后提高。
# realm: 系统 TCP/socket + systemd NOFILE；仅读取业务配置并指出疑似错误参数。
# 不更改密码、加密方式、节点、插件、TCP/UDP 模式、MPTCP 或 DNS。
# apply 默认失败整批回滚；--best-effort 才允许跳过不可写 sysctl。
# keep / --no-services / --app-configs off 均保留此前已应用的设置。
# restore 撤销一次；v3 检查外部修改，v2 恢复遵循原脚本覆盖语义。
# 内核模块的加载无法通过本工具回滚；进程设置需要服务重新启动后生效。
#
# Parameter references checked 2026-09-26:
# https://docs.kernel.org/networking/ip-sysctl.html
# https://docs.kernel.org/admin-guide/sysctl/net.html
# https://docs.kernel.org/admin-guide/cgroup-v2.html
# https://github.com/shadowsocks/shadowsocks-rust
# https://github.com/zhboner/realm
set -euo pipefail
export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
export LC_ALL=C
export PYTHONIOENCODING=utf-8
command -v python3 >/dev/null 2>&1 || {
    echo '需要 python3：apt-get update && apt-get install -y python3 procps iproute2' >&2
    exit 1
}
exec python3 - "$@" <<'VPS_PYTHON'
# UTF-8; intentionally compatible with Python 3.6+.
import argparse
import base64
import collections
import contextlib
import fcntl
import json
import math
import os
from pathlib import Path
import re
import shutil
import signal
import stat
import subprocess
import sys
import tempfile
import time
import uuid

VERSION = '3.1.0'
MIB = 1024 * 1024
ETC = Path('/etc')
PROC = Path('/proc')
SYS = Path('/sys')
RUN = Path('/run')
STATE = Path('/var/lib/vps-proxy-tune')
CONF_NAME = '99-zz-vps-proxy-tune.conf'
MODULE_NAME = '99-vps-proxy-tune.conf'
DROPIN = '90-vps-proxy-tune.conf'
UNIT_RE = re.compile(r'^[A-Za-z0-9_:][A-Za-z0-9_.:@-]*\.service$')
AUTO_UNIT_RE = re.compile(
    r'^(?:realm(?:-server)?|shadowsocks(?:-rust|-libev-server)?|ssserver|'
    r'xray|sing-box|hysteria-server|tuic)(?:@[^@]+)?\.service$')


class TuneError(Exception):
    pass


def info(message):
    print('[INFO] ' + str(message), flush=True)


def warn(message):
    print('[WARN] ' + str(message), file=sys.stderr, flush=True)


def read_text(path, default=''):
    try:
        return Path(path).read_text(encoding='utf-8', errors='replace').strip()
    except OSError:
        return default


def run(args, required=False, timeout=20):
    try:
        p = subprocess.run(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                           universal_newlines=True, encoding='utf-8', errors='replace', timeout=timeout)
    except (OSError, subprocess.TimeoutExpired):
        if required:
            raise TuneError('命令无法完成：' + args[0])
        return None
    if required and p.returncode:
        # Never echo arbitrary application arguments or config diagnostics.
        raise TuneError('命令失败：' + args[0])
    return p


def sysread(key):
    p = run(['sysctl', '-n', key])
    return ' '.join(p.stdout.split()) if p and p.returncode == 0 else None


def syswrite(key, value):
    p = run(['sysctl', '-q', '-w', key + '=' + value])
    return p is not None and p.returncode == 0 and sysread(key) == value


def props(unit, names):
    p = run(['systemctl', 'show', unit] + ['--property=' + n for n in names])
    if p is None or p.returncode:
        return {}
    return dict(line.split('=', 1) for line in p.stdout.splitlines() if '=' in line)


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


def safe_target(path):
    path = Path(path)
    if not path.is_absolute() or '..' in path.parts:
        raise TuneError('拒绝非绝对路径或含 .. 的路径：' + str(path))
    for part in [path] + list(path.parents):
        if part.is_symlink():
            raise TuneError('拒绝符号链接：' + str(part))
        if part.exists():
            s = part.stat()
            if s.st_uid != 0 or s.st_mode & 0o022:
                raise TuneError('写入路径须由 root 管理且不可由组/其他用户写入：' + str(part))
    if path.exists() and not path.is_file():
        raise TuneError('不是常规文件：' + str(path))


def snapshot(path):
    path = Path(path)
    safe_target(path)
    if not path.exists():
        return None
    s = path.stat()
    data = path.read_bytes()
    if len(data) > 8 * MIB:
        raise TuneError('配置文件超过 8 MiB，拒绝自动编辑：' + str(path))
    attrs = {}
    if hasattr(os, 'listxattr'):
        for name in os.listxattr(str(path)):
            attrs[name] = base64.b64encode(os.getxattr(str(path), name)).decode('ascii')
    return dict(data=base64.b64encode(data).decode('ascii'), mode=stat.S_IMODE(s.st_mode),
                uid=s.st_uid, gid=s.st_gid, atime_ns=s.st_atime_ns,
                mtime_ns=s.st_mtime_ns, xattrs=attrs)


def same_file(a, b):
    if a is None or b is None:
        return a is b
    return all(a.get(k) == b.get(k) for k in ('data', 'mode', 'uid', 'gid', 'xattrs'))


def file_image(data, original=None):
    out = dict(original or dict(mode=0o644, uid=0, gid=0, xattrs={}))
    out['data'] = base64.b64encode(data).decode('ascii')
    out.pop('atime_ns', None)
    out.pop('mtime_ns', None)
    return out


def atomic_image(path, image):
    path = Path(path)
    safe_target(path)
    if image is None:
        if path.exists():
            path.unlink()
            fsync_dir(path.parent)
        return
    path.parent.mkdir(mode=0o755, parents=True, exist_ok=True)
    safe_target(path)
    fd, name = tempfile.mkstemp(prefix='.' + path.name + '.', dir=str(path.parent))
    try:
        with os.fdopen(fd, 'wb') as f:
            f.write(base64.b64decode(image['data']))
            f.flush()
            os.fchown(f.fileno(), image['uid'], image['gid'])
            os.fchmod(f.fileno(), image['mode'])
            for attr, value in image.get('xattrs', {}).items():
                os.setxattr(name, attr, base64.b64decode(value))
            if 'mtime_ns' in image:
                os.utime(name, ns=(image['atime_ns'], image['mtime_ns']))
            os.fsync(f.fileno())
        os.replace(name, str(path))
        fsync_dir(path.parent)
    finally:
        if os.path.exists(name):
            os.unlink(name)


def write_private(path, data):
    atomic_image(path, dict(data=base64.b64encode(data).decode('ascii'),
                            mode=0o600, uid=0, gid=0, xattrs={}))


def pointer(name):
    path = STATE / name
    safe_target(path)
    value = read_text(path)
    if value and not re.fullmatch(r'txn-[A-Za-z0-9]+', value):
        raise TuneError('无效备份编号：' + name)
    return value


@contextlib.contextmanager
def locked():
    if STATE.is_symlink():
        raise TuneError('状态目录不能是符号链接')
    STATE.mkdir(mode=0o700, parents=True, exist_ok=True)
    s = STATE.stat()
    if not stat.S_ISDIR(s.st_mode) or s.st_uid != 0 or s.st_mode & 0o022:
        raise TuneError('状态目录必须是由 root 独占管理的目录')
    os.chmod(str(STATE), 0o700)
    safe_target(STATE / 'lock')
    fd = os.open(str(STATE / 'lock'), os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
    try:
        if not stat.S_ISREG(os.fstat(fd).st_mode):
            raise TuneError('无效锁文件')
        try:
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            raise TuneError('另一调优/恢复任务正在运行')
        yield
    finally:
        os.close(fd)


def unescape_mount(value):
    return re.sub(r'\\([0-7]{3})', lambda m: chr(int(m.group(1), 8)), value)


def cgroup_dirs(pid='self'):
    """Locate visible v1/v2 groups using mount roots; include every visible ancestor."""
    groups = []
    for line in read_text(PROC / str(pid) / 'cgroup').splitlines():
        fields = line.split(':', 2)
        if len(fields) == 3:
            groups.append((set(fields[1].split(',')) - {''}, fields[2]))
    found = collections.defaultdict(set)
    for line in read_text(PROC / 'self/mountinfo').splitlines():
        left, sep, right = line.partition(' - ')
        a, b = left.split(), right.split()
        if not sep or len(a) < 5 or len(b) < 3 or b[0] not in ('cgroup', 'cgroup2'):
            continue
        root, mount = unescape_mount(a[3]), Path(unescape_mount(a[4]))
        controllers = set(b[2].split(','))
        for names, group in groups:
            if b[0] == 'cgroup2':
                if names:
                    continue
                kind = 'v2'
            else:
                shared = names & controllers
                if not shared:
                    continue
                kind = ','.join(sorted(shared))
            # Namespace roots may be rebased. The visible mount root is always inspected.
            paths = [mount]
            if '..' not in Path(group).parts:
                if root == '/':
                    relative = group.lstrip('/')
                elif group == root:
                    relative = ''
                elif group.startswith(root.rstrip('/') + '/'):
                    relative = group[len(root):].lstrip('/')
                else:
                    relative = None
                if relative is not None:
                    leaf = mount / relative
                    while leaf != mount and mount in leaf.parents:
                        paths.append(leaf)
                        leaf = leaf.parent
            found[kind].update(paths)
    return found


def cpuset_count(value):
    cpus = set()
    for part in value.split(','):
        if re.fullmatch(r'\d+', part):
            cpus.add(int(part))
        elif re.fullmatch(r'\d+-\d+', part):
            lo, hi = map(int, part.split('-'))
            if 0 <= lo <= hi < 65536:
                cpus.update(range(lo, hi + 1))
    return len(cpus)


def resources(pid='self'):
    raw_mem = re.search(r'^MemTotal:\s+(\d+)', read_text(PROC / 'meminfo'), re.M)
    if not raw_mem:
        raise TuneError('无法读取 MemTotal')
    host_mem = int(raw_mem.group(1)) * 1024
    memory = host_mem
    try:
        cpus = float(len(os.sched_getaffinity(0 if pid == 'self' else int(pid))))
    except (AttributeError, OSError, ValueError):
        cpus = float(os.cpu_count() or 1)
    constrained = False
    for kind, dirs in cgroup_dirs(pid).items():
        names = set(kind.split(','))
        for path in dirs:
            mem_name = 'memory.max' if kind == 'v2' else 'memory.limit_in_bytes'
            if kind == 'v2' or 'memory' in names:
                value = read_text(path / mem_name)
                if value.isdigit() and 0 < int(value) < memory:
                    memory, constrained = int(value), True
                # v2 memory.high is a throttle threshold; use it as a conservative sizing bound.
                high = read_text(path / 'memory.high') if kind == 'v2' else ''
                if high.isdigit() and 0 < int(high) < memory:
                    memory, constrained = int(high), True
            if kind == 'v2' or 'cpu' in names:
                if kind == 'v2':
                    values = read_text(path / 'cpu.max').split()
                else:
                    values = [read_text(path / 'cpu.cfs_quota_us'), read_text(path / 'cpu.cfs_period_us')]
                if len(values) == 2 and all(x.isdigit() for x in values) and int(values[0]) > 0 and int(values[1]) > 0:
                    quota = int(values[0]) / int(values[1])
                    if quota < cpus:
                        cpus, constrained = quota, True
            if kind == 'v2' or 'cpuset' in names:
                count = cpuset_count(read_text(path / ('cpuset.cpus.effective' if kind == 'v2' else 'cpuset.cpus')))
                if count and count < cpus:
                    cpus, constrained = float(count), True
    return dict(host_memory=host_mem, memory=memory, cpus=cpus,
                host_cpus=os.cpu_count() or 1, constrained=constrained)


def tiers(mem_mib):
    # Buffer floor, descriptor target and per-listener backlog ceiling.
    for ceiling, buffer, nofile, backlog in (
            (128, 2, 16384, 1024), (256, 4, 16384, 1024),
            (512, 8, 32768, 2048), (1024, 16, 65536, 4096),
            (4096, 32, 131072, 8192)):
        if mem_mib <= ceiling:
            return buffer, nofile, backlog
    return 64, 262144, 16384


def sizing(hw, args):
    mem = max(1, hw['memory'] // MIB)
    buf, nofile, max_backlog = tiers(mem)
    factor = 2 if args.profile == 'throughput' else 1
    ceiling = max(1, min(256, mem // (16 if factor == 2 else 32)))
    buf = min(ceiling, buf * factor)
    requested = None
    if args.bandwidth_mbps is not None:
        # 2*BDP: Mbps * ms * 250 bytes. Hints are supplied, not inferred from virtual NIC speed.
        requested = max(1, int(math.ceil(args.bandwidth_mbps * args.rtt_ms * 250 / MIB)))
        buf = min(ceiling, max(buf, requested))
    if args.buffer_mib is not None:
        buf = args.buffer_mib
    cpus = max(1, int(math.ceil(hw['cpus'])))
    backlog = min(max_backlog * factor, max(1024, cpus * 1024 * factor))
    return dict(buffer_mib=buf, nofile=args.nofile or nofile,
                backlog=backlog, ceiling_mib=ceiling, bdp_mib=requested)


def is_container():
    p = run(['systemd-detect-virt', '--container', '--quiet'])
    return bool((p is not None and p.returncode == 0) or Path('/.dockerenv').exists()
                or (RUN / '.containerenv').exists() or (PROC / 'vz').exists())


def detect_services(args, systemd):
    if args.no_services:
        return []
    if args.service and not systemd:
        raise TuneError('显式指定服务需要正在运行的 systemd')
    if not systemd:
        return []
    names = list(args.service)
    if not names:
        for verb in ('list-unit-files', 'list-units'):
            p = run(['systemctl', verb, '--type=service', '--all', '--no-legend', '--no-pager', '--plain'])
            if p is None or p.returncode:
                continue
            for line in p.stdout.splitlines():
                parts = line.split()
                if parts and AUTO_UNIT_RE.fullmatch(parts[0]):
                    names.append(parts[0])
    out = []
    for name in sorted(set(names)):
        if not UNIT_RE.fullmatch(name) or name.endswith('@.service'):
            if args.service:
                raise TuneError('服务名必须是具体的 .service 单元')
            continue
        p = props(name, ['Id', 'LoadState', 'MainPID', 'LimitNOFILE', 'LimitNOFILESoft',
                         'MemoryMax', 'TasksMax', 'CPUQuotaPerSecUSec'])
        if p.get('LoadState') != 'loaded':
            if args.service:
                raise TuneError('服务不存在/未加载：' + name)
            continue
        canonical = p.get('Id', name)
        if not UNIT_RE.fullmatch(canonical):
            raise TuneError('无法确认服务规范名称')
        if any(s['name'] == canonical for s in out):
            continue
        out.append(dict(name=canonical, props=p))
    return out


def parse_limit(value):
    if value in ('infinity', '18446744073709551615'):
        return None
    if not value or not value.isdigit():
        raise TuneError('无法读取 systemd LimitNOFILE')
    return int(value)


def limit_string(value):
    return 'infinity' if value is None else str(value)


def service_plans(services, size, args):
    nr = sysread('fs.nr_open')
    if not nr or not nr.isdigit():
        if services:
            raise TuneError('无法读取 fs.nr_open')
        return
    nr = int(nr)
    if args.nofile and args.nofile > nr:
        raise TuneError('--nofile 大于内核 fs.nr_open')
    for service in services:
        p = service['props']
        target = size['nofile']
        pid = p.get('MainPID', '')
        if not args.nofile and pid.isdigit() and int(pid):
            try:
                target = min(target, tiers(resources(pid)['memory'] // MIB)[1])
            except TuneError:
                pass
        if not args.nofile and p.get('MemoryMax', '').isdigit():
            target = min(target, tiers(int(p['MemoryMax']) // MIB)[1])
        target = min(target, nr)
        soft = parse_limit(p.get('LimitNOFILESoft'))
        hard = parse_limit(p.get('LimitNOFILE'))
        soft = None if soft is None else max(soft, target)
        hard = None if soft is None or hard is None else max(hard, soft)
        service.update(target=target, soft=soft, hard=hard)


def build_plan(args, size, loaded=False):
    plan = collections.OrderedDict()

    def put(key, value, grow=False):
        old = sysread(key)
        if old is None:
            warn('内核未提供 {}，跳过'.format(key))
            return
        if grow:
            if not old.isdigit():
                raise TuneError('参数不是预期的整数：' + key)
            value = max(int(old), int(value))
        plan[key] = str(value)

    cap = size['buffer_mib'] * MIB
    grow = args.buffer_policy == 'grow'
    for key, default_key in (('net.core.rmem_max', 'net.core.rmem_default'),
                             ('net.core.wmem_max', 'net.core.wmem_default')):
        minimum = sysread(default_key)
        put(key, max(cap, int(minimum) if minimum and minimum.isdigit() else 0), grow)
    for key in ('net.ipv4.tcp_rmem', 'net.ipv4.tcp_wmem'):
        parts = (sysread(key) or '').split()
        if len(parts) != 3 or not all(x.isdigit() for x in parts):
            warn('无法读取缓冲三元组：' + key)
            continue
        low, middle, high = map(int, parts)
        high = max(cap, low, middle, high if grow else 0)
        put(key, '{} {} {}'.format(low, middle, high))
    for key in ('net.ipv4.tcp_moderate_rcvbuf', 'net.ipv4.tcp_window_scaling', 'net.ipv4.tcp_sack'):
        put(key, 1)
    for key in ('net.core.somaxconn', 'net.ipv4.tcp_max_syn_backlog'):
        put(key, size['backlog'], True)
    if args.mtu_probing != 'keep':
        put('net.ipv4.tcp_mtu_probing', 1, True)
    available = (sysread('net.ipv4.tcp_available_congestion_control') or '').split()
    current = sysread('net.ipv4.tcp_congestion_control')
    chosen = args.cc
    if chosen == 'auto':
        if current not in ('cubic', 'reno', 'bbr'):
            info('保留现有拥塞算法：{}（包括定制内核算法）'.format(current))
            chosen = 'keep'
        else:
            chosen = 'bbr' if 'bbr' in available else 'keep'
            if chosen == 'keep':
                info('BBR 尚未注册；{}，否则保留当前算法'.format('apply 将尝试加载 tcp_bbr' if not loaded else '加载不可用'))
    if chosen != 'keep':
        if chosen in available:
            put('net.ipv4.tcp_congestion_control', chosen)
        elif loaded:
            raise TuneError('内核不支持指定的拥塞算法：' + chosen)
        else:
            info('apply 时探测模块：tcp_' + chosen)
    if args.qdisc == 'fq':
        put('net.core.default_qdisc', 'fq')
    if args.tfo != 'keep':
        fast = sysread('net.ipv4.tcp_fastopen')
        if fast is not None and fast.isdigit():
            put('net.ipv4.tcp_fastopen', (int(fast) | 3) if args.tfo == 'on' else 0)
    return plan


def sanitize_json(text):
    """JSONC comments and trailing commas -> spaces, preserving character offsets."""
    chars = list(text)
    i = 0
    string = False
    while i < len(chars):
        c = text[i]
        if string:
            if c == '\\':
                i += 2
                continue
            if c == '"':
                string = False
            i += 1
            continue
        if c == '"':
            string = True
            i += 1
        elif text.startswith('//', i):
            end = text.find('\n', i)
            end = len(text) if end < 0 else end
            for j in range(i, end):
                chars[j] = ' '
            i = end
        elif text.startswith('/*', i):
            end = text.find('*/', i + 2)
            if end < 0:
                raise ValueError('unclosed comment')
            for j in range(i, end + 2):
                if chars[j] not in '\r\n':
                    chars[j] = ' '
            i = end + 2
        else:
            i += 1
    clean = ''.join(chars)
    # Tokenize strings so commas inside passwords/URLs remain untouched.
    for m in re.finditer(r'"(?:\\.|[^"\\])*"|,(?=\s*[}\]])', clean):
        if m.group(0) == ',':
            chars[m.start()] = ' '
    return ''.join(chars)


def unique_object(pairs):
    out = {}
    for key, value in pairs:
        if key in out:
            raise ValueError('duplicate key')
        out[key] = value
    return out


def json_document(text):
    clean = sanitize_json(text)
    data = json.loads(clean, object_pairs_hook=unique_object,
                      parse_constant=lambda x: (_ for _ in ()).throw(ValueError(x)))
    if not isinstance(data, dict):
        raise ValueError('not object')
    return data, clean


def patch_ss(text, target=None):
    data, clean = json_document(text)
    if 'no_delay' in data and not isinstance(data['no_delay'], bool):
        raise ValueError('no_delay must be boolean')
    changes = {}
    if data.get('no_delay') is not True:
        changes['no_delay'] = True
    if target is not None and 'nofile' in data:
        n = data['nofile']
        if type(n) is not int or n <= 0:
            raise ValueError('invalid nofile')
        if n < target:
            changes['nofile'] = target
    if not changes:
        return text, []
    decoder = json.JSONDecoder()
    i = clean.index('{') + 1
    spans = {}
    while True:
        while i < len(clean) and clean[i].isspace():
            i += 1
        if clean[i] == '}':
            break
        key, i = decoder.raw_decode(clean, i)
        while clean[i].isspace():
            i += 1
        if clean[i] != ':':
            raise ValueError('missing colon')
        i += 1
        while clean[i].isspace():
            i += 1
        start = i
        _, i = decoder.raw_decode(clean, i)
        spans[key] = (start, i)
        while clean[i].isspace():
            i += 1
        if clean[i] == ',':
            i += 1
        elif clean[i] != '}':
            raise ValueError('missing comma')
    edits = []
    missing = []
    for key, value in changes.items():
        if key in spans:
            lo, hi = spans[key]
            edits.append((lo, hi, json.dumps(value)))
        else:
            missing.append('"{}": {}'.format(key, json.dumps(value)))
    if missing:
        start = clean.index('{') + 1
        edits.append((start, start, '\n  ' + ',\n  '.join(missing) + (',' if data else '') + '\n'))
    new = text
    for lo, hi, value in sorted(edits, reverse=True):
        new = new[:lo] + value + new[hi:]
    expected = dict(data)
    expected.update(changes)
    if json_document(new)[0] != expected:
        raise ValueError('round-trip mismatch')
    return new, ['{}={}'.format(k, str(v).lower()) for k, v in changes.items()]


def process_app(service):
    pid = service['props'].get('MainPID', '')
    if not pid.isdigit() or not int(pid):
        return None
    try:
        exe = os.readlink(str(PROC / pid / 'exe'))
        argv = (PROC / pid / 'cmdline').read_bytes().decode('utf-8').rstrip('\0').split('\0')
        # Only native processes sharing this filesystem root are edited.
        a, b = (PROC / pid / 'root').stat(), Path('/').stat()
        if (a.st_dev, a.st_ino) != (b.st_dev, b.st_ino):
            return None
    except (OSError, UnicodeError):
        return None
    base = Path(exe.removesuffix(' (deleted)') if hasattr(exe, 'removesuffix') else exe.replace(' (deleted)', '')).name
    kind = 'ss' if base == 'ssserver' else 'realm' if base == 'realm' else None
    if kind is None:
        return None
    configs = []
    for i, arg in enumerate(argv[1:], 1):
        if arg in ('-c', '--config') and i + 1 < len(argv):
            configs.append(argv[i + 1])
        elif arg.startswith('--config='):
            configs.append(arg.split('=', 1)[1])
        if arg in ('--nofile', '-n') or arg.startswith('--nofile='):
            warn('{}: 启动参数含 NOFILE 覆盖项，请核对运行中进程限制'.format(service['name']))
    if len(configs) != 1:
        return dict(kind=kind, path=None, unit=service)
    path = Path(configs[0])
    if not path.is_absolute():
        try:
            path = Path(os.readlink(str(PROC / pid / 'cwd'))) / path
        except OSError:
            return None
    return dict(kind=kind, path=path, unit=service)


def realm_audit(path):
    try:
        if path.suffix.lower() == '.json':
            data, _ = json_document(path.read_text(encoding='utf-8'))
        else:
            try:
                import tomllib
            except ImportError:
                info('realm TOML 深度检查需 Python 3.11+；系统及服务优化仍可用')
                return
            data = tomllib.loads(path.read_text(encoding='utf-8'))
        networks = [('global', data.get('network', {}))]
        for i, endpoint in enumerate(data.get('endpoints', [])):
            if isinstance(endpoint, dict):
                networks.append(('endpoint #{}'.format(i + 1), endpoint.get('network', {})))
        for label, network in networks:
            if not isinstance(network, dict):
                continue
            for key, upper in (('tcp_timeout', 30), ('udp_timeout', 300)):
                value = network.get(key)
                if type(value) is int and (value == 0 or value > upper):
                    warn('realm {}: {}={}，请按业务复核（tcp_timeout 是连接超时）'.format(label, key, value))
        info('realm: 保留业务配置；套用 TCP/socket 调优及服务 NOFILE')
    except (OSError, ValueError, TypeError, AttributeError):
        warn('realm 配置无法可靠解析，仅应用系统与服务级优化')


def application_plans(args, services):
    apps = []
    if args.app_configs == 'auto':
        for service in services:
            app = process_app(service)
            if app:
                apps.append(app)
    for name in args.ss_config:
        path = Path(name)
        if not path.is_absolute():
            raise TuneError('--ss-config 必须使用绝对路径')
        apps.append(dict(kind='ss', path=path, unit=None, explicit=True))
    # Shared files can serve instances with different cgroup/descriptor limits.
    # A shared application nofile must not exceed the smallest planned target.
    shared_targets = collections.defaultdict(list)
    for app in apps:
        if app['path'] is not None and app['unit'] and 'target' in app['unit']:
            shared_targets[str(app['path'])].append(app['unit']['target'])
    out = []
    seen = set()
    for app in apps:
        path = app['path']
        if path is None:
            info('{}: 无法唯一识别配置路径，保留应用配置'.format(app['kind']))
            continue
        if str(path) in seen:
            continue
        seen.add(str(path))
        if app['kind'] == 'realm':
            if path.is_file():
                realm_audit(path)
            else:
                info('realm 使用配置目录/环境配置；仅做系统及服务优化')
            continue
        try:
            before = snapshot(path)
            if before is None:
                raise ValueError('missing file')
            old = base64.b64decode(before['data']).decode('utf-8')
            unit = app['unit']
            # Changing nofile without controlling the service's inherited limit could break startup.
            targets = shared_targets.get(str(path), [])
            target = min(targets) if targets else None
            if len(set(targets)) > 1:
                warn('多个服务共享 {}；应用 NOFILE 按最低服务目标处理'.format(path))
            new, changes = patch_ss(old, target)
            parsed, _ = json_document(old)
            for key in ('inbound_recv_buffer_size', 'outbound_recv_buffer_size',
                        'inbound_send_buffer_size', 'outbound_send_buffer_size'):
                if parsed.get(key):
                    warn('Shadowsocks {} 已显式设置；该 socket 可能绕过 TCP 自动缓冲选择'.format(key))
            if parsed.get('plugin') or any(isinstance(s, dict) and s.get('plugin') for s in parsed.get('servers', [])):
                info('Shadowsocks 插件保留现状；插件自身 socket 参数不由 ssserver 完全控制')
            if changes:
                out.append(dict(path=str(path), before=before, after=file_image(new.encode('utf-8'), before),
                                summary=', '.join(changes)))
                info('Shadowsocks {}: {}'.format(path, ', '.join(changes)))
            else:
                info('Shadowsocks {}: 相关配置已满足目标'.format(path))
        except (OSError, ValueError, UnicodeError, TuneError, TypeError) as e:
            if app.get('explicit'):
                raise TuneError('无法安全编辑指定 Shadowsocks-rust JSON/JSONC：' + str(path))
            warn('Shadowsocks 配置不能可靠编辑，已跳过：' + str(path))
    return out


def merged_sysctl(plan):
    path = ETC / 'sysctl.d' / CONF_NAME
    text = read_text(path)
    values = collections.OrderedDict()
    if text:
        if 'Managed by vps-proxy-tune' not in text:
            raise TuneError('目标 sysctl 文件不属于本工具，拒绝覆盖：' + str(path))
        for line in text.splitlines():
            line = line.strip()
            if not line or line.startswith(('#', ';')):
                continue
            match = re.fullmatch(r'([a-zA-Z0-9_.]+)\s*=\s*(.*?)\s*', line)
            if not match:
                raise TuneError('已有工具配置含无法识别的行')
            values[match.group(1)] = match.group(2)
    values.update(plan)
    return ('# Managed by vps-proxy-tune ' + VERSION + '; verified keys only.\n' +
            '# keep preserves previously managed entries; restore undoes one apply.\n' +
            ''.join('{} = {}\n'.format(k, v) for k, v in values.items())).encode('utf-8')


def conflicts(plan):
    files = []
    for folder in (ETC / 'sysctl.d', RUN / 'sysctl.d', Path('/usr/local/lib/sysctl.d'), Path('/usr/lib/sysctl.d')):
        files.extend(folder.glob('*.conf'))
    files.append(ETC / 'sysctl.conf')
    found = []
    for path in sorted(set(files)):
        if path == ETC / 'sysctl.d' / CONF_NAME or not path.is_file():
            continue
        for line in read_text(path).splitlines():
            m = re.match(r'^\s*-?([a-zA-Z0-9_./]+)\s*=\s*([^#;]+)', line)
            if m:
                key = m.group(1).replace('/', '.')
                if key in plan and ' '.join(m.group(2).split()) != plan[key]:
                    found.append('{}: {}'.format(path, key))
    for item in found[:12]:
        warn('检测到其他持久化值（实际覆盖取决于加载顺序）：' + item)
    if len(found) > 12:
        warn('其余 {} 项冲突省略；重启后请用 status 核对'.format(len(found) - 12))


class Transaction:
    def __init__(self):
        self.name = 'txn-' + uuid.uuid4().hex[:16]
        self.path = STATE / self.name
        self.path.mkdir(mode=0o700)
        self.data = dict(version=3, status='pending', parent=pointer('latest'),
                         created=time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()),
                         sysctl=[], files=[], services=[])
        self.flush()
        write_private(STATE / 'pending', (self.name + '\n').encode())

    def flush(self):
        write_private(self.path / 'journal.json', json.dumps(self.data, ensure_ascii=False, indent=2).encode('utf-8'))

    def file(self, path, after, before=None, enforce_before=False):
        path = Path(path)
        original = snapshot(path)
        if enforce_before and not same_file(original, before):
            raise TuneError('预览后配置已改变，取消应用：' + str(path))
        if same_file(original, after):
            return
        self.data['files'].append(dict(path=str(path), before=original, after=after))
        self.flush()  # durable undo record before every mutation
        atomic_image(path, after)

    def sysctl(self, key, value, best_effort=False):
        old = sysread(key)
        if old is None:
            raise TuneError('应用前参数变为不可读：' + key)
        if old == value:
            return True
        entry = dict(key=key, before=old, after=value)
        self.data['sysctl'].append(entry)
        self.flush()
        if syswrite(key, value):
            return True
        if sysread(key) != old and not syswrite(key, old):
            raise TuneError('写入失败且无法恢复：' + key)
        # Failed key is back to its original value. Keep it out of persistence.
        self.data['sysctl'].remove(entry)
        self.flush()
        if best_effort:
            warn('写入被拒绝/回读不符，保留原值：' + key)
            return False
        raise TuneError('写入失败/回读不符：' + key)

    def commit(self):
        self.data['status'] = 'committed'
        self.flush()
        write_private(STATE / 'latest', (self.name + '\n').encode())
        atomic_image(STATE / 'pending', None)


def load_journal(name):
    if not re.fullmatch(r'txn-[A-Za-z0-9]+', name):
        raise TuneError('无效备份编号')
    directory = STATE / name
    safe_target(directory / 'journal.json')
    if (directory / 'journal.json').exists():
        data = json.loads((directory / 'journal.json').read_text())
        if data.get('version') != 3:
            raise TuneError('未知备份格式')
        return data
    # Backward-compatible restore of the user's v2.0.0 format.
    for required in ('files.tsv', 'sysctl.tsv', 'parent'):
        if not (directory / required).is_file():
            raise TuneError('备份不完整：' + name)
    data = dict(version=2, parent=read_text(directory / 'parent'), files=[], sysctl=[], services=[])
    for line in read_text(directory / 'files.tsv').splitlines():
        kind, filename = line.split('\t', 1)
        path = Path(filename)
        allowed = (path == ETC / 'sysctl.d' / CONF_NAME or
                   (str(path).startswith(str(ETC / 'systemd/system') + '/') and path.name == DROPIN))
        if not allowed or kind not in ('present', 'absent'):
            raise TuneError('无效 v2 备份记录')
        before = snapshot(directory / 'files' / filename.lstrip('/')) if kind == 'present' else None
        if kind == 'present' and before is None:
            raise TuneError('v2 文件备份缺失')
        data['files'].append(dict(path=str(path), before=before, after=None))
    for line in read_text(directory / 'sysctl.tsv').splitlines():
        key, value = line.split('\t', 1)
        data['sysctl'].append(dict(key=key, before=value, after=None))
    data['services'] = read_text(directory / 'services.txt').splitlines()
    return data


def restore_data(name, systemd, force=False):
    data = load_journal(name)
    parent = data.get('parent', '')
    if parent and not re.fullmatch(r'txn-[A-Za-z0-9]+', parent):
        raise TuneError('无效父备份编号')
    collisions = []
    for item in data['files']:
        current = snapshot(Path(item['path']))
        if data['version'] == 3 and not force and not any(same_file(current, item[k]) for k in ('before', 'after')):
            collisions.append(item['path'])
    for item in data['sysctl']:
        if not re.fullmatch(r'[A-Za-z0-9_.]+', item['key']):
            raise TuneError('无效 sysctl 备份键')
        current = sysread(item['key'])
        if data['version'] == 3 and not force and current not in (item['before'], item['after']):
            collisions.append(item['key'])
    if collisions:
        raise TuneError('检测到应用后的外部修改，未恢复：{}。确认覆盖后使用 restore --force'.format(', '.join(collisions)))
    if data['version'] == 2:
        warn('v2 备份没有应用后指纹；恢复行为与原版一致，会覆盖已记录项')
    failures = []
    for item in reversed(data['files']):
        try:
            atomic_image(Path(item['path']), item['before'])
        except (OSError, TuneError) as e:
            failures.append(str(item['path']))
    for item in reversed(data['sysctl']):
        if sysread(item['key']) != item['before'] and not syswrite(item['key'], item['before']):
            failures.append(item['key'])
    if data['services']:
        if systemd:
            p = run(['systemctl', 'daemon-reload'])
            if p is None or p.returncode:
                failures.append('systemctl daemon-reload')
        else:
            warn('systemd 当前未运行；恢复的服务文件将在其下次加载时生效')
    if failures:
        raise TuneError('恢复未完全成功，可重试：' + ', '.join(failures))
    if parent:
        write_private(STATE / 'latest', (parent + '\n').encode())
    else:
        atomic_image(STATE / 'latest', None)
    atomic_image(STATE / 'pending', None)
    if data['version'] == 3:
        data['status'] = 'restored'
        write_private(STATE / name / 'journal.json', json.dumps(data, ensure_ascii=False, indent=2).encode('utf-8'))
    info('已恢复应用前状态：' + name)
    info('回滚未卸载模块；如重启过代理，进程限制/应用配置需再次重启才会恢复')


def prepare_modules(args, container):
    if container or not shutil.which('modprobe'):
        return []
    available = (sysread('net.ipv4.tcp_available_congestion_control') or '').split()
    current = sysread('net.ipv4.tcp_congestion_control')
    modules = []
    if args.cc == 'bbr' or (args.cc == 'auto' and current in ('cubic', 'reno', 'bbr')):
        modules.append('tcp_bbr')
    elif args.cc == 'cubic':
        modules.append('tcp_cubic')
    if args.qdisc == 'fq':
        modules.append('sch_fq')
    success = []
    for module in modules:
        p = run(['modprobe', module])
        if p is not None and p.returncode == 0:
            success.append(module)
        elif module == 'tcp_bbr' and args.cc == 'auto':
            info('tcp_bbr 模块不可加载，稍后按可用算法选择')
    return success


def apply(args, hw, size, services, apps, systemd, container):
    with locked():
        if pointer('pending'):
            raise TuneError('存在中断事务，请先运行 restore')
        if ((ETC / 'sysctl.d/99-ss-optimization.conf').exists() or
                (ETC / 'systemd/system/ss-network-tuning.service').exists()):
            raise TuneError('检测到旧 ss-network 调优配置，请先停用旧调优器并处理其配置')
        tx = Transaction()
        try:
            modules = prepare_modules(args, container)
            plan = build_plan(args, size, loaded=True)
            conflicts(plan)
            verified = collections.OrderedDict()
            skipped = []
            for key, value in plan.items():
                if tx.sysctl(key, value, args.best_effort and not (key == 'net.ipv4.tcp_congestion_control' and args.cc != 'auto')):
                    verified[key] = value
                    info('{} = {}'.format(key, value))
                else:
                    skipped.append(key)
            before = snapshot(ETC / 'sysctl.d' / CONF_NAME)
            data = merged_sysctl(verified)
            # A previous managed value for a newly failed key must not be reapplied at boot.
            if skipped:
                lines = data.decode().splitlines(True)
                data = ''.join(line for line in lines if line.split('=', 1)[0].strip() not in skipped).encode()
            tx.file(ETC / 'sysctl.d' / CONF_NAME, file_image(data, before))
            if modules:
                path = ETC / 'modules-load.d' / MODULE_NAME
                before = snapshot(path)
                previous = read_text(path)
                if previous and 'Managed by vps-proxy-tune' not in previous:
                    raise TuneError('模块配置不属于本工具，拒绝覆盖')
                names = {line.strip() for line in previous.splitlines() if line.strip() and not line.startswith('#')}
                names.update(modules)
                content = '# Managed by vps-proxy-tune\n' + '\n'.join(sorted(names)) + '\n'
                tx.file(path, file_image(content.encode(), before))
            for service in services:
                unit = service['name']
                path = ETC / 'systemd/system' / (unit + '.d') / DROPIN
                before = snapshot(path)
                # Accept v2's exact two-line drop-in; reject unrelated content at the reserved filename.
                previous = read_text(path)
                if previous and not re.fullmatch(r'(?:# Managed by vps-proxy-tune[^\n]*\n)?\[Service\]\nLimitNOFILE=[0-9a-z:]+', previous):
                    raise TuneError('服务 drop-in 含其他配置，拒绝覆盖：' + str(path))
                content = '# Managed by vps-proxy-tune\n[Service]\nLimitNOFILE={}:{}\n'.format(
                    limit_string(service['soft']), limit_string(service['hard']))
                tx.data['services'].append(unit)
                tx.flush()
                tx.file(path, file_image(content.encode(), before))
            if services:
                run(['systemctl', 'daemon-reload'], required=True)
                for service in services:
                    now = props(service['name'], ['LimitNOFILESoft', 'LimitNOFILE'])
                    soft = parse_limit(now.get('LimitNOFILESoft'))
                    hard = parse_limit(now.get('LimitNOFILE'))
                    expected_soft, expected_hard = service['soft'], service['hard']
                    if ((expected_soft is None and soft is not None) or
                        (soft is not None and soft < expected_soft) or
                        (expected_hard is None and hard is not None) or
                        (hard is not None and hard < expected_hard)):
                        raise TuneError('更高优先级的 drop-in 覆盖了 NOFILE：' + service['name'])
            for app in apps:
                tx.file(Path(app['path']), app['after'], app['before'], enforce_before=True)
                info('已保存应用配置：{} ({})'.format(app['path'], app['summary']))
            tx.commit()
        except BaseException:
            warn('应用未完成，尝试自动回滚：' + tx.name)
            # Do not let a second Ctrl-C interrupt the rollback. SIGKILL remains recoverable via pending.
            signal.signal(signal.SIGINT, signal.SIG_IGN)
            signal.signal(signal.SIGTERM, signal.SIG_IGN)
            try:
                restore_data(tx.name, systemd, force=True)
            except BaseException as e:
                warn('自动回滚未完成；保留 pending，请执行 restore --force。' + str(e))
            raise
        info('应用成功；本次跳过 {} 项；备份 {}'.format(len(skipped), tx.path))
        info('服务没有重启；NOFILE 与应用配置在其下次重启后生效')
        info('fq 仅写入 default_qdisc；当前网卡队列请用 status 核验')
        info('撤销本次：vps-proxy-tune restore；可逐次恢复到 v2 的备份')


def hardware_report(hw, size):
    info('v{} | {} | {}'.format(VERSION, os.uname().release, os.uname().machine))
    info('宿主可见内存 {} MiB；本环境可用上界 {} MiB；CPU {:.2f}/{} 核'.format(
        hw['host_memory'] // MIB, hw['memory'] // MIB, hw['cpus'], hw['host_cpus']))
    info('缓冲目标 {} MiB；NOFILE 基础目标 {}；监听队列目标 {}'.format(
        size['buffer_mib'], size['nofile'], size['backlog']))
    if size['bdp_mib'] is not None:
        info('带宽/RTT 提示的 2×BDP={} MiB；自动新增上限保护={} MiB'.format(size['bdp_mib'], size['ceiling_mib']))
        if size['bdp_mib'] > size['buffer_mib']:
            warn('BDP 超过本次目标；高带宽长 RTT 单流可能仍受窗口限制')
    if hw['memory'] < 128 * MIB:
        warn('内存少于 128 MiB；增大 socket 上限不等于能承载大量并发')


def show_plan(args, size, services, apps):
    plan = build_plan(args, size)
    print('\n{:<42} {:<24} {}'.format('sysctl', 'current', 'planned'))
    for key, value in plan.items():
        print('{:<42} {:<24} {}'.format(key, sysread(key) or 'unavailable', value))
    for service in services:
        info('{}: LimitNOFILE={}:{}（下次重启生效）'.format(service['name'],
             limit_string(service['soft']), limit_string(service['hard'])))
    conflicts(plan)
    if not services:
        info('未识别到代理 systemd 单元；可用 --service 指定')
    info('以上是只读预览；未加载模块、未写入文件、未重启服务')
    info('默认保留更大的缓冲上限；keep 保留原有工具配置；restore 撤销一次 apply')


def counters():
    data = {}
    for name in ('net/snmp', 'net/netstat'):
        lines = read_text(PROC / name).splitlines()
        for i in range(0, len(lines) - 1, 2):
            keys, values = lines[i].split(), lines[i + 1].split()
            if not keys or not values or keys[0] != values[0]:
                continue
            for key, value in zip(keys[1:], values[1:]):
                if value.isdigit():
                    data[keys[0].rstrip(':') + key] = int(value)
    drops, squeezed = 0, 0
    for line in read_text(PROC / 'net/softnet_stat').splitlines():
        fields = line.split()
        if len(fields) >= 3:
            drops += int(fields[1], 16)
            squeezed += int(fields[2], 16)
    data['SoftnetDropped'] = drops
    data['SoftnetTimeSqueeze'] = squeezed
    return data


def interface_report():
    devices = set()
    for family in ('-4', '-6'):
        p = run(['ip', '-j', family, 'route', 'show', 'default'])
        if p is not None and p.returncode == 0:
            try:
                for route in json.loads(p.stdout):
                    if route.get('dev'):
                        devices.add(route['dev'])
                    for nexthop in route.get('nexthops', []):
                        if nexthop.get('dev'):
                            devices.add(nexthop['dev'])
            except (ValueError, TypeError):
                pass
    for dev in sorted(devices):
        base = SYS / 'class/net' / dev
        mtu, speed = read_text(base / 'mtu', '?'), read_text(base / 'speed', '?')
        rx = len(list((base / 'queues').glob('rx-*')))
        tx = len(list((base / 'queues').glob('tx-*')))
        info('{}: MTU={}，RX/TX 队列={}/{}，驱动报告速率={} Mb/s（非套餐带宽）'.format(dev, mtu, rx, tx, speed))
    p = run(['tc', '-s', 'qdisc', 'show'])
    if p is not None and p.returncode == 0:
        print('\n网卡实际 qdisc：\n' + p.stdout.rstrip())


def status(args, services):
    managed = {}
    for line in read_text(ETC / 'sysctl.d' / CONF_NAME).splitlines():
        if line and not line.startswith('#') and '=' in line:
            k, v = line.split('=', 1)
            managed[k.strip()] = v.strip()
    base = ['net.ipv4.tcp_available_congestion_control', 'net.ipv4.tcp_congestion_control',
            'net.core.default_qdisc', 'net.core.rmem_max', 'net.core.wmem_max',
            'net.ipv4.tcp_rmem', 'net.ipv4.tcp_wmem', 'net.ipv4.tcp_fastopen',
            'net.ipv4.tcp_mtu_probing', 'net.core.somaxconn', 'net.ipv4.tcp_max_syn_backlog']
    for key in list(dict.fromkeys(base + list(managed))):
        value = sysread(key)
        drift = ' [与持久化值不符: {}]'.format(managed[key]) if key in managed and value != managed[key] else ''
        print('{:<44} {}{}'.format(key, value or 'unavailable', drift))
    interface_report()
    p = run(['ss', '-s'])
    if p is not None and p.returncode == 0:
        print('\nSocket 汇总：\n' + p.stdout.rstrip())
    for service in services:
        p = service['props']
        info('{}: 配置 soft/hard={}/{}，MemoryMax={}，TasksMax={}'.format(service['name'],
             p.get('LimitNOFILESoft'), p.get('LimitNOFILE'), p.get('MemoryMax'), p.get('TasksMax')))
        pid = p.get('MainPID', '')
        if pid.isdigit() and int(pid):
            for line in read_text(PROC / pid / 'limits').splitlines():
                if line.startswith('Max open files'):
                    info('运行中 PID {}: {}'.format(pid, line))
                    fields = line.split()
                    if len(fields) >= 5 and fields[3].isdigit() and int(fields[3]) < service.get('target', 0):
                        warn(service['name'] + ': 进程限制仍偏低；检查重启与应用 --nofile 覆盖项')
    keys = ['TcpOutSegs', 'TcpRetransSegs', 'TcpExtListenOverflows', 'TcpExtListenDrops',
            'UdpRcvbufErrors', 'UdpSndbufErrors', 'TcpExtTCPMemoryPressures',
            'SoftnetDropped', 'SoftnetTimeSqueeze']
    before = counters()
    print('\n累计计数（不代表本次调优效果）：')
    for key in keys:
        if key in before:
            print('  {}={}'.format(key, before[key]))
    if args.sample:
        info('采样 {} 秒，请让业务保持代表性流量'.format(args.sample))
        time.sleep(args.sample)
        after = counters()
        print('采样增量：')
        for key in keys:
            if key in before and key in after:
                print('  {}={:+d}'.format(key, after[key] - before[key]))
        sent = after.get('TcpOutSegs', 0) - before.get('TcpOutSegs', 0)
        retrans = after.get('TcpRetransSegs', 0) - before.get('TcpRetransSegs', 0)
        if sent > 0 and retrans >= 0:
            info('重传段/发送段增量比 {:.3f}%（不是线路丢包率）'.format(100 * retrans / sent))
    if os.geteuid() == 0 and STATE.exists():
        if pointer('pending'):
            warn('存在未完成事务，请运行 restore')
        if pointer('latest'):
            info('最近备份：' + pointer('latest'))


def bounded_int(low, high):
    def parse(value):
        if not re.fullmatch(r'[0-9]{1,10}', value) or not low <= int(value) <= high:
            raise argparse.ArgumentTypeError('范围 {}..{}'.format(low, high))
        return int(value)
    return parse


def arguments(argv):
    parser = argparse.ArgumentParser(
        prog='vps-proxy-tune', description='VPS 代理网络调优 v' + VERSION + '（无参数打开交互菜单）',
        epilog='apply/restore 需要 root。不会重启代理。auto 只做静态资源分档，不持续自动调参。\n'
               'keep 保留此前工具管理的键；--no-services/--app-configs off 不删除已有修改。\n'
               'restore 逐次撤销；回滚遇到人工修改会停止，--force 明确覆盖。\n'
               '不修改防火墙、路由、DNS、转发、MTU、网卡 offload/RPS/XPS、TCP TIME_WAIT 和重试次数。\n'
               '不会自动开启 MPTCP/Brutal/ECN/TFO，不安装新内核，不把 BBR 宣称为 UDP 加速。',
        formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument('action', nargs='?', default='plan', choices=['plan', 'apply', 'status', 'restore'])
    parser.add_argument('--version', action='version', version=VERSION)
    parser.add_argument('--profile', choices=['balanced', 'throughput'], default='balanced')
    parser.add_argument('--buffer-mib', type=bounded_int(1, 256), help='显式缓冲上限目标；绕过自动内存保护')
    parser.add_argument('--buffer-policy', choices=['grow', 'set'], default='grow', help='grow 保留更高上限；set 允许降低 max')
    parser.add_argument('--bandwidth-mbps', type=bounded_int(1, 1000000), help='目标路径带宽，须与 --rtt-ms 同用')
    parser.add_argument('--rtt-ms', type=bounded_int(1, 10000), help='目标路径 RTT；带宽和 RTT 只作为 BDP 提示')
    parser.add_argument('--cc', choices=['auto', 'bbr', 'cubic', 'keep'], default='auto')
    parser.add_argument('--qdisc', choices=['fq', 'keep'], default='fq', help='仅管理默认值，不替换在线网卡队列')
    parser.add_argument('--tfo', choices=['keep', 'on', 'off'], default='keep')
    parser.add_argument('--mtu-probing', choices=['auto', 'keep'], default='auto', help='auto 至少设为 1，保留已有 2')
    parser.add_argument('--service', action='append', default=[], metavar='UNIT')
    parser.add_argument('--no-services', action='store_true')
    parser.add_argument('--nofile', type=bounded_int(1024, 1048576), help='省略时按有效内存选 16384..262144')
    parser.add_argument('--app-configs', choices=['auto', 'off'], default='auto', help='auto 识别运行中原生 ssserver/realm；不改协议与凭据')
    parser.add_argument('--ss-config', action='append', default=[], metavar='/PATH/CONFIG.JSON', help='明确指定 Shadowsocks-rust JSON/JSONC，仅开启 no_delay')
    parser.add_argument('--allow-container', action='store_true')
    parser.add_argument('--best-effort', action='store_true', help='允许跳过不可写 sysctl；默认写入失败整批回滚')
    parser.add_argument('--sample', type=bounded_int(1, 30), default=0, help='status 计数器采样秒数')
    parser.add_argument('--force', action='store_true', help='仅 restore：覆盖应用后的外部修改')
    args = parser.parse_args(argv)
    if args.no_services and args.service:
        parser.error('--no-services 与 --service 不能同用')
    if (args.bandwidth_mbps is None) != (args.rtt_ms is None):
        parser.error('--bandwidth-mbps 和 --rtt-ms 必须一起指定')
    if args.buffer_mib is not None and args.bandwidth_mbps is not None:
        parser.error('--buffer-mib 与带宽/RTT 提示请选择一种')
    if args.force and args.action != 'restore':
        parser.error('--force 只用于 restore')
    if args.sample and args.action != 'status':
        parser.error('--sample 只用于 status')
    return args


def main(argv):
    if sys.version_info < (3, 6):
        raise TuneError('需要 Python 3.6+')
    args = arguments(argv)
    os.umask(0o077)
    os.environ['LC_ALL'] = 'C'
    os.environ['PATH'] = '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin'
    for command in ('sysctl', 'ip'):
        if not shutil.which(command):
            raise TuneError('缺少 {}；安装依赖：apt-get install python3 procps iproute2'.format(command))
    if args.action in ('apply', 'restore') and os.geteuid() != 0:
        raise TuneError('apply/restore 需要 root，请使用 sudo')
    systemd = (RUN / 'systemd/system').is_dir() and bool(shutil.which('systemctl'))
    if args.action == 'restore':
        with locked():
            name = pointer('pending') or pointer('latest')
            if not name:
                raise TuneError('没有本工具可恢复的备份')
            restore_data(name, systemd, args.force)
        return
    distro = read_text(ETC / 'os-release')
    match = re.search(r'^ID=[\"\']?([a-zA-Z0-9_-]+)', distro, re.M)
    if not match or match.group(1) not in ('debian', 'ubuntu'):
        raise TuneError('本脚本面向 Debian / Ubuntu')
    container = is_container()
    if container:
        warn('检测到容器；只能修改获准访问的内核参数，部分参数可能由宿主管理')
        if args.action == 'apply' and not args.allow_container:
            raise TuneError('容器应用需 --allow-container；按需加 --best-effort')
    hw = resources()
    size = sizing(hw, args)
    hardware_report(hw, size)
    services = detect_services(args, systemd)
    service_plans(services, size, args)
    if args.action == 'status':
        status(args, services)
        return
    apps = application_plans(args, services)
    if args.action == 'plan':
        show_plan(args, size, services, apps)
    else:
        def interrupted(signum, frame):
            raise KeyboardInterrupt()
        signal.signal(signal.SIGTERM, interrupted)
        apply(args, hw, size, services, apps, systemd, container)


# Interactive front end. Read /dev/tty because stdin contains the Python heredoc.
class Console:
    def __enter__(self):
        try:
            self.input = open('/dev/tty', 'r', encoding='utf-8')
            self.output = open('/dev/tty', 'w', encoding='utf-8', buffering=1)
        except OSError:
            raise TuneError('没有交互终端，请在 SSH 终端粘贴运行；已有命令行参数仍可使用')
        return self

    def __exit__(self, *unused):
        self.input.close()
        self.output.close()

    def say(self, text=''):
        print(text, file=self.output, flush=True)

    def ask(self, prompt, default=''):
        self.output.write('{}{}：'.format(prompt, ' [{}]'.format(default) if default != '' else ''))
        self.output.flush()
        line = self.input.readline()
        if not line:
            raise EOFError()
        return line.strip() or str(default)

    def number(self, prompt, low, high, default):
        while True:
            text = self.ask(prompt, default)
            if re.fullmatch(r'[0-9]{1,10}', text) and low <= int(text) <= high:
                return int(text)
            self.say('请输入 {} 到 {} 之间的整数。'.format(low, high))

    def choose(self, title, labels, default=1):
        self.say('\n' + title)
        for i, label in enumerate(labels, 1):
            self.say('  {}. {}'.format(i, label))
        return self.number('请选择', 1, len(labels), default)


def menu_defaults():
    return dict(profile='balanced', buffer_mib=None, buffer_policy='grow',
                bandwidth_mbps=None, rtt_ms=None, cc='auto', qdisc='fq', tfo='keep',
                mtu_probing='auto', service=[], no_services=False, nofile=None,
                app_configs='auto', ss_config=[], allow_container=False, best_effort=False)


def menu_arguments(options, action):
    result = [action]
    for key, value in options.items():
        flag = '--' + key.replace('_', '-')
        if value is None or value is False:
            continue
        if value is True:
            result.append(flag)
        elif isinstance(value, list):
            for item in value:
                result.extend([flag, item])
        else:
            result.extend([flag, str(value)])
    return result


def menu_summary(ui, options):
    buf = '自动按内存计算'
    if options['buffer_mib'] is not None:
        buf = '{} MiB'.format(options['buffer_mib'])
    elif options['bandwidth_mbps'] is not None:
        buf = '{} Mbps / RTT {} ms'.format(options['bandwidth_mbps'], options['rtt_ms'])
    services = '不新增服务限制' if options['no_services'] else ', '.join(options['service']) or '自动识别'
    ui.say('当前：{} | 缓冲 {} | 拥塞 {}'.format(options['profile'], buf, options['cc']))
    ui.say('服务：{} | NOFILE {} | 应用配置 {}'.format(services, options['nofile'] or '自动', options['app_configs']))


def menu_units():
    units = set()
    for verb in ('list-units', 'list-unit-files'):
        p = run(['systemctl', verb, '--type=service', '--all', '--no-legend', '--no-pager', '--plain'])
        if p is not None and p.returncode == 0:
            for line in p.stdout.splitlines():
                words = line.split()
                if words and UNIT_RE.fullmatch(words[0]) and not words[0].endswith('@.service'):
                    units.add(words[0])
    return sorted(units)


def menu_select_units(ui, selected=None):
    all_units = menu_units()
    keyword = ui.ask('服务名筛选关键词（留空显示常见代理，* 显示全部）')
    units = [u for u in all_units if keyword == '*' or
             (keyword.lower() in u.lower() if keyword else AUTO_UNIT_RE.fullmatch(u) or u in (selected or []))]
    if not units:
        ui.say('没有匹配单元；可在服务设置中选择“输入服务名”。')
        return None
    for i, unit in enumerate(units, 1):
        ui.say('  {}. {}'.format(i, unit))
    while True:
        text = ui.ask('输入编号，多选以空格或逗号分隔；0 返回', '0')
        if text == '0':
            return None
        tokens = text.replace('，', ',').replace(',', ' ').split()
        if tokens and all(re.fullmatch(r'[0-9]{1,6}', t) and 1 <= int(t) <= len(units) for t in tokens):
            return list(dict.fromkeys(units[int(t) - 1] for t in tokens))
        ui.say('编号无效，请重新选择。')


def menu_settings(ui, options):
    choices = ['性能档：balanced / throughput', '缓冲区：自动 / 带宽与 RTT / 自定 MiB',
               '缓冲上限策略：保留更高值 / 允许降低', '拥塞算法：自动 / BBR / CUBIC / 保持',
               '默认队列：fq / 保持', 'TCP Fast Open：保持 / 开启 / 关闭',
               'MTU 黑洞探测：自动 / 保持', '服务：自动识别 / 按编号选择 / 输入名称 / 不修改',
               '文件描述符上限：自动 / 自定义', '应用配置：自动识别 / 保持 / 指定 Shadowsocks-rust 文件',
               '容器：拒绝应用 / 允许尝试', '写入失败：整批回滚 / 跳过失败的 sysctl',
               '重置本次菜单参数为默认值', '返回主菜单']
    while True:
        menu_summary(ui, options)
        choice = ui.choose('参数设置（按需修改，选择 14 返回）', choices, 14)
        if choice == 14:
            return
        if choice == 13:
            options.clear()
            options.update(menu_defaults())
            continue
        if choice in (1, 3, 4, 5, 6, 7, 11, 12):
            fields = {
                1: ('profile', ['自动平衡（推荐）', '吞吐优先'], ['balanced', 'throughput']),
                3: ('buffer_policy', ['保留已有更高上限（推荐）', '允许降低 max，保留 min/default'], ['grow', 'set']),
                4: ('cc', ['自动探测；保留自定义算法', '指定 BBR', '指定 CUBIC', '保持原设置'], ['auto', 'bbr', 'cubic', 'keep']),
                5: ('qdisc', ['fq 默认值（不会替换当前网卡队列）', '保持原设置'], ['fq', 'keep']),
                6: ('tfo', ['保持原设置（推荐）', '开启客户端与服务端 TFO 位', '关闭 TFO'], ['keep', 'on', 'off']),
                7: ('mtu_probing', ['自动开启黑洞探测，保留已有增强模式', '保持原设置'], ['auto', 'keep']),
                11: ('allow_container', ['容器内拒绝应用（默认）', '允许在容器内尝试'], [False, True]),
                12: ('best_effort', ['写入失败则整批回滚（推荐）', '跳过不可写的 sysctl，保存成功项'], [False, True])}
            key, labels, values = fields[choice]
            index = ui.choose(choices[choice - 1], labels, values.index(options[key]) + 1)
            options[key] = values[index - 1]
        elif choice == 2:
            mode = ui.choose('缓冲区目标', ['自动按可用内存', '用带宽和 RTT 估算', '指定 MiB（覆盖自动内存保护）'])
            change = dict(buffer_mib=None, bandwidth_mbps=None, rtt_ms=None)
            if mode == 2:
                change['bandwidth_mbps'] = ui.number('目标路径带宽 Mbps', 1, 1000000, options['bandwidth_mbps'] or 1000)
                change['rtt_ms'] = ui.number('目标路径 RTT 毫秒', 1, 10000, options['rtt_ms'] or 100)
            elif mode == 3:
                change['buffer_mib'] = ui.number('每个 socket 的上限目标 MiB', 1, 256, options['buffer_mib'] or 32)
            options.update(change)
        elif choice == 8:
            mode = ui.choose('服务选择', ['自动识别常见代理', '按编号选择已安装单元', '输入具体服务名', '不新增服务限制'])
            units = []
            if mode == 2:
                units = menu_select_units(ui, options['service'])
                if units is None:
                    continue
            elif mode == 3:
                while True:
                    units = ui.ask('具体 .service 名称，多个以空格分隔；0 返回').replace('，', ' ').replace(',', ' ').split()
                    if units == ['0']:
                        break
                    if units and all(UNIT_RE.fullmatch(u) and not u.endswith('@.service') for u in units):
                        break
                    ui.say('请输入如 realm.service 或 ssserver@main.service 的完整名称。')
                if units == ['0']:
                    continue
            options.update(service=list(dict.fromkeys(units)), no_services=(mode == 4))
        elif choice == 9:
            mode = ui.choose('NOFILE', ['按有效内存自动选择（保留更高限制）', '自定义上限目标'])
            target = ui.number('文件描述符数量', 1024, 1048576, options['nofile'] or 65536) if mode == 2 else None
            options['nofile'] = target
        elif choice == 10:
            mode = ui.choose('应用配置', ['自动识别运行中的原生 ssserver / realm', '本次保留应用配置', '手动指定 Shadowsocks-rust JSON/JSONC'])
            paths = []
            if mode == 3:
                while True:
                    path = ui.ask('输入配置绝对路径（每次一个；空行结束；0 取消）')
                    if path == '0':
                        paths = None
                        break
                    if not path:
                        if paths:
                            break
                        ui.say('至少输入一个路径，或输入 0 取消。')
                    elif Path(path).is_absolute():
                        paths.append(path)
                    else:
                        ui.say('需要完整绝对路径，例如 /etc/shadowsocks-rust/config.json。')
                if paths is None:
                    continue
            options.update(app_configs='auto' if mode == 1 else 'off', ss_config=list(dict.fromkeys(paths)))


def menu_run(argv):
    previous = {sig: signal.getsignal(sig) for sig in (signal.SIGINT, signal.SIGTERM)}
    try:
        main(argv)
        return True
    except KeyboardInterrupt:
        warn('操作中断；若提示存在未完成事务，可从菜单重新执行回滚')
    except (TuneError, OSError, ValueError) as e:
        warn(str(e))
    except SystemExit as e:
        if not e.code:
            return True
    finally:
        for sig, handler in previous.items():
            signal.signal(sig, handler)
    return False


def menu_restart(ui, options):
    units = menu_select_units(ui, options['service'])
    if not units:
        return
    ui.say('即将重启：' + ', '.join(units))
    ui.say('这些服务的现有代理连接会断开，SSH 若经过它们也可能中断。')
    if ui.choose('是否重启选定服务？', ['取消', '重启以上服务'], 1) != 2:
        return
    for unit in units:
        p = run(['systemctl', 'restart', unit], timeout=60)
        active = run(['systemctl', 'is-active', '--quiet', unit])
        if p is not None and p.returncode == 0 and active is not None and active.returncode == 0:
            ui.say(unit + '：已重启，当前 active。')
        else:
            ui.say(unit + '：未能确认重启成功；请检查该服务日志。')


def interactive_menu():
    if os.geteuid() != 0:
        raise TuneError('交互安装/管理需要 root，请先执行 sudo -i')
    options = menu_defaults()
    with Console() as ui:
        ui.say('\nVPS 网络调优 v' + VERSION + ' · 交互版')
        ui.say('参数仅在本次菜单会话内保留；已应用的系统配置及备份会持久保存。')
        try:
            while True:
                menu_summary(ui, options)
                action = ui.choose('主菜单', ['应用当前参数（初始为自适应默认值）', '只读预览当前参数',
                    '修改参数', '查看状态与网络计数器', '回滚最近一次应用',
                    '重启选定服务，使 NOFILE / 应用配置生效', '退出'], 2)
                if action == 7:
                    return
                if action == 3:
                    menu_settings(ui, options)
                    continue
                if action == 1:
                    if is_container() and not options['allow_container']:
                        if ui.choose('检测到容器，是否允许尝试应用？', ['返回，不应用', '允许本次菜单会话在容器内尝试'], 1) == 1:
                            continue
                        options['allow_container'] = True
                    argv = menu_arguments(options, 'apply')
                    if menu_run(argv):
                        ui.say('调优已完成。服务尚未重启；需要时从主菜单选择 6。')
                elif action == 2:
                    menu_run(menu_arguments(options, 'plan'))
                elif action == 4:
                    seconds = ui.number('计数器采样秒数（0 表示只查看当前状态）', 0, 30, 5)
                    argv = menu_arguments(options, 'status')
                    if seconds:
                        argv += ['--sample', str(seconds)]
                    menu_run(argv)
                elif action == 5:
                    mode = ui.choose('回滚选项', ['正常回滚：遇到外部修改就停止',
                        '强制回滚：覆盖已记录项后续的人工修改', '返回主菜单'], 1)
                    if mode != 3:
                        menu_run(['restore'] + (['--force'] if mode == 2 else []))
                elif action == 6:
                    menu_restart(ui, options)
                ui.ask('按回车返回主菜单')
        except (KeyboardInterrupt, EOFError):
            ui.say('\n已退出菜单。')

if __name__ == '__main__':
    try:
        if len(sys.argv) == 1 or sys.argv[1:] == ['menu']:
            interactive_menu()
        else:
            main(sys.argv[1:])
    except KeyboardInterrupt:
        warn('操作已中断；如有 pending，请运行 restore')
        sys.exit(130)
    except (TuneError, OSError, ValueError) as e:
        warn(str(e))
        sys.exit(1)
VPS_PYTHON
VPS_PROGRAM
bash -n "$VPS_SCRIPT_TMP"
python3 - "$VPS_SCRIPT_TMP" <<'VPS_VALIDATE'
import sys
from pathlib import Path
text = Path(sys.argv[1]).read_text(encoding='utf-8')
body = text.split("<<'VPS_PYTHON'\n", 1)[1].rsplit('\nVPS_PYTHON', 1)[0]
compile(body, 'vps-proxy-tune', 'exec')
VPS_VALIDATE
if [[ -e /usr/local/sbin/vps-proxy-tune ]]; then
    [[ -f /usr/local/sbin/vps-proxy-tune ]] || { echo '已有安装路径不是常规文件。' >&2; exit 1; }
    mkdir -p /var/lib/vps-proxy-tune/install-backups
    chmod 0700 /var/lib/vps-proxy-tune/install-backups
    VPS_OLD_BACKUP=$(mktemp /var/lib/vps-proxy-tune/install-backups/program.XXXXXXXX)
    cp -a -- /usr/local/sbin/vps-proxy-tune "$VPS_OLD_BACKUP"
    echo "原程序已备份：$VPS_OLD_BACKUP"
fi
chmod 0755 "$VPS_SCRIPT_TMP"
mv -f -- "$VPS_SCRIPT_TMP" /usr/local/sbin/vps-proxy-tune
echo '安装完成；以后输入 vps-proxy-tune 即可打开菜单。'
bash /usr/local/sbin/vps-proxy-tune menu
VPS_INSTALL