#!/usr/bin/env python3
"""inner.py -- runs INSIDE the engine's unit and its no-network sandbox: the egress probe, the server, its health,
G0, then the phase's requests. Writes only to the arm's run directory; talks only to loopback.

    python3 inner.py --spec <run dir>/spec.json

ORDER (each step writes one line to status.jsonl, which the outer driver follows):
  1. probe     the egress probe (`egress_probe`): only `lo` exists, a DNS lookup fails, the canary connect fails, a
               loopback listener answers -- all four, or status "preflight_refused" and exit before any engine starts.
  2. launch    the server as this process's child (a fresh API key from `secrets`, passed in the environment; every
               STRATA_* variable but the key dropped; the variable names recorded).
  3. ready     GET /health answers 200 (both engines) within the timeout.
  4. g0        Strata: the native AVX2 line and no control-vector line (engines.strata_g0); a CPU refusal is the
               stop condition "cpu_refused". llama.cpp: one CUDA device, no Vulkan, the offload line. An EXIT before
               /health is STOPPED `exited_before_ready` (re-openable) unless the log carries a line G0 refuses on.
  4b. identity llama.cpp: the start banner's `build N (<commit>)` against the arm's commit (a mismatch is an
               attempt-scoped PREFLIGHT refusal); then the thread line of either engine, for every row (PREREG A3).
  5. phase     speed: for each context x repeat, one streamed chat request (256 tokens, temperature 0, thinking off,
               no seed) timed on this process's clock with the kit's rule (engine.decode_chat_sse + finish_record);
               quality: the Jev task_c and doorman task_d generators, unedited, through a patched `call`;
               fit / load: a short hold with no request.
  6. done
"""
import argparse
import errno
import http.client
import json
import os
import secrets
import socket
import subprocess
import sys
import time
import types

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.join(HERE, "vendor", "vllm-seats"))
sys.dont_write_bytecode = True

import common as C            # noqa: E402
import engines as EN          # noqa: E402
import prompts as P           # noqa: E402
import engine as KE           # noqa: E402  the kit's engine.py (stream record, SSE reader, timing rule)
import engine_llamaserver as KL   # noqa: E402  the kit's device gate
import gates as KG            # noqa: E402  the kit's offload gate
import nonces as KN           # noqa: E402  the kit's seeded nonce


class Inner:
    def __init__(self, spec):
        self.s = spec
        self.run_dir = spec["run_dir"]
        self.status_path = os.path.join(self.run_dir, "status.jsonl")
        self.rows_path = os.path.join(self.run_dir, "inner-rows.jsonl")
        self.port = int(spec["port"])
        self.key = secrets.token_hex(24)
        self.proc = None
        self.server_log = os.path.join(self.run_dir, "server.log")
        self.engine_log = spec.get("strata_engine_log")
        self.threads = None

    # ---- status ------------------------------------------------------------------------------ #
    def status(self, phase, **kw):
        C.append_jsonl(self.status_path, dict({"t": round(time.time(), 3), "utc": C.utc(), "phase": phase}, **kw))

    # ---- 1. the probe -------------------------------------------------------------------------- #
    def egress_probe(self):
        """Never sends a packet off the box. The branch is the spec's `egress_mode`, never the interfaces seen.

        netns (the only mode `run` and `fit` accept): `verdict` is "enforced" only if ALL four hold -- only `lo`
        exists; a DNS lookup fails; a connect to the kit's documentation-range canary (egress.CANARY_ADDR) fails;
        a loopback listener answers. With any interface beyond `lo` the probe tries neither the lookup nor the
        connect (either could leave the box) and reads NOT ENFORCED with that reason (AUDIT-HARNESS I-5).

        ipfilter (`driver.py probe-ipfilter` only, a dry-run probe that never launches a server): a TCP connect to the
        box's OWN non-loopback address, where the outer driver holds a listener it has just reached itself, must be
        dropped by the filter. A refused connect (ECONNREFUSED) reached the host's stack, so it reads NOT ENFORCED,
        as does a connect that succeeds; only a dropped or forbidden connect (a timeout, EPERM, EACCES, an
        unreachable network or host) reads enforced. The address is learned from a UDP connect (a route lookup, no
        packet) and never written down."""
        import egress as KX
        names = [n for _, n in socket.if_nameindex()]
        mode = self.s.get("egress_mode") or "netns"
        r = {"ifaces_count": len(names), "only_lo": names == ["lo"], "mode": mode}
        why = []
        if mode == "netns":
            if not r["only_lo"]:
                r["dns"] = "not tried (interfaces beyond lo: a lookup could leave the box)"
                r["outside_connect"] = "not tried (interfaces beyond lo: a connect could leave the box)"
                why.append(f"the sandbox left {len(names)} interfaces, not only lo")
            else:
                try:
                    socket.getaddrinfo(self.s.get("probe_host", "github.com"), 443)
                    r["dns"] = "RESOLVED"
                except OSError as e:
                    r["dns"] = f"failed ({type(e).__name__})"
                c = socket.socket()
                c.settimeout(3)
                try:
                    c.connect(KX.CANARY_ADDR)
                    r["outside_connect"] = "CONNECTED"
                except OSError as e:
                    r["outside_connect"] = f"failed ({errno.errorcode.get(e.errno, e.errno)})"
                finally:
                    c.close()
                if not r["dns"].startswith("failed"):
                    why.append("a DNS lookup resolved")
                if not r["outside_connect"].startswith("failed"):
                    why.append("the canary connect did not fail: " + r["outside_connect"])
        elif mode == "ipfilter":
            r["dns"] = "not tried (no sandbox: a lookup would leave the box)"
            u = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            try:
                u.connect(KX.CANARY_ADDR)
                own = u.getsockname()[0]
            finally:
                u.close()
            c = socket.socket()
            c.settimeout(3)
            try:
                c.connect((own, int(self.s["probe_listener_port"])))
                r["outside_connect"] = "CONNECTED (the box's own non-loopback address, held by the driver)"
            except socket.timeout:
                r["outside_connect"] = "failed (ETIMEDOUT: dropped)"
            except OSError as e:
                r["outside_connect"] = f"failed ({errno.errorcode.get(e.errno, e.errno)})"
            finally:
                c.close()
            dropped = ("ETIMEDOUT", "EPERM", "EACCES", "ENETUNREACH", "EHOSTUNREACH")
            if not any(f"({x}" in r["outside_connect"] for x in dropped):
                why.append("the connect to the box's own non-loopback address was not dropped: " + r["outside_connect"]
                           + (" (a refusal means the packet reached the host's stack)"
                              if "ECONNREFUSED" in r["outside_connect"] else ""))
        else:
            why.append(f"unknown egress_mode {mode!r}")
        ls = socket.socket()
        ls.bind(("127.0.0.1", 0))
        ls.listen(1)
        c2 = socket.socket()
        c2.settimeout(3)
        try:
            c2.connect(ls.getsockname())
            r["loopback"] = "connected"
        except OSError as e:
            r["loopback"] = f"failed ({errno.errorcode.get(e.errno, e.errno)})"
        finally:
            c2.close()
            ls.close()
        if r["loopback"] != "connected":
            why.append("loopback does not answer (the client could not reach the server)")
        r["verdict"] = "enforced" if not why else "NOT ENFORCED"
        r["why"] = "; ".join(why) or ("only lo; DNS fails; the canary connect fails; loopback answers"
                                      if mode == "netns" else "a connect to the box's own non-loopback address is "
                                      "dropped (the IP filter is enforced); loopback answers")
        return r

    # ---- 2-3. the server ----------------------------------------------------------------------- #
    def launch(self):
        """The server as this process's child. Its environment: the unit's, less proxies and every STRATA_* variable
        but the key (engines.server_environment, AUDIT-HARNESS I-2). Returns (pid, the variable NAMES it was handed,
        the names dropped): names only, never a value."""
        env, dropped = EN.server_environment(os.environ, self.s.get("server_env"), self.s["key_env"], self.key)
        fh = open(self.server_log, "ab")
        self.proc = subprocess.Popen(self.s["server_argv"], env=env, stdout=fh, stderr=subprocess.STDOUT,
                                     stdin=subprocess.DEVNULL, cwd=self.s.get("server_cwd") or self.run_dir)
        fh.close()
        return self.proc.pid, sorted(env), dropped

    def call(self, method, path, body=None, timeout=60.0, auth=True):
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=timeout)
        try:
            data = None if body is None else json.dumps(body).encode()
            h = {"Content-Type": "application/json"} if data else {}
            if auth:
                h["Authorization"] = f"Bearer {self.key}"
            conn.request(method, path, body=data, headers=h)
            r = conn.getresponse()
            t = r.read().decode("utf-8", "replace")
            try:
                return r.status, json.loads(t)
            except ValueError:
                return r.status, t
        except (OSError, http.client.HTTPException) as e:
            return None, f"{type(e).__name__}: {e}"
        finally:
            conn.close()

    def wait_ready(self):
        t0 = time.monotonic()
        deadline = t0 + float(self.s.get("health_timeout_s", 1200))
        while time.monotonic() < deadline:
            if self.proc.poll() is not None:
                return {"ready": False, "exited": self.proc.returncode, "seconds": round(time.monotonic() - t0, 2)}
            st, obj = self.call("GET", "/health", timeout=3.0, auth=False)
            if st == 200:
                return {"ready": True, "seconds": round(time.monotonic() - t0, 2),
                        "health": obj if isinstance(obj, dict) else str(obj)[:200]}
            time.sleep(1.0)
        return {"ready": False, "timeout_s": self.s.get("health_timeout_s"), "seconds": round(time.monotonic() - t0, 2)}

    def logs(self):
        return EN.read_text(self.engine_log) if self.engine_log else "", EN.read_text(self.server_log)

    def g0(self, ready):
        eng, srv = self.logs()
        if not ready.get("ready") and ready.get("exited") is not None:
            # AUDIT-HARNESS B-2: an exit before /health is G0's only where the engine's own words say so
            g = EN.early_exit_verdict(self.s["engine"], eng, srv, KL.device_list)
            g["health"] = ready
            return g
        if self.s["engine"] == "strata":
            g = EN.strata_g0(eng, srv)
            if not ready.get("ready") and g["verdict"] == "pass":
                g.update(verdict="refused", why=f"/health never answered ({ready})")
            if ready.get("ready"):
                # the engine's INFO line reaches only the server (its stdout pipe), so its facts are read where the
                # server publishes them: GET /metrics "engine" (server.py:643-667)
                st, met = self.call("GET", "/metrics", timeout=10.0)
                info = (met or {}).get("engine") if st == 200 and isinstance(met, dict) else None
                g["engine_info"] = info
                if info is not None and str(info.get("cvec", "0")) not in ("0", "", "None") and \
                        g["verdict"] == "pass":
                    g.update(verdict="refused", why=f"/metrics engine.cvec reads {info.get('cvec')!r}: a control "
                                                    f"vector is loaded")
                if info is None and g["verdict"] == "pass":
                    g.update(verdict="refused", why=f"/metrics did not answer ({st}): the cvec state is unread")
        else:
            if not ready.get("ready"):
                return {"verdict": "refused", "why": f"/health never answered ({ready})",
                        "log_tail": srv[-1500:]}
            g = EN.llama_g0(srv, KL.device_list, KG.all_layers)
        g["health"] = ready
        return g

    def alive(self):
        return self.proc is not None and self.proc.poll() is None

    def identity(self):
        """AUDIT-HARNESS I-8, llama.cpp: the `build N (<commit>)` banner the server printed at start, against the
        arm's commit. (Strata prints no build line; its tree and build are checked by the driver before the lock.)"""
        if self.s["engine"] != "llama.cpp":
            return None
        want = self.s.get("expect_commit")
        b = EN.parse_llama_build(self.logs()[1])
        if b is None:
            return {"ok": False, "why": "the server log carries no `build N (<commit>)` line", "expected": want}
        ok = EN.commit_matches(b["commit_short"], want)
        return dict(b, ok=ok, expected=want, why=("the start banner names the arm's commit" if ok else
                                                  f"the start banner names {b['commit_short']}, not {want}"))

    # ---- 5a. speed ------------------------------------------------------------------------------ #
    def stream_chat(self, user_text, max_tokens, timeout):
        rec = KE.new_record(self.s["engine"], self.s["arm"], "chat")
        body = {"model": self.s.get("model_name", "qwen3.8-flash-next"),
                "messages": [{"role": "user", "content": user_text}], "max_tokens": int(max_tokens),
                "temperature": 0, "stream": True, "stream_options": {"include_usage": True},
                "chat_template_kwargs": {"enable_thinking": False}}
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=timeout)
        payloads = []
        t0 = time.monotonic()
        rec["t_send_epoch"] = time.time()
        try:
            conn.request("POST", "/v1/chat/completions", body=json.dumps(body).encode(),
                         headers={"Content-Type": "application/json", "Authorization": f"Bearer {self.key}"})
            resp = conn.getresponse()
            rec["http_status"] = resp.status
            if resp.status != 200:
                rec["error"] = resp.read().decode("utf-8", "replace")[:400]
            else:
                KE.decode_chat_sse(_Tee(resp, payloads), t0, rec)
        except (OSError, http.client.HTTPException) as e:
            rec["error"] = f"{type(e).__name__}: {e}"
        finally:
            conn.close()
        rec["t_end_epoch"] = time.time()
        timings = None
        for p in payloads:
            try:
                o = json.loads(p)
            except ValueError:
                continue
            if isinstance(o, dict) and o.get("timings"):
                timings = o["timings"]
        rec["server_timings"] = timings
        # AUDIT-HARNESS B-1: frames that carried generated text, counted from the same payloads the kit read (the
        # kit counts token_frames, which include the finish frame; a frame may carry several tokens on Strata)
        rec["content_frames"] = sum(1 for p in payloads if _content_frame(p))
        KE.finish_record(rec)
        rec["status"] = "ok" if not rec.get("error") and rec.get("finish_reason") else "failed"
        if rec["status"] != "ok":
            KE.unscore(rec)
        return rec

    def speed(self):
        slices = C.read_json(self.s["slices_path"])
        if slices is None:
            if self.s["engine"] != "llama.cpp":
                return {"stop": {"kind": "no_slices", "sentence": "REFUSED: SLICES.json is absent; an L arm builds it"}}
            self.status("slices_build")
            sl = P.build_slices(self._tokenize, log=lambda m: self.status("slices_progress", note=m))
            if os.path.exists(self.s["slices_path"]):        # another run froze it meanwhile: never overwrite
                slices = C.read_json(self.s["slices_path"])
            else:
                C.write_json(self.s["slices_path"], sl)
                slices = sl
            self.status("slices_frozen", sha={k: v["sha256"][:12] for k, v in slices["slices"].items()},
                        tokens={k: v["tokens"] for k, v in slices["slices"].items()})
        repeats = int(self.s.get("repeats", 3))
        n = 0
        for ctx in self.s["contexts"]:
            sl = slices["slices"][ctx]
            for rep in range(repeats):
                n += 1
                text, nonce = P.prompt(sl["text"], ctx, rep, KN)
                eng_off = os.path.getsize(self.engine_log) if self.engine_log and os.path.exists(self.engine_log) else 0
                srv_off = os.path.getsize(self.server_log) if os.path.exists(self.server_log) else 0
                self.status("run_start", run=n, ctx=ctx, repeat=rep)
                rec = self.stream_chat(text, self.s.get("max_tokens", 256), float(self.s.get("request_timeout_s",
                                                                                             2400)))
                row = {"kind": "run", "run": n, "ctx": ctx, "repeat": rep, "slice_sha256": sl["sha256"],
                       "slice_tokens_llama": sl["tokens"], "prompt_sha256": C.sha256_bytes(text.encode()),
                       "nonce": nonce.strip(), "requests_since_start": n, "stream": _slim(rec)}
                if self.s["engine"] == "strata":
                    st, met = self.call("GET", "/metrics", timeout=10.0)
                    req = (met.get("requests") or [None])[0] if st == 200 and isinstance(met, dict) else None
                    eng_slice = _since(self.engine_log, eng_off) + "\n" + _since(self.server_log, srv_off)
                    row["strata"] = EN.strata_request_facts(eng_slice, req)
                C.append_jsonl(self.rows_path, row)
                self.status("run_end", run=n, ctx=ctx, repeat=rep, ok=rec["status"] == "ok",
                            t0=rec.get("t_send_epoch"), t1=rec.get("t_end_epoch"))
                if rec["status"] != "ok" and not self.alive():
                    return {"stop": {"kind": "engine_died", "sentence": f"STOPPED: the server exited (rc "
                                     f"{self.proc.returncode}) during run {n} ({ctx}, repeat {rep}): "
                                     f"{rec.get('error')}"}}
                if self.stop_requested():
                    return {"stop": {"kind": "outer_stop", "sentence": "STOPPED by the outer driver"}}
        return {"runs": n}

    def _tokenize(self, text):
        st, obj = self.call("POST", "/tokenize", {"content": text, "add_special": False}, timeout=600.0)
        if st != 200 or not isinstance(obj, dict) or "tokens" not in obj:
            raise RuntimeError(f"/tokenize answered {st}: {str(obj)[:200]}")
        return len(obj["tokens"])

    def stop_requested(self):
        return os.path.exists(os.path.join(self.run_dir, "STOP-REQUEST"))

    # ---- 5b. quality ---------------------------------------------------------------------------- #
    def quality(self):
        import quality as Q
        R, A = Q.load_jev()
        inner = self

        def call(arm, prompt, *, predict, timeout=900.0):
            body = {"model": inner.s.get("model_name", "qwen3.8-flash-next"),
                    "messages": [{"role": "user", "content": prompt}], "max_tokens": int(predict), "temperature": 0,
                    "chat_template_kwargs": {"enable_thinking": False}}
            t0 = time.monotonic()
            st, out = inner.call("POST", "/v1/chat/completions", body, timeout=timeout)
            secs = time.monotonic() - t0
            if st != 200 or not isinstance(out, dict):
                raise R.SeatError(f"HTTP {st}: {str(out)[:300]}")
            ch = (out.get("choices") or [{}])[0]
            u = out.get("usage") or {}
            Q.LAST.clear()
            Q.LAST.update(timings=out.get("timings"), finish=ch.get("finish_reason"))
            return {"logprobs": [], "message": {"content": (ch.get("message") or {}).get("content") or ""},
                    "prompt_eval_count": u.get("prompt_tokens"), "eval_count": u.get("completion_tokens"),
                    "prompt_eval_duration": None, "total_duration": secs * 1e9,
                    "done_reason": ch.get("finish_reason"), "_seconds": secs}
        R.call = call
        orig = R.decide

        def decide(arm, state, instructions, options, *, noul=False):
            row = orig(arm, state, instructions, options, noul=noul)
            tm = Q.LAST.get("timings") or {}
            row.update(x_engine=inner.s["engine"], x_arm=inner.s["arm"], x_finish_reason=Q.LAST.get("finish"),
                       x_server_prompt_ms=tm.get("prompt_ms"), x_server_predicted_ms=tm.get("predicted_ms"),
                       x_num_predict=8, x_think=False, x_epoch_end=time.time(),
                       x_threads=_thread_figure(getattr(inner, "threads", None)))
            return row
        R.decide = decide
        arm = R.Arm(self.s["arm"], self.s.get("model_name", "qwen3.8-flash-next"), "generate",
                    runtime="strata-bench", host=f"http://127.0.0.1:{self.port}", cards=("card0",),
                    num_ctx=131072, quant=self.s.get("quant", ""), box="the laptop", power_cap_w=0)
        counts = {}
        for task in ("c", "d"):
            path = os.path.join(self.run_dir, f"{self.s['arm']}.{task}.jsonl")
            self.status("task_start", task=task)
            t_a = time.time()
            k = 0
            for row in R.TASKS[task](arm, {}, {}):
                k += 1
                row.update(arm=self.s["arm"], stamp=C.utc())
                C.append_jsonl(path, row)
                if self.stop_requested():
                    return {"stop": {"kind": "outer_stop", "sentence": "STOPPED by the outer driver"}}
                if not self.alive():
                    return {"stop": {"kind": "engine_died", "sentence": f"STOPPED: the server exited during task "
                                                                         f"{task} item {k}"}}
            counts[task] = k
            self.status("task_end", task=task, items=k, t0=t_a, t1=time.time())
        return {"items": counts}

    # ---- the whole ------------------------------------------------------------------------------ #
    def main(self):
        pr = self.egress_probe()
        self.status("probe", probe=pr)
        if pr["verdict"] != "enforced":
            self.status("preflight_refused", kind="egress", sentence=f"REFUSED: the egress block is not enforced "
                                                                     f"inside the unit: {pr['why']}")
            return 3
        if self.s.get("probe_only"):                       # driver.py probe-ipfilter: never a server
            self.status("done", probe_only=True)
            return 0
        pid, env_names, dropped = self.launch()
        self.status("launch", pid=pid, env_names=env_names, env_dropped=dropped)
        ready = self.wait_ready()
        self.status("ready" if ready.get("ready") else "not_ready", **ready)
        g = self.g0(ready)
        self.status("g0", g0=g)
        if g["verdict"] == "cpu_refused":
            self.status("stop", kind="cpu_refused", sentence=f"STOPPED: {g['why']}")
            self._stop_server()
            return 6
        if g["verdict"] == "exited_before_ready":
            self.status("stop", kind="exited_before_ready",
                        sentence=f"STOPPED (exited_before_ready; re-open with --attempt N+1): {g['why']}")
            self._stop_server()
            return 6
        if g["verdict"] != "pass":
            self.status("g0_refused", sentence=f"REFUSED (G0): {g['why']}")
            self._stop_server()
            return 4
        ident = self.identity()
        if ident is not None:
            self.status("identity", identity=ident)
            if not ident["ok"]:
                self.status("preflight_refused", kind="engine_identity",
                            sentence=f"REFUSED (engine identity; re-open with --attempt N+1 once the path is right): "
                                     f"{ident['why']}")
                self._stop_server()
                return 3
        self.threads = EN.parse_threads(self.s["engine"], "\n".join(self.logs()), self.s.get("server_argv"))
        self.status("threads", threads=self.threads)
        try:
            if self.s["phase"] == "speed":
                res = self.speed()
            elif self.s["phase"] == "quality":
                res = self.quality()
            else:                                           # fit / load: loaded and answering; hold, no request
                time.sleep(float(self.s.get("fit_hold_s", 10)))
                res = {self.s["phase"]: "held", "seconds": self.s.get("fit_hold_s", 10)}
        except Exception as e:                             # noqa: BLE001 -- recorded; the outer driver decides
            self.status("error", error=f"{type(e).__name__}: {e}")
            self._stop_server()
            return 5
        if res.get("stop"):
            self.status("stop", **res["stop"])
            self._stop_server()
            return 6
        self.status("done", **res)
        self._stop_server()
        return 0

    def _stop_server(self):
        if self.proc is not None and self.proc.poll() is None:
            self.proc.terminate()
            try:
                self.proc.wait(timeout=30)
            except subprocess.TimeoutExpired:
                self.proc.kill()


class _Tee:
    """A response wrapper that keeps every `data:` payload while engine.sse_events reads it."""

    def __init__(self, resp, sink):
        self.resp, self.sink = resp, sink

    def readline(self):
        line = self.resp.readline()
        s = line.strip()
        if s.startswith(b"data:"):
            self.sink.append(s[5:].strip())
        return line

    def read(self, *a):
        return self.resp.read(*a)


def _since(path, off):
    if not path or not os.path.exists(path):
        return ""
    with open(path, "rb") as fh:
        fh.seek(off)
        return fh.read().decode("utf-8", "replace")


def _thread_figure(t):
    """The thread count a row carries (PREREG A3): llama.cpp's n_threads/n_threads_batch, Strata's pool workers."""
    if not t:
        return None
    if t.get("engine") == "llama.cpp":
        return {"n_threads": t.get("n_threads"), "n_threads_batch": t.get("n_threads_batch")}
    return {"workers": t.get("workers"), "host_thread": t.get("host_thread")}


def _content_frame(payload):
    try:
        o = json.loads(payload)
    except ValueError:
        return False
    if not isinstance(o, dict):
        return False
    for ch in (o.get("choices") or [])[:1]:
        return bool((ch.get("delta") or {}).get("content"))
    return False


def _slim(rec):
    """The stream record as inner-rows.jsonl keeps it. AUDIT-HARNESS B-1: the completion text STAYS (it is what a
    token count is checked against, and launch-once forbids re-running a speed arm); only `generated_ids` goes. The
    text is internal under PREREG section 7's Q-9 ruling: house agents may read it to audit the bench, and no summary,
    page, post or kit prints it (summary.py prints counts only). It never reaches rows.jsonl."""
    out = {k: v for k, v in rec.items() if k != "generated_ids"}
    return out


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--spec", required=True)
    a = ap.parse_args()
    spec = C.read_json(a.spec)
    sys.exit(Inner(spec).main())


if __name__ == "__main__":
    main()