#!/usr/bin/env python3
"""ferry-metrics-exporter — a Prometheus exporter for the ferry LAN AI relay.

Serves Prometheus text (v0.0.4) on http://127.0.0.1:9092/metrics for the
llm-ferry observability stack (VictoriaMetrics + Grafana). It emits the
`ferry_` metric contract in observ/CONTRACT.md — meta, health/serving,
cumulative traffic counters, and litellm.yaml topology — with $0 cost: it
reads the local proxy access log plus litellm's /health/liveliness and
/v1/models (no model calls, no token spend). Standard library only.

It deliberately REUSES the proven parsers from the sibling `ferry-dash`
dashboard (loaded via importlib — that file has no .py extension):

  * Activity       — the incremental proxy-log tailer + cumulative counters
  * find_log(port) — proxy-log auto-discovery (incl. macOS /var/folders TMPDIR)
  * load_topology  — the tiny litellm.yaml topology parser (no PyYAML dep)
  * http_json      — the passive /health + /v1/models caller
  * ferry_live     — lanes(), chains(), EventTail, ExhaustionState

so the log regex, discovery, and topology parse are single-sourced with the
dashboard. The exporter EXTENDS Activity (JointActivity) to also track the
joint (client, status) distribution and cumulative backend-event counters
that Prometheus needs but the dashboard's marginals do not carry.

Usage:
  ferry-metrics-exporter                      # serve on http://127.0.0.1:9092
  ferry-metrics-exporter --port 9092 --ferry http://127.0.0.1:8090
  ferry-metrics-exporter --config ~/.config/ferry/litellm.yaml --key local
  ferry-metrics-exporter --events ~/…/ferry-events.ndjson --rules ~/…/event-rules.json

Endpoints:
  GET /metrics  -> Prometheus text; Content-Type text/plain; version=0.0.4
  GET /healthz  -> {"ok": true}
  anything else -> 404
"""
import argparse
import importlib.machinery
import importlib.util
import os
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

BUILD_VERSION = "1"
POLL_INTERVAL_SECONDS = 5.0     # cache the litellm calls so a 15s VM scrape doesn't hammer :8090
CONTENT_TYPE = "text/plain; version=0.0.4; charset=utf-8"


# ── reuse: load the sibling ferry-dash (no .py extension) as a module ───────
def load_ferry_dash():
    """Import ferry-dash's proven Activity / find_log / load_topology / http_json.

    ferry-dash lives at the repo/worktree root (this file is under observ/).
    Loading it only executes its top-level definitions — its server main() is
    guarded by `if __name__ == "__main__"`, so nothing binds a socket.
    """
    here = os.path.dirname(os.path.abspath(__file__))
    candidates = [
        os.path.join(here, "..", "ferry-dash"),          # observ/ -> <root>/ferry-dash
        os.path.join(here, "..", "..", "ferry-dash"),     # fallback if nested deeper
        os.path.expanduser("~/code/llm-ferry/ferry-dash"),
    ]
    tried = []
    for c in candidates:
        c = os.path.abspath(c)
        tried.append(c)
        if os.path.isfile(c):
            loader = importlib.machinery.SourceFileLoader("ferry_dash", c)
            spec = importlib.util.spec_from_loader("ferry_dash", loader)
            mod = importlib.util.module_from_spec(spec)
            loader.exec_module(mod)
            return mod
    raise ImportError("ferry-dash not found (looked in: %s)" % ", ".join(tried))


_FD = load_ferry_dash()
http_json = _FD.http_json
load_topology = _FD.load_topology
find_log = _FD.find_log
Activity = _FD.Activity
# lib/ferry_live.py, reached through ferry-dash's own loader so the lane
# resolution and the lane->deployment-id map are single-sourced with the
# dashboard. None when the module is absent: every event family then has zero
# samples and omits itself, and the rest of the scrape is untouched.
live_mod = _FD._live()


# ── traffic: extend the dashboard's tailer with the joint + event counters ──
def classify_backend_event(rules, line):
    """Classify one proxy-log line as "quota_exhausted", "rate_limited", or None.

    Canonical logic is lib/ferry_live.classify_log_line — it applies the
    operator's ~/.config/ferry/event-rules.json rules plus a vendor-neutral
    floor, so this delegates to it whenever the module loaded. When it did
    not (lib/ferry_live.py missing), this inlines the SAME vendor-neutral
    floor so the two kinds are still detected with zero config: litellm's
    RateLimitError / a bare "rate_limit" mention -> rate_limited, and
    "insufficient_quota" / "insufficient credits" -> quota_exhausted.
    """
    if live_mod:
        return live_mod.classify_log_line(rules, line)
    low = line.lower()
    if "ratelimiterror" in low or "rate_limit" in low:
        return "rate_limited"
    if "insufficient_quota" in low or "insufficient credits" in low:
        return "quota_exhausted"
    return None


class JointActivity(Activity):
    """Activity + the (client, status) joint counters and cumulative backend
    events Prometheus needs. Reuses the base regex (ACCESS) and the
    incremental offset/truncation tail verbatim; it only adds dimensions the
    dashboard's marginals cannot reconstruct. Backend-event detection is
    classify_backend_event() (above), not a copy of the base class's
    substrings, so the two kinds — quota_exhausted, rate_limited — are
    single-sourced with lib/ferry_live.classify_log_line and with the
    per-deployment ExhaustionState in Collector below.

    All counters are in-memory and CUMULATIVE since exporter start (poll only
    ever adds), so `ferry_requests_total` never decreases between scrapes.
    """

    def __init__(self, logpath, rules=None):
        super().__init__(logpath)
        self.by_client_status = {}                          # (ip, status) -> count
        kinds = live_mod.TAP_KINDS if live_mod else ("quota_exhausted", "rate_limited")
        self.backend_events = {k: 0 for k in kinds}
        self.backend_event_ts = {}                          # kind -> epoch of last occurrence
        # The classifier table, loaded ONCE in build_collector() and shared
        # with Collector's ExhaustionState below, so a proxy-log line and an
        # NDJSON tap event are judged by the exact same rules. None (a caller
        # that constructs this directly, e.g. tests, without a rules file) is
        # fine — classify_backend_event() then falls straight to the floor.
        self.rules = rules

    def poll(self):
        if not self.logpath or not os.path.exists(self.logpath):
            return
        try:
            size = os.path.getsize(self.logpath)
            if size < self.offset:                          # log truncated/rotated (ferry restart)
                self.offset = 0
            with open(self.logpath, "r", errors="replace") as f:
                f.seek(self.offset)
                chunk = f.read()
                self.offset = f.tell()
        except Exception:
            return
        new = 0
        with self.lock:
            for line in chunk.splitlines():
                m = self.ACCESS.search(line)
                if m:
                    ip, _method, path, status = m.groups()
                    # count only real inference; skip /v1/models + /health polls
                    if path.startswith("/v1/chat/completions"):
                        self.total += 1
                        new += 1
                        self.by_status[status] = self.by_status.get(status, 0) + 1
                        self.by_client[ip] = self.by_client.get(ip, 0) + 1
                        key = (ip, status)
                        self.by_client_status[key] = self.by_client_status.get(key, 0) + 1
                kind = classify_backend_event(self.rules, line)
                if kind:
                    now = time.time()
                    self.backend_events[kind] = self.backend_events.get(kind, 0) + 1
                    self.backend_event_ts[kind] = now
                    self.last_event = {"kind": kind, "t": now}
            self.buckets.append(new)
            self.buckets = self.buckets[-40:]

    def traffic_snapshot(self):
        with self.lock:
            return {
                "by_client_status": dict(self.by_client_status),
                "backend_events": dict(self.backend_events),
                "backend_event_ts": dict(self.backend_event_ts),
            }


# ── Prometheus text rendering (v0.0.4) ──────────────────────────────────────
def esc_label(v):
    return str(v).replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")


def fmt_value(v):
    """A plain decimal, or None if the value is missing/NaN/inf (=> omit series)."""
    if v is None:
        return None
    if isinstance(v, bool):
        return "1" if v else "0"
    if isinstance(v, int):
        return str(v)
    try:
        f = float(v)
    except (TypeError, ValueError):
        return None
    if f != f or f == float("inf") or f == float("-inf"):     # NaN / inf -> omit
        return None
    if f.is_integer():
        return str(int(f))
    return repr(f)


def render_labels(pairs):
    if not pairs:
        return ""
    return "{" + ",".join('%s="%s"' % (k, esc_label(v)) for k, v in pairs) + "}"


class Family:
    """One metric NAME: HELP + TYPE emitted once, then its (sorted) samples."""

    def __init__(self, name, help_text, typ):
        self.name = name
        self.help_text = help_text
        self.typ = typ
        self.samples = []                    # (label_pairs_or_None, value)

    def add(self, value, labels=None):
        self.samples.append((labels, value))
        return self

    def render(self):
        body = []
        for labels, value in self.samples:
            s = fmt_value(value)
            if s is None:                    # omit a series whose source value is missing; never NaN
                continue
            body.append("%s%s %s" % (self.name, render_labels(labels), s))
        if not body:
            return []                        # no samples -> omit the whole family (incl. HELP/TYPE)
        body.sort()                          # deterministic ordering within the family
        return ["# HELP %s %s" % (self.name, self.help_text),
                "# TYPE %s %s" % (self.name, self.typ)] + body


# ── collector ───────────────────────────────────────────────────────────────
class Collector:
    def __init__(self, ferry, key, config_path, activity, started=None,
                 events_path=None, rules_path=None, rules=None):
        self.ferry = ferry.rstrip("/")
        self.key = key
        self.config_path = config_path
        self.activity = activity
        self.started = started if started is not None else time.time()
        self.poll_interval = POLL_INTERVAL_SECONDS
        # -- the event half: everything below is optional ------------------
        # No events file, or no lib/ferry_live.py, means every event family has
        # zero samples and omits itself. The topology and traffic halves are
        # untouched, so an exporter running against a proxy with the tap off
        # exports exactly what it did before this feature existed.
        self._live = live_mod
        self._tail = None
        self._exhaustion = None
        if events_path and self._live and os.path.exists(events_path):
            try:
                # From EOF, not from the start: a 64MB backlog replayed into a
                # cumulative counter on every restart would be pure fiction.
                # These counters mean "since exporter start", exactly like
                # ferry_requests_total.
                self._tail = self._live.EventTail(events_path)
                # `rules`, when given, is the classifier table build_collector()
                # already loaded ONCE and shares with JointActivity's backend-
                # event counters, so a proxy-log line and an NDJSON tap event
                # are judged by the exact same rules object. A direct caller
                # (e.g. tests) that has no pre-loaded rules can still pass a
                # path via rules_path and this loads it itself.
                loaded_rules = rules if rules is not None else self._live.load_rules(rules_path)
                self._exhaustion = self._live.ExhaustionState(loaded_rules)
            except Exception:
                self._tail = None
                self._exhaustion = None
        self._events = {}       # (lane, deployment, provider, outcome) -> n
        self._edges = {}        # (lane, from, to, code) -> n
        self._dropped = 0
        # one topology parse per scrape, shared by three blocks
        self._topo = None
        self._topo_ts = None
        # litellm poll cache (refreshed at most every poll_interval)
        self._llm_lock = threading.Lock()
        self._llm_ts = 0.0
        self._ferry_up = False
        self._health_ms = None
        self._models = []

    def topology(self, now):
        """The litellm.yaml parse for THIS scrape.

        Three blocks read it; `now` is identical across them within one
        render(), so the file is parsed once and never cached across scrapes —
        a config edit is visible on the very next one.
        """
        if self._topo_ts != now or self._topo is None:
            self._topo = load_topology(self.config_path)
            self._topo_ts = now
        return self._topo

    # -- litellm health/serving, cached so a fast scrape can't hammer :8090 --
    def refresh_litellm(self, now=None, force=False):
        now = time.time() if now is None else now
        with self._llm_lock:
            if not force and self._llm_ts > 0 and (now - self._llm_ts) < self.poll_interval:
                return
            live = http_json(self.ferry + "/health/liveliness", self.key, timeout=4)
            self._ferry_up = bool(live.get("ok"))
            self._health_ms = live.get("ms")
            models = []
            if self._ferry_up:
                m = http_json(self.ferry + "/v1/models", self.key, timeout=4)
                if m.get("ok"):
                    data = (m.get("json") or {}).get("data") or []
                    models = [x.get("id") for x in data
                              if isinstance(x, dict) and x.get("id")]
            self._models = models
            self._llm_ts = now

    def render(self):
        """Never raises. Always returns at least ferry_exporter_up + build_info."""
        now = time.time()
        out = []
        # -- meta (always emitted, before anything that could fail) ----------
        out += Family("ferry_exporter_up",
                      "1 if the ferry metrics exporter is running.", "gauge").add(1).render()
        out += Family("ferry_exporter_build_info",
                      "Ferry metrics exporter build info.", "gauge").add(
                          1, [("version", BUILD_VERSION)]).render()
        out += Family("ferry_scrape_timestamp_seconds",
                      "Unix time when this scrape was produced.", "gauge").add(now).render()
        out += Family("ferry_exporter_uptime_seconds",
                      "Seconds since the exporter started.", "gauge").add(
                          int(now - self.started)).render()
        # -- dynamic sources; each guarded so one failure can't drop others --
        for block in (self._health_block, self._traffic_block, self._topology_block,
                      self._lanes_block, self._events_block):
            try:
                out += block(now)
            except Exception as e:
                sys.stderr.write("ferry-metrics-exporter: %s error: %s\n"
                                 % (getattr(block, "__name__", "collect"), e))
        return "\n".join(out) + "\n"

    # -- health / serving -------------------------------------------------
    def _health_block(self, now):
        self.refresh_litellm(now)
        out = []
        out += Family("ferry_up",
                      "1 if litellm /health/liveliness returns ok, else 0.", "gauge").add(
                          1 if self._ferry_up else 0).render()
        if self._ferry_up and self._health_ms is not None:
            out += Family("ferry_health_check_latency_ms",
                          "Round-trip ms to litellm /health/liveliness (passive, no token spend).",
                          "gauge").add(self._health_ms).render()
        out += Family("ferry_models_served",
                      "Number of model ids served by litellm /v1/models.", "gauge").add(
                          len(self._models)).render()
        mi = Family("ferry_model_info", "One series per served model id.", "gauge")
        for mid in self._models:
            mi.add(1, [("model", mid)])
        out += mi.render()
        return out

    # -- traffic (cumulative counters from the proxy log) -----------------
    def _traffic_block(self, now):
        self.activity.poll()
        snap = self.activity.traffic_snapshot()
        out = []
        req = Family("ferry_requests_total",
                     "Inference requests (/v1/chat/completions) by client IP and HTTP status.",
                     "counter")
        for (client, status), n in snap["by_client_status"].items():
            req.add(n, [("client", client), ("status", status)])
        out += req.render()
        ev = Family("ferry_backend_events_total",
                    "Backend events detected in the proxy log, by kind.", "counter")
        for kind, n in snap["backend_events"].items():
            if n:                               # omit a kind that has never fired
                ev.add(n, [("kind", kind)])
        out += ev.render()
        ts = Family("ferry_backend_event_timestamp_seconds",
                    "Unix time of the last backend event of each kind.", "gauge")
        for kind, epoch in snap["backend_event_ts"].items():
            ts.add(epoch, [("kind", kind)])
        out += ts.render()
        return out

    # -- topology (parse litellm.yaml) ------------------------------------
    def _topology_block(self, now):
        topo = self.topology(now)
        out = []
        groups = topo.get("groups") or {}
        if not topo.get("error"):
            counts = [g.get("count", 0) for g in groups.values()]
            pool = max(counts) if counts else 0
            out += Family("ferry_worker_pool_size",
                          "Number of deployments in the largest pooled model_name group.",
                          "gauge").add(pool).render()
            dep = Family("ferry_deployment_info",
                         "One series per unique (model_name, underlying model) deployment.",
                         "gauge")
            seen = set()
            for mname, g in groups.items():
                for model in (g.get("models") or []):
                    key = (mname, model)
                    if key in seen:
                        continue
                    seen.add(key)
                    dep.add(1, [("model_name", mname), ("model", model)])
            out += dep.render()
            # The driving lane is `orch`; `orchestrator` is the pre-rename name and is
            # still accepted so an old config keeps reporting a real chain length
            # instead of silently flatlining at 0.
            _fbs = topo.get("fallbacks") or {}
            fb = _fbs.get("orch") or _fbs.get("orchestrator") or []
            out += Family("ferry_fallback_chain_length",
                          "Length of the driving lane's strict fallback chain (0 if none).",
                          "gauge").add(len(fb)).render()
        # config mtime is independent of parse success
        try:
            mtime = os.path.getmtime(self.config_path)
        except Exception:
            mtime = None
        out += Family("ferry_route_config_mtime_seconds",
                      "Modification time (unix seconds) of litellm.yaml.", "gauge").add(
                          mtime).render()
        return out


    # -- lanes: the chain and the pools, with labels -----------------------
    def _lanes_block(self, now):
        """Every hop of every lane as its own series.

        `ferry_worker_pool_size` and `ferry_fallback_chain_length` are NOT
        replaced by these — ferry-backends.json panels 1-2 and the alert rules
        read the unlabelled scalars, and they keep their exact meaning. These
        add the dimensions the scalars cannot carry: which lane, which position,
        which deployment.
        """
        topo = self.topology(now)
        if topo.get("error") or not self._live:
            return []
        hop_fam = Family(
            "ferry_lane_hop",
            "One series per deployment sitting at a position in a lane's chain.",
            "gauge")
        length = Family(
            "ferry_lane_chain_length",
            "Hops in a lane's chain, counting the primary (1 = no fallbacks).",
            "gauge")
        pool = Family(
            "ferry_pool_size",
            "Deployments sharing one model_name — a pool litellm load-balances "
            "across, not an ordered chain it walks. 0 means the hop is named by "
            "a chain but defined nowhere.",
            "gauge")
        sizes = {}
        for lane in self._live.lanes(topo):
            length.add(len(lane["hops"]), [("lane", lane["name"])])
            for position, hop in enumerate(lane["hops"]):
                sizes[hop["name"]] = hop["pool_size"]
                for d in hop["deployments"]:
                    hop_fam.add(1, [
                        ("lane", lane["name"]),
                        ("position", str(position)),
                        ("hop", hop["name"]),
                        # An unset model_info.id cannot be joined to an event.
                        # Saying "unknown" makes the gap visible; omitting the
                        # series would hide a deployment that really is routed to.
                        ("deployment", d["id"] or "unknown"),
                        ("model", d["model"]),
                        ("provider", d["provider"]),
                        ("pool_size", str(hop["pool_size"])),
                    ])
        for name in sorted(sizes):
            pool.add(sizes[name], [("hop", name)])
        return hop_fam.render() + length.render() + pool.render()

    # -- events: per-request truth from the front-door tap -----------------
    def _events_block(self, now):
        """Fold every event appended since the last scrape, then render.

        The tap's stream is the ONLY place a request is joined to the
        deployment that served it — the proxy access log carries no model at
        all — so these families exist for facts nothing else in the stack can
        report.
        """
        if not self._tail:
            return []
        topo = self.topology(now)
        chains, provider_of = {}, {}
        if not topo.get("error"):
            try:
                chains = self._live.chains(topo)
                for lane in self._live.lanes(topo):
                    for hop in lane["hops"]:
                        for d in hop["deployments"]:
                            if d["id"]:
                                provider_of[d["id"]] = d["provider"]
            except Exception:
                chains, provider_of = {}, {}

        for rec in self._tail.read_new():
            self._fold_event(rec, chains)

        out = []
        ev = Family("ferry_events_total",
                    "Requests seen by the front-door tap, by lane, serving "
                    "deployment, provider and outcome. Cumulative since "
                    "exporter start.", "counter")
        for (lane, dep, provider, outcome), n in self._events.items():
            ev.add(n, [("lane", lane), ("deployment", dep),
                       ("provider", provider), ("outcome", outcome)])
        out += ev.render()

        ed = Family("ferry_fallback_edges_total",
                    "Times a lane moved from one deployment to the next, with "
                    "the status code that caused it. Cumulative since exporter "
                    "start.", "counter")
        for (lane, frm, to, code), n in self._edges.items():
            ed.add(n, [("lane", lane), ("from_deployment", frm),
                       ("to_deployment", to), ("code", code)])
        out += ed.render()

        snap = self._exhaustion.snapshot() if self._exhaustion else {}
        state = Family("ferry_deployment_state",
                       "1 for the state a deployment is in RIGHT NOW; a state "
                       "it has left emits nothing, so an alert clears when the "
                       "outage does.", "gauge")
        since = Family("ferry_deployment_state_since_seconds",
                       "Seconds the deployment has held its current state.",
                       "gauge")
        for dep in sorted(snap):
            info = snap[dep]
            state.add(1, [("deployment", dep),
                          ("provider", provider_of.get(dep, "unknown")),
                          ("state", info["state"])])
            since.add(int(max(0, now - info["since"])), [("deployment", dep)])
        out += state.render()
        out += since.render()

        # Always emitted once a tap is being read, including at 0: a counter
        # that only appears after the first drop cannot be rate()'d, and an
        # overflowing tap makes every other number on this page an undercount.
        out += Family("ferry_events_dropped_total",
                      "Events the tap's bounded queue dropped rather than "
                      "block a response, as announced in the stream.",
                      "counter").add(self._dropped).render()
        return out

    def _fold_event(self, rec, chains):
        """One record into the cumulative counters and the exhaustion state."""
        if not isinstance(rec, dict):
            return
        if rec.get("notice") == "dropped":
            # A different shape from a request record, discriminated on one key
            # rather than guessed from missing fields. `n` is cumulative in the
            # writing process, so the newest notice carries the true total and a
            # proxy restart reads as an ordinary counter reset.
            try:
                self._dropped = max(self._dropped, int(rec.get("n") or 0))
            except Exception:
                pass
            return

        raw_lane = rec.get("lane")
        lane = raw_lane or "unknown"
        served = rec.get("deployment") or "unknown"
        provider = rec.get("provider") or "unknown"
        try:
            status = int(rec.get("status") or 0)
        except Exception:
            status = 0
        outcome = "ok" if 200 <= status < 300 else "error"
        key = (lane, served, provider, outcome)
        self._events[key] = self._events.get(key, 0) + 1

        chain = chains.get(raw_lane) or []
        for i, hop in enumerate(rec.get("hop_errors") or []):
            if not isinstance(hop, dict):
                continue
            frm = chain[i] if i < len(chain) else ""
            to = chain[i + 1] if i + 1 < len(chain) else ""
            if not to:
                to = rec.get("deployment") or ""
            # A hop with no configured model_info.id leaves an empty slot, and
            # an edge with an unknown end would attribute a failure to a
            # backend that may be healthy. Counting nothing beats counting the
            # wrong thing. Likewise a non-zero `fallbacks` with no hop_errors
            # yields no edge: without a code and both ends there is no edge.
            if not frm or not to:
                continue
            code = str(hop.get("code") or "")
            ekey = (lane, frm, to, code)
            self._edges[ekey] = self._edges.get(ekey, 0) + 1

        if self._exhaustion:
            self._exhaustion.observe(rec, chain=chain or None)


# ── HTTP server ─────────────────────────────────────────────────────────────
def make_handler(collector):
    class Handler(BaseHTTPRequestHandler):
        def _send(self, code, body, ctype):
            b = body.encode() if isinstance(body, str) else body
            self.send_response(code)
            self.send_header("Content-Type", ctype)
            self.send_header("Content-Length", str(len(b)))
            self.end_headers()
            try:
                self.wfile.write(b)
            except Exception:
                pass

        def do_GET(self):
            path = self.path.split("?")[0]
            if path == "/metrics":
                try:
                    text = collector.render()
                except Exception as e:            # last-ditch: never 500
                    sys.stderr.write("ferry-metrics-exporter: render failed: %s\n" % e)
                    text = ("# HELP ferry_exporter_up 1 if the ferry metrics exporter is running.\n"
                            "# TYPE ferry_exporter_up gauge\nferry_exporter_up 1\n"
                            "# HELP ferry_exporter_build_info Ferry metrics exporter build info.\n"
                            "# TYPE ferry_exporter_build_info gauge\n"
                            'ferry_exporter_build_info{version="%s"} 1\n' % BUILD_VERSION)
                self._send(200, text, CONTENT_TYPE)
            elif path == "/healthz":
                self._send(200, '{"ok":true}', "application/json")
            else:
                self._send(404, "not found\n", "text/plain; charset=utf-8")

        def log_message(self, *a):                # silence per-request stderr spam
            pass

    return Handler


def build_collector(args, started=None):
    ferry_port = args.ferry.rsplit(":", 1)[-1].split("/")[0]
    logpath = args.log or find_log(ferry_port)
    rules_path = getattr(args, "rules", None)
    # Load the classifier table ONCE, here — before the tailer (JointActivity)
    # is constructed — so the proxy-log backend-event counters and the
    # per-deployment ExhaustionState below both classify against the exact
    # same rules object instead of each re-parsing event-rules.json.
    rules = live_mod.load_rules(rules_path) if live_mod else None
    activity = JointActivity(logpath, rules=rules)
    collector = Collector(args.ferry, args.key, args.config, activity, started=started,
                          events_path=getattr(args, "events", None),
                          rules=rules)
    return collector, logpath


def parse_args(argv=None):
    ap = argparse.ArgumentParser(
        description="Prometheus metrics exporter for the ferry LAN AI relay.")
    ap.add_argument("--port", type=int, default=9092, help="exporter port (default 9092)")
    ap.add_argument("--ferry", default="http://127.0.0.1:8090", help="ferry/litellm base URL")
    ap.add_argument("--config", default=os.path.expanduser("~/.config/ferry/litellm.yaml"),
                    help="litellm.yaml route config path")
    ap.add_argument("--log", default=None, help="proxy log path (auto-discovered if omitted)")
    ap.add_argument("--key", default="local", help="proxy auth bearer (default 'local')")
    # Same defaults as ferry-dash, so both read the one stream the tap writes.
    ap.add_argument("--events", default=os.path.join(
        (os.environ.get("TMPDIR") or "/tmp").rstrip("/"),
        "ferry-logs", "ferry-events.ndjson"),
        help="per-request event stream written by the front-door tap")
    ap.add_argument("--rules", default=os.path.expanduser(
        "~/.config/ferry/event-rules.json"),
        help="exhaustion classifier table (see event-rules.example.json)")
    return ap.parse_args(argv)


def main(argv=None, serve=True):
    args = parse_args(argv)
    collector, logpath = build_collector(args)
    try:
        srv = ThreadingHTTPServer(("127.0.0.1", args.port), make_handler(collector))
    except OSError as e:
        raise SystemExit("ferry-metrics-exporter: cannot bind 127.0.0.1:%d (%s). Try --port."
                         % (args.port, e))
    print("ferry-metrics-exporter -> http://127.0.0.1:%d/metrics   "
          "(ferry %s, config %s, log %s, events %s)"
          % (args.port, collector.ferry, args.config, logpath or "not found",
             args.events if collector._tail else "off"))
    print("Ctrl-C to stop.")
    if not serve:
        return srv, collector
    try:
        srv.serve_forever()
    except KeyboardInterrupt:
        print("\nferry-metrics-exporter stopped.")
    finally:
        srv.server_close()
    return 0


if __name__ == "__main__":
    main()
