NeoTachyon/tachyon/files/usr/lib/service/event_controller.uc

#!/usr/bin/env ucode
//
// Event controller: the single place that observes the system and turns
// observations into facts on the event bus. It never repairs anything —
// remediation lives in service/watchdog.uc, which subscribes to these facts.
//
// Two kinds of source feed the bus:
//
//   push  — the kernel or procd tells us something happened (ubus service
//           events, syslog lines, the honeypot FIFO). Zero cost when idle.
//   probe — nothing pushes "DNS stopped resolving", so a probe tier asks.
//           Probing lives here, in one loop, instead of being duplicated
//           inside every healer: one tick collects each observation exactly
//           once and publishes it, where the old code called is_dns_working()
//           from two healers and get_sing_box_pid() from eight.
//
// Probe tiers and their intervals are unchanged from the timer-based
// watchdog: fast 15s, normal adaptive 120/300s, slow 300s.

let fs = require("fs");
let uci_core = require("core.uci");
let common = require("core.common");
let helpers = require("core.helpers");

const CONFIG_NAME = getenv("TACHYON_CONFIG_NAME") || "tachyon";

// Minimum spacing between two increments of the DNS stall streak. Matches the
// floor of the old adaptive normal tier (120s), which is the rate at which
// ai_heal_dns() used to step that counter before probing moved to the fast tier.
const DNS_STREAK_INTERVAL = 120;

// Consecutive WAN probe misses required before WAN_DOWN is published. The old
// check fired on a single observation: one transient gap in `ip route` (a
// DHCP renew re-applying the route, a reload_firewall window) made the healer
// bounce the whole interface, which then really dropped the address and the
// route — a self-inflicted WAN flap every ~5 minutes (issue #31).
const WAN_FAIL_THRESHOLD = 3;

let as_string = common.as_string;
let shell_quote = common.shell_quote;
let command_from_args = common.command_from_args;
let command_success_from_args = common.command_success_from_args;
let is_process_name_running = helpers.is_process_name_running;

// ─── Event type vocabulary ────────────────────────────────────────────────────
// Every fact the controller can publish. Subscribers reference these constants
// rather than bare strings so a typo is a load-time error, not silent silence.
const EV = {
    SINGBOX_STOPPED:        "singbox.stopped",
    FIREWALL_RELOADED:      "firewall.reloaded",
    OOM_DETECTED:           "oom.detected",
    OOM_RECOVERABLE:        "oom.recoverable",
    SMARTDETECT_CANDIDATE:  "smartdetect.candidate",
    URLTEST_SWITCHED:       "urltest.switched",
    HONEYPOT_HIT:           "honeypot.hit",
    PROXY_DOWN:             "proxy.down",
    PROXY_UP:               "proxy.up",
    DNS_DOWN:               "dns.down",
    DNS_UP:                 "dns.up",
    MEMORY_LOW:             "memory.low",
    NFT_MISSING:            "nft.missing",
    QOS_MISSING:            "qos.missing",
    TPROXY_DOWN:            "tproxy.down",
    RPCD_FD_LEAK:           "rpcd.fd_leak",
    WAN_DOWN:               "wan.down",
    WAN_UP:                 "wan.up",
    SUBNETS_EMPTY:          "subnets.empty",
    SUBSCRIPTION_UNREACHABLE: "subscription.unreachable",
    CONFIG_CORRUPT:         "config.corrupt",
    SECTIONS_EMPTY:         "sections.empty",
    ANOMALY_RECONNECTS:     "anomaly.reconnects",
    PAUSE_EXPIRED:          "pause.expired",
    NFQUEUE_DOWN:           "nfqueue.down",
    NFQUEUE_UP:             "nfqueue.up",
    TICK:                   "tick"
};

// ─── Shell helpers ────────────────────────────────────────────────────────────
function command_capture(command) {
    let pipe = fs.popen(command, "r");
    if (!pipe)
        return { status: 1, output: "" };
    let data = pipe.read("all");
    let status = pipe.close();
    if (status > 255) status = int(status / 256);
    return { status, output: data == null ? "" : as_string(data) };
}

function command_output_from_args(args) {
    let result = command_capture(command_from_args(args) + " 2>/dev/null");
    return result.status == 0 ? result.output : "";
}

let current_tick_ctx = null;

function settings() {
    if (current_tick_ctx && current_tick_ctx.settings)
        return current_tick_ctx.settings;
    return common.object_or_empty(uci_core.get_all(CONFIG_NAME, "settings"));
}

function setting(key, default_val) {
    let cfg = settings();
    let val = cfg[key];
    return val != null ? as_string(val) : as_string(default_val);
}

function enabled(key, default_val) {
    return setting(key, default_val) == "1";
}

function process_running(pid, expected_name) {
    if (match(as_string(pid), /^[0-9]+$/) == null)
        return false;
    if (expected_name != null && expected_name != "")
        return is_process_name_running(pid, expected_name);
    return fs.stat("/proc/" + pid) != null;
}

// ─── Shared observations ──────────────────────────────────────────────────────
// Resolved once per tick and handed to every probe that needs them. This is
// what removes the repeated get_sing_box_pid() / is_dns_working() calls that
// the per-healer detection prologues used to make.

function get_sing_box_pid() {
    for (let path in [ "/var/run/sing-box.pid", "/var/run/sing-box/sing-box.pid" ]) {
        let pid = trim(fs.readfile(path) || "");
        if (pid != "" && process_running(pid, "sing-box")) return pid;
    }

    let ubus_res = command_capture("ubus call service list '{\"name\":\"sing-box\"}' 2>/dev/null");
    if (ubus_res.status == 0 && ubus_res.output != "") {
        let matched = match(ubus_res.output, /"pid":\s*([0-9]+)/);
        if (matched && matched[1] != "") {
            let pid = matched[1];
            if (process_running(pid, "sing-box")) return pid;
        }
    }

    let pidof_res = command_capture("pidof sing-box 2>/dev/null");
    if (pidof_res.status == 0 && pidof_res.output != "") {
        let fields = split(trim(pidof_res.output), /[ \t]+/);
        if (length(fields) > 0 && fields[0] != "") {
            let pid = fields[0];
            if (process_running(pid, "sing-box")) return pid;
        }
    }

    return "";
}

// Scans /proc for a sing-box executable. Slower than get_sing_box_pid(), used
// only to confirm a suspected stop before publishing the fact.
function sing_box_in_proc() {
    let proc = fs.opendir("/proc");
    if (!proc) return "";
    let entry;
    let pid = "";
    while ((entry = proc.read()) != null) {
        if (match(entry, /^[0-9]+$/)) {
            let exe = fs.readlink("/proc/" + entry + "/exe") || "";
            let slash = rindex(exe, "/");
            if ((slash >= 0 ? substr(exe, slash + 1) : exe) == "sing-box") {
                pid = entry;
                break;
            }
        }
    }
    proc.close();
    return pid;
}

function is_dns_working() {
    return command_success_from_args([ "sh", "-c",
        "nslookup -timeout=2 yandex.ru 127.0.0.1 >/dev/null 2>&1 || " +
        "nslookup -timeout=2 connectivitycheck.gstatic.com 127.0.0.1 >/dev/null 2>&1 || " +
        "nslookup -timeout=2 google.com 127.0.0.1 >/dev/null 2>&1" ]);
}

function check_tachyon_cli_running() {
    // pgrep -f matches against the full command line and is cheaper than
    // scanning all of /proc ourselves: it returns only tachyon PIDs, then we
    // read /proc/<pid>/cmdline only for those (typically 0-1 processes),
    // not for every process on the system.
    let pids_res = command_capture("pgrep -f '/usr/bin/tachyon' 2>/dev/null");
    if (pids_res.status != 0 || trim(pids_res.output) == "") return false;
    for (let pid in split(trim(pids_res.output), "\n")) {
        pid = trim(pid);
        if (pid == "" || match(pid, /^[0-9]+$/) == null) continue;
        let cmdline = fs.readfile("/proc/" + pid + "/cmdline") || "";
        if (index(cmdline, "start") >= 0 || index(cmdline, "restart") >= 0 ||
            index(cmdline, "reload") >= 0 || index(cmdline, "stop") >= 0)
            return true;
    }
    return false;
}

function is_list_update_running() {
    let pid = trim(fs.readfile("/var/run/tachyon_list_update.pid") || "");
    return process_running(pid, "ucode");
}

function is_reload_in_progress() {
    return fs.stat("/var/run/tachyon.reload.lock") != null || check_tachyon_cli_running();
}

// Reads a port out of the generated sing-box config. Both callers fall back to
// a compiled-in default when the file cannot be parsed, and that fallback is
// worth a log line: a watchdog probing 4534 while sing-box listens elsewhere
// reports the proxy as down forever and heals it in a loop. The failure used to
// be swallowed by an empty catch, so the symptom was a permanently unhealthy
// proxy with nothing in the log to connect it to the config.
//
// Logged once per process — these are called on every tick, and a config that
// fails to parse fails to parse every time.
let sb_config_parse_reported = false;

function parse_singbox_config() {
    let data = fs.readfile("/etc/sing-box/config.json");
    if (data == null)
        return null;

    try {
        return json(data);
    }
    catch (e) {
        if (!sb_config_parse_reported) {
            sb_config_parse_reported = true;
            command_success_from_args([ "logger", "-t", "tachyon",
                "[err] Watchdog: /etc/sing-box/config.json is unparseable (" + as_string(e) +
                "); falling back to default inbound ports, proxy checks may probe the wrong port" ]);
        }
        return null;
    }
}

// Reads the http/mixed inbound port from the generated sing-box config.
// Cached because the config only changes across a reload.
let cached_proxy_port = null;
let cached_proxy_host = null;
let cached_tproxy_port = null;
function proxy_port() {
    if (cached_proxy_port !== null)
        return as_string(cached_proxy_port);

    // Empty when the generated config has no http/mixed inbound: probing the
    // default port would measure a dead listener and report the proxy broken.
    let port = "";
    let host = "127.0.0.1";
    let sb_cfg = parse_singbox_config();
    if (sb_cfg && sb_cfg.inbounds) {
        // 1. Prefer service-mixed-in
        for (let inb in sb_cfg.inbounds) {
            if (inb.tag == "service-mixed-in" && inb.listen_port != null) {
                port = as_string(inb.listen_port);
                let h = as_string(inb.listen || "127.0.0.1");
                if (h != "0.0.0.0" && h != "::" && h != "") host = h;
                break;
            }
        }
        // 2. Localhost listeners
        if (port == "") {
            for (let inb in sb_cfg.inbounds) {
                if ((inb.type == "http" || inb.type == "mixed") && inb.listen_port != null) {
                    let listen = as_string(inb.listen || "");
                    if (listen == "127.0.0.1" || listen == "0.0.0.0" || listen == "::" || listen == "") {
                        port = as_string(inb.listen_port);
                        host = "127.0.0.1";
                        break;
                    }
                }
            }
        }
        // 3. Any http or mixed inbound
        if (port == "") {
            for (let inb in sb_cfg.inbounds) {
                if ((inb.type == "http" || inb.type == "mixed") && inb.listen_port != null) {
                    port = as_string(inb.listen_port);
                    let h = as_string(inb.listen || "127.0.0.1");
                    if (h != "0.0.0.0" && h != "::" && h != "") host = h;
                    break;
                }
            }
        }
    }
    cached_proxy_port = port;
    cached_proxy_host = host;
    return port;
}

function proxy_host() {
    if (cached_proxy_host !== null)
        return cached_proxy_host;
    proxy_port();
    return cached_proxy_host || "127.0.0.1";
}

function forget_proxy_port() {
    cached_proxy_port = null;
    cached_proxy_host = null;
    cached_tproxy_port = null;
    sb_config_parse_reported = false;
}

function tproxy_port() {
    if (cached_tproxy_port !== null)
        return cached_tproxy_port;
    let port = 4530;
    let sb_cfg = parse_singbox_config();
    if (sb_cfg && sb_cfg.inbounds) {
        for (let inb in sb_cfg.inbounds) {
            if (inb.type == "tproxy") {
                port = int(inb.listen_port || port);
                break;
            }
        }
    }
    cached_tproxy_port = port;
    return port;
}

// ─── Log line classification ──────────────────────────────────────────────────
// Pure functions: a syslog line in, a fact (or null) out. No I/O, so the
// classifier is directly testable — see tests/event_controller.sh.

// Extract a candidate domain from a sing-box log line. Returns null when the
// line carries no usable hostname. Kept separate from log handling so the
// pattern can be exercised directly by tests.
function smart_detect_extract_domain(line) {
    if (line == null) return null;
    let text = as_string(line);
    let m = match(text, /"([a-zA-Z0-9][a-zA-Z0-9.-]{1,60}\.[a-zA-Z]{2,})(:[0-9]+)?"/);
    if (!m) m = match(text, /target[= ]([a-zA-Z0-9][a-zA-Z0-9.-]{1,60}\.[a-zA-Z]{2,})/);
    if (!m || !m[1]) return null;
    let domain = m[1];
    if (length(domain) < 5) return null;
    if (index(domain, "*") >= 0 || index(domain, "?") >= 0) return null;
    if (index(domain, "..") >= 0) return null;
    if (index(domain, "-") == 0 || substr(domain, length(domain) - 1) == "-") return null;
    return domain;
}

// Classifies one syslog line into zero or more facts, mirroring the branch
// structure of the old handle_log_line() exactly:
//
//   - the keyword pre-filter rejects the overwhelming majority of lines before
//     any lowercasing or regex work happens;
//   - an OOM line is terminal — the original returned right after it, so a line
//     that is both an OOM and a URLTest line yields only the OOM fact;
//   - smart-detect and URLTest are NOT exclusive: the original fell through
//     from one to the other, so both facts can come from a single line. Hence
//     an array return rather than a single fact.
//
// The netlink/nlbwmon discrimination is preserved: misreading a netlink
// warning as OOM would shrink GOMEMLIMIT for nothing.
function classify_log_line(line) {
    if (!line || line == "") return [];

    // Fast keyword pre-filter: skip 95%+ of irrelevant log lines instantly
    if (index(line, "direct") < 0 && index(line, "DIRECT") < 0 &&
        index(line, "memory") < 0 && index(line, "oom") < 0 && index(line, "OOM") < 0 &&
        index(line, "URLTest") < 0 && index(line, "proxy") < 0) {
        return [];
    }

    let line_lower = lc(line);

    let is_oom = (index(line_lower, "oom-killer") >= 0 ||
                  index(line_lower, "out of memory: kill process") >= 0 ||
                  (index(line_lower, "kernel:") >= 0 && index(line_lower, "out of memory") >= 0) ||
                  (index(line_lower, "sing-box") >= 0 && index(line_lower, "out of memory") >= 0) ||
                  index(line_lower, "fatal error: out of memory") >= 0);
    let is_netlink_warning = (index(line_lower, "netlink") >= 0 || index(line_lower, "nlbwmon") >= 0);

    // Terminal, as in the original: no further classification after an OOM.
    if (is_oom && !is_netlink_warning)
        return [ { type: EV.OOM_DETECTED, payload: {} } ];

    let facts = [];

    if ((index(line_lower, "direct") >= 0 || index(line_lower, "DIRECT") >= 0) &&
        (index(line_lower, "failed") >= 0 || index(line_lower, "timeout") >= 0 ||
         index(line_lower, "reset") >= 0)) {
        let domain = smart_detect_extract_domain(line);
        if (domain != null)
            push(facts, { type: EV.SMARTDETECT_CANDIDATE, payload: { domain: domain } });
    }

    if (index(line, "URLTest") >= 0 || index(line_lower, "selected proxy") >= 0 ||
        index(line_lower, "switch proxy") >= 0)
        push(facts, { type: EV.URLTEST_SWITCHED, payload: {} });

    return facts;
}

// ─── Controller ───────────────────────────────────────────────────────────────
// Owns streak counters and latency history: both describe the *observation*,
// not the repair, so they belong on this side of the boundary. Subscribers
// receive the streak in the payload and compare it against their own
// threshold, exactly as the old healers compared their private counters.

function controller(bus, opts) {
    let options = (type(opts) == "object") ? opts : {};
    let log = (type(options.log) == "function") ? options.log : function(msg, lvl) {};

    let state = {
        proxy_consecutive_fails: 0,
        dns_consecutive_fails: 0,
        dns_fail_streak: 0,
        dns_streak_stamp: 0,
        last_proxy_probe: 0,
        last_dns_probe: 0,
        proxy_latency_history: [],
        dns_latency_history: [],
        syslog_start_time: 0,
        logread_pipe_fd: -1,
        last_oom_time: 0,
        healthy_streak: 0,
        subscription_fail_streak: 0,
        wan_fail_streak: 0
    };

    let self = { EV: EV, state: state };
    current_tick_ctx = null;

    // Keeps the last 20 samples. Bounded so a long-lived watchdog cannot grow
    // its heap through the history array.
    function push_history(history, entry) {
        push(history, entry);
        if (length(history) > 20) {
            return slice(history, length(history) - 20);
        }
        return history;
    }

    // ── Push source: syslog ───────────────────────────────────────────────────
    // Returns the list of facts actually published (after the OOM replay guard,
    // the smart_detect setting and the URLTest throttle), so tests can assert
    // on publication rather than on classification alone.
    self.handle_log_line = function(line) {
        let published = [];

        for (let fact in classify_log_line(line)) {
            if (fact.type == EV.OOM_DETECTED) {
                let now = time();
                // Ignore replay of the historical log buffer that logread -f
                // dumps on start, and collapse an OOM storm into one fact.
                if (state.syslog_start_time > 0 && (now - state.syslog_start_time > 3) &&
                    (now - state.last_oom_time > 60)) {
                    state.last_oom_time = now;
                    bus.emit(EV.OOM_DETECTED, {});
                    push(published, EV.OOM_DETECTED);
                }
                continue;
            }

            if (fact.type == EV.SMARTDETECT_CANDIDATE) {
                if (setting("smart_detect", "0") != "1") continue;
                bus.emit(EV.SMARTDETECT_CANDIDATE, fact.payload);
                push(published, EV.SMARTDETECT_CANDIDATE);
                continue;
            }

            // URLTest switches are chatty; the old check_urltest_switches()
            // throttled itself to one run per 5s, so the fact carries the same.
            if (fact.type == EV.URLTEST_SWITCHED) {
                if (bus.emit_once(EV.URLTEST_SWITCHED, {}, 5) < 0) continue;
                push(published, EV.URLTEST_SWITCHED);
            }
        }

        return published;
    };

    // ── Push source: ubus ─────────────────────────────────────────────────────
    self.handle_ubus_service_stop = function(name, reason) {
        if (as_string(name) != "sing-box") return false;
        bus.emit(EV.SINGBOX_STOPPED, { reason: as_string(reason) });
        return true;
    };

    // handle_ubus_firewall_reload is defined further down, next to the probes it
    // re-runs. Declaration order in this file is load-bearing: ucode captures a
    // closure's upvalues when the closure is created, so a local declared later
    // in the same scope is not captured at all — the name resolves as a global
    // and comes back null. That fails at call time, which neither `ucode -c` nor
    // `ucode -S -c` can see. Every helper here precedes its callers.

    // ── Push source: honeypot FIFO ────────────────────────────────────────────
    self.handle_honeypot_line = function(line) {
        let ip = trim(as_string(line));
        if (ip == "" || match(ip, /^[0-9a-fA-F:.]+$/) == null) return false;
        bus.emit(EV.HONEYPOT_HIT, { ip: ip });
        return true;
    };

    // ── Probe: expired pause ──────────────────────────────────────────────────
    // Declared ahead of probe_singbox, which consults it on every fast tick.
    function probe_pause() {
        let val = trim(fs.readfile("/tmp/tachyon_paused_until") || "");
        if (val == "") return false;
        if (int(val) <= time()) {
            bus.emit(EV.PAUSE_EXPIRED, {});
            return false;
        }
        return true; // still paused
    }
    self.is_paused = probe_pause;

    // ── Shared observation context ───────────────────────────────────────────
    // Resolved once at tick entry and reused across all probes within the tick.
    // Eliminates redundant forks of pgrep /usr/bin/tachyon, /proc traversals,
    // and duplicate pid lookups.
    function create_tick_context() {
        let now = time();
        let cfg = common.object_or_empty(uci_core.get_all(CONFIG_NAME, "settings"));
        let reload = is_reload_in_progress();
        let list_upd = is_list_update_running();
        let sb_pid = get_sing_box_pid();
        let sb_running = (sb_pid != "" && process_running(sb_pid, "sing-box"));
        let port = proxy_port();
        let host = proxy_host();
        let paused = probe_pause();

        return {
            now: now,
            settings: cfg,
            reload_in_progress: reload,
            list_update_running: list_upd,
            singbox_pid: sb_pid,
            singbox_running: sb_running,
            proxy_port: port,
            proxy_host: host,
            is_paused: paused
        };
    }
    self.create_tick_context = create_tick_context;
    self.set_tick_context = function(ctx) { current_tick_ctx = ctx; };
    self.clear_tick_context = function() { current_tick_ctx = null; };

    // ── Probe: sing-box liveness ──────────────────────────────────────────────
    // The expensive /proc scan only runs once the cheap pidfile/ubus/pidof
    // path has already failed, preserving the original fast-path ordering.
    function probe_singbox() {
        let ctx = current_tick_ctx || create_tick_context();
        if (setting("recovery_bypass", "0") == "1") return;
        // An active pause means sing-box is stopped on purpose.
        if (ctx.is_paused) return;
        if (ctx.reload_in_progress) return;
        if (ctx.list_update_running) return;

        let pid = ctx.singbox_pid;
        if (pid != "" && ctx.singbox_running) return;

        // No configured sections means sing-box is legitimately absent.
        let has_sections = false;
        let uci_sections = uci_core.get_all(CONFIG_NAME);
        if (uci_sections) {
            for (let k in keys(uci_sections)) {
                if (uci_sections[k][".type"] == "section") {
                    has_sections = true;
                    break;
                }
            }
        }
        if (!has_sections) return;

        if (sing_box_in_proc() == "")
            bus.emit(EV.SINGBOX_STOPPED, { reason: "process missing from /proc" });
    }

    // ── Probe: proxy reachability ─────────────────────────────────────────────
    // One measurement serves both proxy subscribers (the old
    // ai_heal_proxy_health and ai_heal_proxy_connectivity each ran their own
    // curl). Feature flags are deliberately NOT checked here: the two
    // subscribers are gated by different settings, so the flag belongs to the
    // subscriber. The probe only skips when measuring is pointless or its
    // result would be misleading.
    function probe_proxy() {
        let ctx = current_tick_ctx || create_tick_context();
        if (!bus.has(EV.PROXY_DOWN) && !bus.has(EV.PROXY_UP)) return;
        // A reload tears the proxy down on purpose; a sample taken now would
        // land in the latency history as a fault that never happened.
        if (ctx.reload_in_progress) return;

        let pid = ctx.singbox_pid;
        if (pid == "" || !ctx.singbox_running) return;

        let port = ctx.proxy_port;
        // No http/mixed inbound in the generated config: there is nothing to
        // measure, so the proxy must not be declared broken on a dead port.
        if (port == "") return;
        let host = ctx.proxy_host;

        // Adaptive interval pacing: honor ai_proxy_health_interval (default 30s)
        // when healthy (consecutive fails == 0).
        // On failure, switch to fast retry on every tick (10-15s) until restored.
        let proxy_interval = int(ctx.settings.ai_proxy_health_interval || "30");
        if (proxy_interval < 15) proxy_interval = 15;

        if (state.proxy_consecutive_fails == 0 && state.last_proxy_probe > 0 &&
            (ctx.now - state.last_proxy_probe < proxy_interval))
            return;

        state.last_proxy_probe = ctx.now;
        let check_url = ctx.settings.ai_proxy_health_url || "https://cp.cloudflare.com/generate_204";

        let started = time();
        let ok = command_success_from_args([
            "curl", "-s", "-o", "/dev/null",
            "--connect-timeout", "3", "--max-time", "5",
            "--proxy", "http://" + host + ":" + port,
            check_url
        ]);

        // Fallback confirmation: never declare proxy down based on a single URL.
        // Cloudflare or specific CDN endpoints can encounter rate limits or transient DPI packet loss.
        let fallback_url = null;
        if (!ok) {
            fallback_url = index(check_url, "gstatic.com") >= 0
                ? "https://detectportal.firefox.com/success.txt"
                : "https://connectivitycheck.gstatic.com/generate_204";
            ok = command_success_from_args([
                "curl", "-s", "-o", "/dev/null",
                "--connect-timeout", "3", "--max-time", "5",
                "--proxy", "http://" + host + ":" + port,
                fallback_url
            ]);
        }
        let elapsed = (time() - started) * 1000;

        state.proxy_latency_history = push_history(state.proxy_latency_history,
            { ok: ok, ms: elapsed, ts: time() });

        if (ok) {
            state.proxy_consecutive_fails = 0;
            bus.emit(EV.PROXY_UP, { host: host, port: port, ms: elapsed });
            return;
        }

        state.proxy_consecutive_fails++;

        // Distinguish "proxy is broken" from "the whole uplink is down": only
        // the former is worth restarting sing-box over.
        // Check primary URL, and if that fails, check fallback URL directly too
        // in case the primary URL is blocked directly on the ISP WAN.
        let direct_ok = command_success_from_args([
            "curl", "-s", "-I", "--connect-timeout", "3", "--max-time", "5",
            check_url
        ]);
        if (!direct_ok && fallback_url != null) {
            direct_ok = command_success_from_args([
                "curl", "-s", "-I", "--connect-timeout", "3", "--max-time", "5",
                fallback_url
            ]);
        }

        bus.emit(EV.PROXY_DOWN, {
            host: host,
            port: port,
            streak: state.proxy_consecutive_fails,
            direct_ok: direct_ok,
            ms: elapsed
        });
    }

    // Clearing the streak also clears its pacing stamp, so the first failure
    // after a recovery is counted immediately rather than waiting out the rest
    // of an interval that started before the resolver came back.
    function clear_dns_streak() {
        state.dns_fail_streak = 0;
        state.dns_streak_stamp = 0;
    }

    // ── Probe: DNS resolution ─────────────────────────────────────────────────
    // As with the proxy probe, one resolution attempt serves both DNS
    // subscribers; each carries its own enable flag and its own threshold.
    function probe_dns() {
        let ctx = current_tick_ctx || create_tick_context();
        if (!bus.has(EV.DNS_DOWN) && !bus.has(EV.DNS_UP)) return;
        if (setting("recovery_bypass", "0") == "1") return;

        // A reload restarts the resolver; failures during it are expected and
        // must not accumulate, so the streak resets exactly as before.
        if (ctx.reload_in_progress) {
            clear_dns_streak();
            return;
        }

        let pid = ctx.singbox_pid;
        if (pid == "" || !ctx.singbox_running) {
            clear_dns_streak();
            return;
        }

        // Adaptive interval pacing: honor ai_dns_interval (default 60s)
        // when healthy (consecutive fails == 0).
        // On failure, switch to fast retry on every tick (10-15s) until restored.
        let dns_interval = int(ctx.settings.ai_dns_interval || "60");
        if (dns_interval < 15) dns_interval = 15;

        if (state.dns_consecutive_fails == 0 && state.last_dns_probe > 0 &&
            (ctx.now - state.last_dns_probe < dns_interval))
            return;

        state.last_dns_probe = ctx.now;

        let started = time();
        let ok = is_dns_working();
        let elapsed = (time() - started) * 1000;

        state.dns_latency_history = push_history(state.dns_latency_history,
            { ok: ok, ms: elapsed, ts: time() });

        if (ok) {
            state.dns_consecutive_fails = 0;
            clear_dns_streak();
            bus.emit(EV.DNS_UP, { ms: elapsed });
            return;
        }

        state.dns_consecutive_fails++;

        // Two counters at two rates, because the two DNS healers were measured on
        // two different tiers before the split: dns_consecutive_fails came from
        // the fast tier (15s), dns_fail_streak from the normal-tier audit (120s+).
        // The probe now runs once at the fast rate, so the streak is only stepped
        // once per DNS_STREAK_INTERVAL to keep "3 failures" meaning the same
        // ~6 minutes of trouble it meant before, not 45 seconds.
        let now = time();
        if (now - state.dns_streak_stamp >= DNS_STREAK_INTERVAL) {
            state.dns_streak_stamp = now;
            state.dns_fail_streak++;
            log("Watchdog: DNS check failed (" + as_string(state.dns_fail_streak) + "/3 failures)", "warn");
        }

        bus.emit(EV.DNS_DOWN, {
            streak: state.dns_fail_streak,
            consecutive: state.dns_consecutive_fails,
            ms: elapsed
        });
    }

    // ── Probe: memory ─────────────────────────────────────────────────────────
    function read_mem_available_mb() {
        let mem_info = fs.readfile("/proc/meminfo") || "";
        for (let line in split(mem_info, "\n")) {
            if (index(line, "MemAvailable:") == 0) {
                let fields = split(trim(line), /[ \t]+/);
                if (length(fields) >= 2)
                    return int(fields[1]) / 1024;
                break;
            }
        }
        return -1;
    }
    self.read_mem_available_mb = read_mem_available_mb;

    function probe_memory() {
        let free_mb = read_mem_available_mb();
        if (free_mb >= 0 && free_mb < 15)
            bus.emit(EV.MEMORY_LOW, { free_mb: free_mb });
    }

    // ── Probe: nftables table and QoS marks ───────────────────────────────────
    function probe_nftables() {
        let ctx = current_tick_ctx || create_tick_context();
        if (setting("recovery_bypass", "0") == "1") return;
        if (fs.stat("/var/run/tachyon/native_internet_restored") != null) return;
        if (ctx.list_update_running) return;
        if (ctx.reload_in_progress) return;

        let nft_table = getenv("NFT_TABLE_NAME") || "TachyonTable";
        let routing_mode = ctx.settings.routing_mode || "nftables";
        let out_nft = command_output_from_args(["sh", "-c",
            "nft list table inet " + nft_table + " | grep -E 'chain|tproxy|priority_rules|dscp'; exit 0"]);

        if (routing_mode == "nftables") {
            if (index(out_nft, "tproxy") < 0 || index(out_nft, "priority_rules") < 0) {
                bus.emit(EV.NFT_MISSING, { table: nft_table });
                return;
            }
        }

        let qos_pref = ctx.settings.qos_priority_engine || "";
        if (qos_pref == "0") return;
        if ((index(out_nft, "dscp set 0x2e") < 0 && index(out_nft, "dscp set ef") < 0) ||
            (index(out_nft, "dscp set 0x22") < 0 && index(out_nft, "dscp set af41") < 0))
            bus.emit(EV.QOS_MISSING, { table: nft_table });
    }

    // ── Probe: TPROXY port liveness ───────────────────────────────────────────
    function probe_tproxy() {
        let ctx = current_tick_ctx || create_tick_context();
        if (setting("recovery_bypass", "0") == "1") return;
        if (fs.stat("/var/run/tachyon/native_internet_restored") != null) return;
        if (ctx.reload_in_progress) return;

        let pid = ctx.singbox_pid;
        if (pid == "" || !ctx.singbox_running) return;

        let port = tproxy_port();
        // /proc/net/tcp encodes the local port in hex; 0A is TCP_LISTEN.
        let hex_port = sprintf("%04X", port);
        let listening = false;
        for (let proc_file in ["/proc/net/tcp6", "/proc/net/tcp"]) {
            let tcp_data = fs.readfile(proc_file) || "";
            for (let line in split(tcp_data, "\n")) {
                if (index(line, ":" + hex_port + " ") >= 0 && index(line, " 0A ") >= 0) {
                    listening = true;
                    break;
                }
            }
            if (listening) break;
        }

        if (!listening)
            bus.emit(EV.TPROXY_DOWN, { port: port });
    }

    // ── Probe: rpcd file descriptor leak ──────────────────────────────────────
    // rpcd accumulates FDs from LuCI API calls. Past ~1024 it can no longer
    // fork /usr/bin/tachyon and LuCI reports everything as stopped. 512 is
    // half the limit — early enough to restart without user-visible impact.
    const RPCD_FD_THRESHOLD = 512;

    function probe_rpcd() {
        let rpcd_pid = null;
        let proc_dir = fs.opendir("/proc");
        if (!proc_dir) return;
        let entry;
        while ((entry = proc_dir.read()) != null) {
            if (!match(entry, /^[0-9]+$/)) continue;
            let comm = trim(fs.readfile("/proc/" + entry + "/comm") || "");
            if (comm == "rpcd") { rpcd_pid = entry; break; }
        }
        proc_dir.close();
        if (!rpcd_pid) return;

        let fd_dir = fs.opendir("/proc/" + rpcd_pid + "/fd");
        if (!fd_dir) return;
        let fd_count = 0;
        while ((entry = fd_dir.read()) != null) {
            if (match(entry, /^[0-9]+$/)) fd_count++;
        }
        fd_dir.close();

        if (fd_count > RPCD_FD_THRESHOLD)
            bus.emit(EV.RPCD_FD_LEAK, { fd_count: fd_count, threshold: RPCD_FD_THRESHOLD });
    }

    function get_default_route_dev() {
        let data = fs.readfile("/proc/net/route");
        if (data != null && data != "") {
            let lines = split(data, "\n");
            for (let i = 1; i < length(lines); i++) {
                let line = trim(lines[i]);
                if (line == "") continue;
                let fields = split(line, /[ \t]+/);
                if (length(fields) >= 4 && fields[1] == "00000000") {
                    let flags = int("0x" + fields[3]);
                    if ((flags & 1) != 0) { // RTF_UP
                        return fields[0];
                    }
                }
            }
        }
        let rt = command_capture("ip -4 route show default 2>/dev/null").output;
        if (!rt || rt == "") rt = command_capture("ip route show table all default 2>/dev/null").output;
        if (rt && rt != "") {
            let m = match(rt, /dev\s+([^\s]+)/);
            if (m && m[1]) return m[1];
        }
        return null;
    }

    // ── Probe: WAN address and default route ──────────────────────────────────
    // WAN_DOWN is published only after WAN_FAIL_THRESHOLD consecutive misses,
    // mirroring the streak logic of the other probes. A reload restarts the
    // firewall, which can briefly drop the default route; failures seen while
    // that is in progress are not counted, and a healthy pass resets the
    // streak so one transient gap cannot accumulate into a false trigger.
    function probe_wan() {
        let ctx = current_tick_ctx || create_tick_context();
        if (ctx.reload_in_progress) {
            state.wan_fail_streak = 0;
            return;
        }
        if (setting("recovery_bypass", "0") == "1") return;

        let def_dev = get_default_route_dev();

        // If Tachyon has an explicit output interface, use it as the real WAN.
        // This supports modemmanager/wwan, USB modems, and non-standard uplinks
        // where network.interface.wan may be unused or disconnected.
        let configured_device = trim(setting("output_network_interface", ""));
        let device = configured_device != "" ? configured_device : (def_dev || "");

        let no_address = true;
        let no_gateway = true;
        let iface = device;

        // 1. If explicit output interface is configured, check it directly
        if (configured_device != "") {
            // First check if ubus knows it as a logical interface (e.g. wan2, wwan, modem)
            let ubus_data = command_output_from_args([
                "ubus", "-S", "call", "network.interface." + configured_device, "status"
            ]);
            let ubus_obj = null;
            try {
                ubus_obj = json(ubus_data);
            } catch (e) {}

            if (type(ubus_obj) == "object" && ubus_obj.up == true) {
                iface = ubus_obj.l3_device || ubus_obj.device || iface;
                let addrs = common.array_or_empty(ubus_obj["ipv4-address"]);
                let addrs6 = common.array_or_empty(ubus_obj["ipv6-address"]);
                if (length(addrs) > 0 || length(addrs6) > 0)
                    no_address = false;
                if (length(common.array_or_empty(ubus_obj.route)) > 0 || (ubus_obj.data && ubus_obj.data.gateway))
                    no_gateway = false;
            }

            // Verify L3 device directly if address not confirmed yet
            if (no_address && iface != "") {
                let addr_out = command_capture("ip addr show " + shell_quote(iface) + " 2>/dev/null").output;
                if (index(addr_out, "inet ") >= 0 || index(addr_out, "inet6 ") >= 0)
                    no_address = false;
            }

            if (def_dev != null && (def_dev == device || def_dev == iface))
                no_gateway = false;
        }

        // 2. If no explicit output interface is configured, preserve standard WAN ubus check
        if (configured_device == "") {
            let ubus_data = command_output_from_args([ "ubus", "-S", "call", "network.interface.wan", "status" ]);
            let ubus_obj = null;
            try {
                ubus_obj = json(ubus_data);
            } catch (e) {}

            if (type(ubus_obj) == "object" && ubus_obj.up == true) {
                iface = ubus_obj.l3_device || ubus_obj.device || iface;
                let addrs = common.array_or_empty(ubus_obj["ipv4-address"]);
                let addrs6 = common.array_or_empty(ubus_obj["ipv6-address"]);
                if (length(addrs) > 0 || length(addrs6) > 0)
                    no_address = false;
                if (length(common.array_or_empty(ubus_obj.route)) > 0 || (ubus_obj.data && ubus_obj.data.gateway))
                    no_gateway = false;
            }

            if (iface == "") {
                let proto = uci_core.get("network", "wan", "proto") || "dhcp";
                let wan_dev = uci_core.get("network", "wan", "device") || uci_core.get("network", "wan", "ifname");
                if (wan_dev)
                    iface = (proto == "pppoe") ? "pppoe-wan" : wan_dev;
                else
                    iface = (proto == "pppoe") ? "pppoe-wan" : (def_dev || "eth0");
            }

            if (no_address && iface != "") {
                let addr_out = command_capture("ip addr show " + shell_quote(iface) + " 2>/dev/null").output;
                if (index(addr_out, "inet ") >= 0 || index(addr_out, "inet6 ") >= 0)
                    no_address = false;
            }

            if (def_dev != null)
                no_gateway = false;
        }

        if (no_address && def_dev != null) {
            let addr_out = command_capture("ip addr show " + shell_quote(def_dev) + " 2>/dev/null").output;
            if (index(addr_out, "inet ") >= 0 || index(addr_out, "inet6 ") >= 0) {
                no_address = false;
                iface = def_dev;
            }
        }

        // Check bridge / LAN interfaces for client / AP / bridge-station modes
        if (no_address) {
            let check_devs = def_dev != null ? [ def_dev, "br-lan", "lan" ] : [ "br-lan", "lan" ];
            for (let _, d in check_devs) {
                let addr_out = command_capture("ip -4 addr show " + shell_quote(d) + " 2>/dev/null").output;
                if (index(addr_out, "inet ") >= 0) {
                    no_address = false;
                    if (iface == "" || iface == "eth0") iface = d;
                    break;
                }
            }
        }

        // 3. Final fallback: verify that a default route exists in routing table
        if (no_gateway) {
            let route_out = command_capture("ip -4 route show default 2>/dev/null").output;
            if (index(route_out, "default") >= 0) {
                no_gateway = false;
            } else {
                let route_all = command_capture("ip route show table all default 2>/dev/null").output;
                if (!route_all || route_all == "") route_all = command_capture("ip route 2>/dev/null").output;
                if (index(route_all, "default") >= 0)
                    no_gateway = false;
            }
        }

        // Fast internet connectivity probe before declaring WAN failure
        if (no_address || no_gateway) {
            let direct_ok = command_success_from_args([
                "curl", "-s", "-o", "/dev/null",
                "--connect-timeout", "2", "--max-time", "3",
                "http://connectivitycheck.gstatic.com/generate_204"
            ]);
            if (direct_ok) {
                no_address = false;
                no_gateway = false;
            }
        }

        if (!no_address && !no_gateway) {
            state.wan_fail_streak = 0;
            bus.emit(EV.WAN_UP, { iface: iface });
            return;
        }

        state.wan_fail_streak++;
        log("Watchdog: WAN check failed (" + as_string(state.wan_fail_streak) + "/" +
            as_string(WAN_FAIL_THRESHOLD) + " misses)", "warn");
        if (state.wan_fail_streak < WAN_FAIL_THRESHOLD) return;

        bus.emit(EV.WAN_DOWN, {
            iface: iface,
            no_address: no_address,
            no_gateway: no_gateway,
            streak: state.wan_fail_streak
        });
    }

    // ── Probe: community subnet nft sets ──────────────────────────────────────
    function probe_subnet_sets() {
        let ctx = current_tick_ctx || create_tick_context();
        if (setting("recovery_bypass", "0") == "1") return;
        if (ctx.reload_in_progress) return;
        if (ctx.list_update_running) return;

        let pid = ctx.singbox_pid;
        if (pid == "" || !ctx.singbox_running) return;

        let nft_table = getenv("NFT_TABLE_NAME") || "TachyonTable";
        let all_sections = uci_core.get_all(CONFIG_NAME);
        if (!all_sections) return;

        let empty = [];
        for (let sec_name in keys(all_sections)) {
            let s = all_sections[sec_name];
            if (s[".type"] != "section" || s.enabled != "1") continue;
            if (!s.community_lists) continue;

            // A non-zero exit means the set does not exist for this section
            // (no subnet community configured) — not an emptiness problem.
            let set_name = "tachyon_rule_" + sec_name + "_subnets";
            let result = command_capture(
                command_from_args(["nft", "list", "set", "inet", nft_table, set_name]) + " 2>/dev/null");
            if (result.status != 0 || result.output == "") continue;
            if (index(result.output, "elements") < 0)
                push(empty, set_name);
        }

        if (length(empty) > 0)
            bus.emit(EV.SUBNETS_EMPTY, { sets: empty });
    }

    // ── Probe: subscription URL reachability ──────────────────────────────────
    function probe_subscription() {
        let sub_url = trim(setting("subscription_url", ""));
        if (sub_url == "") return;

        let res = command_capture(
            "curl -s -o /dev/null -w %{http_code} --connect-timeout 10 " + shell_quote(sub_url) + " 2>&1");
        let code = int(res.output);
        if (res.status == 0 && code >= 200 && code < 400) {
            state.subscription_fail_streak = 0;
            return;
        }
        // Two consecutive failures before alerting: a single network hiccup
        // (provider momentarily unreachable) must not trigger a Telegram message.
        state.subscription_fail_streak++;
        if (state.subscription_fail_streak < 2) return;

        bus.emit(EV.SUBSCRIPTION_UNREACHABLE, { url: sub_url, code: code, streak: state.subscription_fail_streak });
    }

    // ── Probe: UCI config integrity ───────────────────────────────────────────
    function probe_config() {
        let data = fs.readfile("/etc/config/tachyon");
        if (data == null || data == "")
            bus.emit(EV.CONFIG_CORRUPT, {});
    }

    // ── Probe: proxy sections with subscriptions but no usable outbounds ──────
    function probe_empty_sections() {
        let ctx = current_tick_ctx || create_tick_context();
        if (!enabled("ai_section_failover_enabled", "1")) return;
        if (ctx.reload_in_progress) return;

        let connections = require("config.connections");
        let sections = common.object_or_empty(uci_core.get_all(CONFIG_NAME));
        let empty = [];

        for (let name in sections) {
            let s = common.object_or_empty(sections[name]);
            let action = common.as_string(s.action || "");
            if (!connections.is_connections_action(action)) continue;
            let sub_urls = common.list_option(s, "subscription_url");
            if (length(sub_urls) == 0) continue;

            let cache = common.object_or_empty(
                common.read_json_file("/var/run/tachyon/section-cache/" + name + ".json"));
            let usable = length(common.array_or_empty(cache.servers)) +
                         length(common.array_or_empty(cache.urls)) +
                         length(common.array_or_empty(cache.selector_urls)) +
                         length(common.array_or_empty(cache.domain)) +
                         length(common.array_or_empty(cache.domain_suffix)) +
                         length(common.array_or_empty(cache.ip_cidr));
            if (usable > 0) continue;

            push(empty, name);
        }

        if (length(empty) > 0)
            bus.emit(EV.SECTIONS_EMPTY, { sections: empty });
    }

    // ── Probe: reconnect rate anomaly ─────────────────────────────────────────
    function probe_anomalies() {
        if (!enabled("ai_anomaly_detection_enabled", "1")) return;

        let threshold = int(setting("ai_anomaly_reconnect_threshold", "10"));
        let reconnects = int(trim(fs.readfile("/tmp/tachyon_reconnect_count") || "0"));

        if (reconnects > threshold)
            bus.emit(EV.ANOMALY_RECONNECTS, { count: reconnects, threshold: threshold });
    }

    // ── Probe: OOM scale recovery window ──────────────────────────────────────
    // Half an hour without an OOM means the shrunk GOMEMLIMIT can start
    // creeping back up; the healer decides by how much.
    // A scale of 1.0 means there is nothing to restore, so the marker file is
    // dropped here rather than published as a fact: removing a stale file is an
    // observation cleanup, not a repair a subscriber should have to schedule.
    function probe_oom_recovery() {
        let now = time();
        if (state.last_oom_time == 0) return;
        if (now - state.last_oom_time < 1800) return;

        let scale_path = "/etc/tachyon/mem_scale";
        let scale_data = fs.readfile(scale_path);
        if (scale_data == null) return;
        let current_scale = double(trim(as_string(scale_data)));
        if (current_scale >= 1.0) {
            // Absent file already satisfies the caller; fs.unlink throws on ENOENT.
            try { fs.unlink(scale_path); } catch (e) {}
            return;
        }
        if (current_scale < 0.1) return;

        bus.emit(EV.OOM_RECOVERABLE, { last_oom: state.last_oom_time, scale: current_scale });
    }

    // ── Probe: nfqueue/zapret runtime health ──────────────────────────────────
    // Checks if nfqws/nfqws2 supervisor processes are alive by counting PID
    // files. A mismatch means the DPI bypass engine crashed or was killed.
    function probe_nfqueue() {
        let ctx = current_tick_ctx || create_tick_context();
        if (!bus.has(EV.NFQUEUE_DOWN) && !bus.has(EV.NFQUEUE_UP)) return;
        if (ctx.reload_in_progress) return;

        let pid_dirs = [
            "/var/run/tachyon/zapret/child-pid",
            "/var/run/tachyon/zapret2/child-pid"
        ];

        let configured_nfqueue = { zapret: 0, zapret2: 0 };
        let sections = uci_core.get_all(CONFIG_NAME);
        if (type(sections) == "object") {
            for (let sname, sec in sections) {
                if (type(sec) == "object" && sec.enabled == "1") {
                    if (sec.action == "zapret")
                        configured_nfqueue.zapret++;
                    else if (sec.action == "zapret2")
                        configured_nfqueue.zapret2++;
                }
            }
        }

        for (let i = 0; i < length(pid_dirs); i++) {
            let child_dir = pid_dirs[i];
            let kind = index(child_dir, "zapret2") >= 0 ? "zapret2" : "zapret";
            let cfg_count = configured_nfqueue[kind];
            let expected = 0;
            let running = 0;
            let dir = fs.opendir(child_dir);
            if (dir) {
                let entry;
                while ((entry = dir.read()) != null) {
                    if (entry == "." || entry == "..") continue;
                    expected++;
                    let pid_data = fs.readfile(child_dir + "/" + entry);
                    if (pid_data == null) continue;
                    let pid = trim(as_string(pid_data));
                    if (pid == "" || match(pid, /^[0-9]+$/) == null) continue;
                    if (fs.stat("/proc/" + pid) != null) {
                        running++;
                    }
                }
                dir.close();
            }

            if (expected == 0 && cfg_count > 0)
                expected = cfg_count;

            if (expected == 0) continue;

            if (running == expected) {
                bus.emit(EV.NFQUEUE_UP, { running: running, expected: expected });
            } else {
                bus.emit(EV.NFQUEUE_DOWN, {
                    running: running,
                    expected: expected,
                    dir: child_dir
                });
            }
        }
    }

    // ── Tier drivers ──────────────────────────────────────────────────────────
    // Each probe is wrapped so one failing observation cannot stop the tier;
    // this mirrors the safe_call() isolation the old per-check loop had.
    function run_probe(fn, name) {
        try {
            fn();
        }
        catch (e) {
            log("Probe " + name + " failed: " + as_string(e), "err");
        }
    }

    // ── Push source: ubus firewall.reload ─────────────────────────────────────
    // Declared here, after run_probe and the two probes it drives, because a
    // closure cannot capture a local declared below it (see the note above
    // handle_honeypot_line). A firewall reload can wipe our table and flush the
    // community sets, so those two probes re-run immediately instead of waiting
    // for the next tier tick — this is what check_firewall_rules() did on this
    // event.
    self.handle_ubus_firewall_reload = function() {
        bus.emit(EV.FIREWALL_RELOADED, {});
        current_tick_ctx = create_tick_context();
        run_probe(probe_nftables, "nftables");
        run_probe(probe_subnet_sets, "subnet_sets");
        return true;
    };

    self.probe_fast = function(ctx) {
        current_tick_ctx = ctx || create_tick_context();
        run_probe(probe_singbox, "singbox");
        run_probe(probe_proxy, "proxy");
        run_probe(probe_dns, "dns");
    };

    self.probe_normal = function(ctx) {
        current_tick_ctx = ctx || create_tick_context();
        run_probe(probe_pause, "pause");
        run_probe(probe_memory, "memory");
        run_probe(probe_rpcd, "rpcd");
        run_probe(probe_nftables, "nftables");
        run_probe(probe_tproxy, "tproxy");
    };

    self.probe_slow = function(ctx) {
        current_tick_ctx = ctx || create_tick_context();
        run_probe(probe_subnet_sets, "subnet_sets");
        run_probe(probe_wan, "wan");
        run_probe(probe_subscription, "subscription");
        run_probe(probe_config, "config");
        run_probe(probe_empty_sections, "empty_sections");
        run_probe(probe_anomalies, "anomalies");
        run_probe(probe_oom_recovery, "oom_recovery");
        run_probe(probe_nfqueue, "nfqueue");
    };

    // Health is graded by how long we go without publishing a fault: after a
    // long clean streak the normal tier can afford to slow down.
    self.adaptive_normal_interval = function() {
        if (!enabled("ai_adaptive_intervals_enabled", "1")) return 120;
        return state.healthy_streak > 25 ? 300 : 120;
    };

    self.note_healthy = function() { state.healthy_streak++; };
    self.note_incident = function() { state.healthy_streak = 0; };
    self.note_oom = function(ts) { state.last_oom_time = ts != null ? int(ts) : time(); };
    self.set_syslog_start = function(ts) { state.syslog_start_time = int(ts); };

    // Streak resets belong to whoever acted on the fact: the old healers zeroed
    // their counter right after restarting, so that a restart-in-flight would
    // not immediately trip the same threshold again on the next tick.
    self.reset_dns_streak = clear_dns_streak;
    self.reset_dns_consecutive = function() {
        state.dns_consecutive_fails = 0;
        state.last_dns_probe = 0;
    };
    self.reset_proxy_consecutive = function() {
        state.proxy_consecutive_fails = 0;
        state.last_proxy_probe = 0;
    };

    // Latency history and streak counters feed ai-status-full and the metrics
    // export, which are still owned by the watchdog.
    self.proxy_latency_history = function() { return state.proxy_latency_history; };
    self.dns_latency_history = function() { return state.dns_latency_history; };
    self.proxy_consecutive_fails = function() { return state.proxy_consecutive_fails; };
    self.dns_consecutive_fails = function() { return state.dns_consecutive_fails; };
    self.healthy_streak = function() { return state.healthy_streak; };

    self.forget_proxy_port = forget_proxy_port;
    self.get_sing_box_pid = get_sing_box_pid;
    self.is_dns_working = is_dns_working;
    self.is_reload_in_progress = is_reload_in_progress;
    self.is_list_update_running = is_list_update_running;
    self.check_tachyon_cli_running = check_tachyon_cli_running;
    self.proxy_port = proxy_port;
    self.setting = setting;
    self.settings = settings;
    self.enabled = enabled;

    return self;
}

function module_exports() {
    return {
        controller,
        classify_log_line,
        smart_detect_extract_domain,
        setting,
        settings,
        enabled,
        EV
    };
}

if (sourcepath(1) != null && sourcepath(1) != "")
    return module_exports();

// CLI surface for the shell test suite, which cannot import a ucode module.
let mode = ARGV[0] || "";

if (mode == "classify") {
    let facts = classify_log_line(ARGV[1]);
    if (length(facts) == 0) exit(1);
    // One type per line, in publication order.
    for (let fact in facts)
        print(fact.type + "\n");
    exit(0);
}
else if (mode == "classify-domain") {
    for (let fact in classify_log_line(ARGV[1])) {
        if (fact.payload.domain != null) {
            print(fact.payload.domain + "\n");
            exit(0);
        }
    }
    exit(1);
}
else if (mode == "extract-domain") {
    let extracted = smart_detect_extract_domain(ARGV[1]);
    if (extracted == null) exit(1);
    print(extracted + "\n");
    exit(0);
}
else if (mode == "event-types") {
    for (let k in sort(keys(EV)))
        print(k + "=" + EV[k] + "\n");
    exit(0);
}
else {
    warn("Usage: service/event_controller.uc <classify|classify-domain|extract-domain|event-types> ...\n");
    exit(1);
}