#!/usr/bin/env python3
"""@coder — bullpen worker: turn a request into a Lua file, write it, run it, report.
Reactive. Engine = dspark via the gated proxy. It ONLY ever writes into its sandbox and
runs `lua5.4` there as an unprivileged user with a timeout, and — when the ticket names a
SPEC — `bullpen-attest` under the SAME bound. Nothing else (mechanical bound). It holds no
room token, so an anchored spec comes back red here BY DESIGN; see attest().
  bullpen-coder            # room loop (systemd)
  bullpen-coder --once "a lua function that reverses a string, with a self-test"
"""
import hashlib, json, os, re, shutil, subprocess, sys, time, urllib.request, urllib.error
sys.path[:0] = [p for p in (os.path.join(os.path.dirname(os.path.realpath(__file__)), "..", "lib"), "/usr/local/lib/bullpen") if os.path.isdir(p)]
import bullpen_worker as bw
import bullpen_config as cfg

SANDBOX = cfg.CODER_DIR
# Gemessen 2026-08-09 gegen `[local] qwen3.6-coding`: ein realistischer
# Auftrag (1205 Prompt-Token -> 1128 Antwort-Token) braucht 307 s bei
# 3,67 Tok/s. Die alte Grenze von 200 s war damit fuer eine echte
# Codegenerierung nie erreichbar -- der Fehlschlag sah nach Ausfall aus
# und war Arithmetik. Der Motor selbst antwortet auf einen Einzeiler in 6 s.
# 1800 wie Lurker und oc-run: dieselbe Schranke fuer jeden Modellaufruf.
# @coder faehrt als einziger ein LOKALES CPU-Modell (gemessen 3,67 Tok/s),
# also traf die knappste Frist die langsamste Maschine. 900 s sind bei
# dieser Rate rund 3300 Token Ausgabe — weniger, als eine vollstaendige
# Datei braucht.
CODEGEN_TIMEOUT = int(os.environ.get("BULLPEN_CODEGEN_TIMEOUT", "1800"))
# Ausgabebudget EINES Aufrufs. Muss das Denken mittragen, nicht nur den Code.
CODEGEN_MAX_TOKENS = int(os.environ.get("BULLPEN_CODEGEN_MAX_TOKENS", "16384"))
# C-2c-ladder: Eskalations-Leiter DEFAULT-AN (Operator-Entscheid 2026-08-22,
# fuer alle Anfragen). "0" = aus (Legacy-Pfad 0.2.10 inkl. altem
# BULLPEN_RETRY_ON_FAIL), jeder andere Wert = an — Spiegel der
# Flip-Semantik aus C-2b-T2x-flip. Sprosse 3 (Geschwister) faehrt den in
# Kampagne 3 gemessenen Arbeitspunkt (medium + 8192 + cut0, siehe
# lib/bullpen_ladder.py); leere SIBLING_URL = Leiter endet nach dem Retry.
LADDER = os.environ.get("BULLPEN_LADDER", "") != "0"
SIBLING_URL = os.environ.get("BULLPEN_SIBLING_URL", "http://dirac.fritz.box:8080")


def _int_env(name, default):
    # R1-Linie aus C-2b-T2x-flip (Review F7): Muell in einer Tuning-Variable
    # der default-an-Leiter darf den Dienst nie crash-loopen — warnen und
    # mit dem Default fahren. <= 0 zaehlt als Muell (urlopen(timeout=0)).
    raw = os.environ.get(name)
    if raw is None:
        return default
    try:
        v = int(raw)
    except ValueError:
        v = 0
    if v <= 0:
        print(f"bullpen-coder: ignoriere ungueltiges {name}={raw!r}, "
              f"Default {default}", file=sys.stderr)
        return default
    return v


SIBLING_TIMEOUT = _int_env("BULLPEN_SIBLING_TIMEOUT_S", 900)
# Zeitbudget der Sprossen 2+3 JENSEITS von Sprosse 1; Sprosse 1 wird nie
# beschnitten (Task 10 bestand bei 2283 s — langsame Gruenlaeufer sind kein
# Regressmaterial). Rest < 60 s => Sprosse uebersprungen, nie Mini-Timeout.
LADDER_EXTRA = _int_env("BULLPEN_LADDER_EXTRA_S", 1800)
ENGINE  = cfg.PROXY   # gated proxy (gate+failover)
MODEL   = cfg.MODEL
MAXBODY = 1400

SYS = ("You are a Lua coder. Given a task and any reference material, output ONLY valid Lua "
       "source that solves it AND includes a small self-test with sample data printed to stdout. "
       "No markdown, no prose, no code fences — just the Lua.")

def gen_code(request, timeout=None):
    """Returns (code, served_model). The proxy may fail over, so the model actually SERVING
    the request isn't necessarily MODEL — the response body's own `model` field is ground
    truth; report that, not the ask."""
    payload = json.dumps({"model": MODEL,
        "messages": [{"role": "system", "content": SYS}, {"role": "user", "content": request}],
        # No temperature here. The backend unit carries the model card's row for
        # whatever it serves; a request parameter would override it, and then the
        # card has no say. max_tokens stays -- that is this caller's budget, not a
        # property of the model.
        #
        # 1400 stammte von einem Modell ohne Denkmodus. qwen3.6-coding denkt (so
        # sieht es die Modellkarte vor) und legt die Denkschritte in ein eigenes
        # Feld `reasoning_content`, das aber vom selben Budget zehrt: gemessen
        # 311 Tokens fuer EINE Zeile Lua. Bei einem echten Auftrag war das Budget
        # vor dem ersten Zeichen Code aufgebraucht, `content` kam leer zurueck,
        # und hier entstand eine 0-Zeichen-Datei.
        "max_tokens": CODEGEN_MAX_TOKENS}).encode()
    req = urllib.request.Request(ENGINE, payload, {"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=(timeout or CODEGEN_TIMEOUT)) as r:
        d = json.load(r)
    wahl = d["choices"][0]
    txt = wahl["message"].get("content") or ""
    if not txt.strip():
        # Leer ist nicht dasselbe wie "Modell kann es nicht". Der Grund steht in
        # derselben Antwort; ihn wegzuwerfen hat am 2026-08-12 zwei Umlaeufe und
        # eine falsche Schuldzuweisung an das Modell gekostet.
        grund = wahl.get("finish_reason")
        nutzung = d.get("usage") or {}
        denk = (wahl["message"].get("reasoning_content")
                or wahl["message"].get("reasoning") or "")
        raise RuntimeError(
            "Modell lieferte leeren content (finish_reason=%s, completion_tokens=%s, "
            "Budget=%s%s)" % (grund, nutzung.get("completion_tokens"), CODEGEN_MAX_TOKENS,
                              ", %d Zeichen davon ins Denken" % len(denk) if denk else ""))
    m = re.search(r"```(?:lua)?\s*(.*?)```", txt, re.S)      # strip fences if the model adds them
    return (m.group(1) if m else txt).strip(), d.get("model")

def run_lua(path):
    os.chmod(path, 0o644)
    r = subprocess.run(
        ["timeout", "10", "su", "-s", "/bin/sh", "nobody", "-c", f"cd {SANDBOX} && lua5.4 {path}"],
        capture_output=True, text=True)
    out = r.stdout.strip()
    if r.stderr.strip():
        out += ("\n[stderr] " + r.stderr.strip())
    return out.strip(), r.returncode

SPEC_RE = re.compile(r"^\s*SPEC:\s*(\S+)\s*$", re.M)


def attest(impl, spec):
    """Run bullpen-attest under the same bound as the generated code itself.

    `su nobody` with a timeout — one named command with fixed arguments, not a
    shell. The test is executed by attest, so it runs caged too; that matters,
    because a spec is a file from a repo and this worker's whole point is that
    nothing it touches gets more than lua5.4 and a clock.

    Returns (certificate_text, rc) or (None, None) when the tool is absent.
    """
    tool = shutil.which("bullpen-attest") or "/opt/bullpen-src/bin/bullpen-attest"
    if not os.path.exists(tool):
        return None, None
    r = subprocess.run(
        ["timeout", "120", "su", "-s", "/bin/sh", "nobody", "-c",
         f"cd {SANDBOX} && {tool} --impl {impl} --test {spec}"],
        capture_output=True, text=True)
    return ((r.stdout or "") + (r.stderr or "")).strip(), r.returncode


PROBE_SHADOW = os.environ.get("BULLPEN_PROBE_SHADOW") == "1"
PROBE_LOG = os.environ.get("BULLPEN_PROBE_LOG", "/var/lib/bullpen-probe/shadow.jsonl")


def _shadow_begin(req, rid):
    """Roadmap P1: score the pre-generation state (shadow only — a number for
    a log line, never a decision). Returns the started record, or None. With
    the env gate off this returns IMMEDIATELY — no dict, no hashing, no I/O:
    the diff must be exactly dead then (review B1). Everything else is inside
    the try: fail-open by construction."""
    if not PROBE_SHADOW:
        return None
    try:
        import bullpen_probe
        score, embed_ms = bullpen_probe.score_messages(
            [{"role": "system", "content": SYS}, {"role": "user", "content": req}])
        return {"ts": int(time.time()), "ask_id": rid, "score": score,
                "embed_ms": embed_ms,
                "req_hash": hashlib.sha256(str(req).encode()).hexdigest()[:16]}
    except Exception:
        return None


def _shadow_write(rec, **extra):
    if rec is None:
        return
    try:
        import bullpen_probe
        bullpen_probe.shadow_log(PROBE_LOG, dict(rec, **extra))
    except Exception:
        pass


def dispatch(msg):
    req, rid = msg.get("body", ""), msg.get("id")
    os.makedirs(SANDBOX, exist_ok=True)
    t0 = time.time()
    deadline = t0 + CODEGEN_TIMEOUT + LADDER_EXTRA   # nur die Leiter liest sie
    shadow = _shadow_begin(req, rid)
    try:
        code, served = gen_code(req)
    except urllib.error.HTTPError as e:
        # Der Server schickt bei 404/503 einen JSON-Rumpf mit dem Grund und oft
        # der Liste der bekannten Modelle. Ohne ihn ist die Meldung nicht
        # diagnostizierbar -- genau das kostete am 2026-08-11 eine Stunde.
        try:
            detail = e.read()[:300].decode("utf-8", "replace")
        except Exception:
            detail = "<Rumpf nicht lesbar>"
        _shadow_write(shadow, gen_error=f"HTTP {e.code}")
        return (f"codegen error: HTTP {e.code} von {e.url} "
                f"(Modell {MODEL!r}) — {detail}")
    except Exception as e:
        _shadow_write(shadow, gen_error=type(e).__name__)
        return f"codegen error: {type(e).__name__}: {e} (Modell {MODEL!r}, Ziel {ENGINE})"
    path = f"{SANDBOX}/job{rid}.lua"
    open(path, "w").write(code + "\n")
    out, rc = run_lua(path)
    retried = 0
    rc_first = rc
    # Roadmap P2 (rc-gated retry): the self-test already told us the truth —
    # retry on OBSERVED failure, not on a prediction. One redraw (the serving
    # unit samples at temp 0.7, so a second draw is a genuinely new attempt;
    # one redraw rescued 4/27 = ~15% of deterministic fails [measured]). Env-gated: without
    # BULLPEN_RETRY_ON_FAIL=1 this block is dead and the path is today's.
    # The retry result only REPLACES the first one when it actually passes.
    served_first = served
    retry_extra = {}
    if rc != 0 and (LADDER or os.environ.get("BULLPEN_RETRY_ON_FAIL") == "1"):
        rest = deadline - time.time()
        if LADDER and rest < 60:
            # Kein Mini-Timeout-Theater: ohne echte Restzeit wird die
            # Sprosse verbucht, nicht simuliert (Kontrakt paragraph 2).
            retry_extra["retry_skipped"] = 1
        else:
            try:
                # B2: der zweite Wurf bekommt nur das halbe Budget, damit der
                # Worst Case unter der Collection bleibt; die Leiter deckelt
                # zusaetzlich mit der Restzeit der Deadline.
                cap = CODEGEN_TIMEOUT // 2
                if LADDER:
                    cap = max(60, min(cap, int(rest)))
                code2, served2 = gen_code(req, timeout=cap)
                path2 = f"{SANDBOX}/job{rid}r.lua"
                open(path2, "w").write(code2 + "\n")
                out2, rc2 = run_lua(path2)
                retried = 1
                retry_extra["retry_served"] = served2
                if rc2 == 0:
                    code, served, path, out, rc = code2, served2, path2, out2, rc2
            except Exception as e:
                # ein gescheiterter Retry-VERSUCH darf die erste Antwort nie
                # kosten — aber unsichtbar sein darf er auch nicht (Review B3)
                retry_extra["retry_error"] = type(e).__name__

    # Sprosse 3 (C-2c-ladder): Geschwister-Zug, nur bei weiter rotem Selbst-
    # test. Fail-open in jede Richtung — ein Geschwister-Fehlschlag kostet
    # nie die vorhandene Antwort, ein leerer Draft zaehlt als rot.
    sib_extra = {}
    sib_rescued = False
    if LADDER and rc != 0 and SIBLING_URL:
        rest = deadline - time.time()
        if rest < 60:
            sib_extra["sibling_skipped"] = 1
        else:
            try:
                import bullpen_ladder
                code3, toks, cut = bullpen_ladder.sibling_cut0(
                    req, SIBLING_URL, min(SIBLING_TIMEOUT, int(rest)))
                if code3.strip():
                    path3 = f"{SANDBOX}/job{rid}s.lua"
                    open(path3, "w").write(code3 + "\n")
                    out3, rc3 = run_lua(path3)
                else:
                    path3, out3, rc3 = None, "", 1
                sib_extra.update(sibling_rc=rc3, sibling_tokens=toks,
                                 sibling_cut_fired=cut)
                if rc3 == 0:
                    # kein URL-Leak in den Raum (Review F11) — die Adresse
                    # steht dem Betreiber in der Unit-Umgebung, nicht jedem
                    # Raumteilnehmer
                    code, served, path, out, rc = (
                        code3, "sibling", path3, out3, rc3)
                    sib_rescued = True
            except Exception as e:
                sib_extra["sibling_error"] = type(e).__name__

    ladder_extra = {}
    if LADDER:
        ladder_extra["rung"] = (1 if rc_first == 0 else
                                3 if sib_rescued else
                                2 if rc == 0 else 0)
    _shadow_write(shadow, served_model=served, served_model_first=served_first,
                  lua_rc=rc, lua_rc_first=rc_first, retried=retried,
                  code_len=len(code), **retry_extra, **sib_extra, **ladder_extra)
    status = "ran OK" if rc == 0 else f"exit {rc}"
    if sib_rescued:
        status += " (Geschwister)"
    elif retried:
        status += " (nach Retry)" if rc == 0 else " (Retry half nicht)"
    if rc != 0 and LADDER:
        runs = (["gen"] + (["retry"] if retried else [])
                + (["sibling"] if "sibling_rc" in sib_extra else []))
        # Stabilitaet behauptet nur, wer mehr als einen Wurf gesehen hat.
        if len(runs) >= 2:
            status += f" — STABILER FEHLSCHLAG (Leiter: {'+'.join(runs)} rot)"
    model_tag = f"asked {MODEL}, served by {served}" if served and served != MODEL else f"model {MODEL}"
    body = (f"wrote {path} ({len(code)} chars, {model_tag}) — {status}. "
            f"output:\n{out[:400]}\n--- code ---\n{code}")

    # A ticket that names a spec gets a certificate, not a claim. Without one
    # the reply is unchanged — the old behaviour is the default, and a caller
    # opts in by naming the test.
    m = SPEC_RE.search(req)
    if m:
        spec = m.group(1)
        urkunde, arc = attest(path, spec)
        if urkunde is None:
            body += ("\n--- ABNAHME ---\nbullpen-attest ist in diesem Container nicht "
                     "vorhanden; ich kann nicht attestieren. Das ist zu melden, nicht zu "
                     "umgehen.")
        else:
            body += f"\n--- ABNAHME (rc {arc}) ---\n{urkunde[:1200]}"
            if "LMCP_PROBE_URL" in urkunde:
                body += ("\n\nHINWEIS ZUR GRENZE: diese Spezifikation verankert sich an einem "
                         "laufenden Server. Ich halte kein Raum-Token — ich fuehre frisch "
                         "erzeugten Code aus, und ein Token in meiner Umgebung waere fuer "
                         "diesen Code lesbar. Ein verankerter Abnahmetest gehoert deshalb zu "
                         "einer Rolle, die Zugangsdaten halten darf. Das ist die Eingrenzung, "
                         "kein Fehlschlag meinerseits.")
    return body

if __name__ == "__main__":
    if len(sys.argv) > 2 and sys.argv[1] == "--once":
        print(dispatch({"body": sys.argv[2], "id": 0}))
    else:
        bw.run("coder", dispatch, online="coder online — writes+runs Lua (sandboxed); attests when the ticket names SPEC:. @coder <task>",
               ack="…coding")
