Upload app_kimi.py

app_kimi.py

@@ -10,6 +10,18 @@ SCOPE: this collects data only. It does NOT train anything -- no DPO loop,
 no optimizer, nothing that touches gradients. Output is training-ready JSONL
 for someone else's training scripts to consume.
 
+WHAT CHANGED IN v10.1 (field report on the live shards: reasoning-tag leak, 5x unclosed thinking, CSS run as Python)
+----------------------------------------------------------------------------------------------------------------
+  * _strip_reasoning_leak() now also removes an ORPHAN close tag (a close with no open) in addition to complete
+    pairs and dangling opens, for both delimiter conventions (<think> and kimi's triangle-glyph tags);
+    _validate_fields() rejects any field still containing the glyph form. (All workers.)
+  * Code fences: only ```python / ```py / ```python3 blocks are executed (an untagged fence used to match, so
+    CSS/HTML/shell was run as Python and failed with a guaranteed SyntaxError); a block fenced as python that is
+    plainly HTML/CSS/shell is skipped for execution; PROPOSE and RECOVERY templates now say other languages must
+    use their own fence tag. (All workers.)
+  * kimi only: KimiThinkBudgetClient -- capped two-stage generation (GIJL_KIMI_THINK_BUDGET, default 1200), the
+    same cure MinimaxThinkBudgetClient applies to minimax, adapted to kimi writing its own opening tag.
+
 WHAT CHANGED IN v10 (code-enforced general agent-thinker trajectories, single-file-per-worker storage,
 anonymization, Dataset Viewer fix)
 ----------------------------------------------------------------------------------------------------------------
@@ -2041,6 +2053,161 @@ class LlamaCppClient:
         return ok
 
 
+_KIMI_THINK_OPEN = "\u25c1think\u25b7"
+_KIMI_THINK_CLOSE = "\u25c1/think\u25b7"
+_KIMI_THINK_BUDGET = int(os.environ.get("GIJL_KIMI_THINK_BUDGET", "1200"))
+
+
+def _strip_special_keep_think(tok, text: str, keep_atem: bool) -> str:
+    """Like _strip_special_tokens, but never deletes a think-delimiter token: if kimi's glyph tags are
+    registered as special tokens, the stock stripper would remove them and the stage logic (and the later
+    stripper) could no longer see where reasoning starts and ends."""
+    for t in set(getattr(tok, "all_special_tokens", []) or []):
+        if not t:
+            continue
+        low = t.lower()
+        if "think" in low or (keep_atem and "atem" in low):
+            continue
+        text = text.replace(t, "")
+    return text
+
+
+class KimiThinkBudgetClient(LlamaCppClient):
+    """v10.1, this worker only -- field evidence from the live shard: kimi writes its own reasoning block
+    delimited by the triangle-glyph think tags (29 opens vs 6 closes in one file), i.e. in roughly four of five
+    turns it never closes the block inside the token budget, so the visible answer is empty or cut mid-thought.
+    Same disease as minimax's (see MinimaxThinkBudgetClient), same cure: generation is split into two
+    /completion calls against the same slot. Stage 1 runs with the close tag as a stop string and a capped
+    budget (GIJL_KIMI_THINK_BUDGET, default 1200 tokens). If it stops on the close tag, or hits the cap while a
+    block is open, the close tag is written (forced in the second case) and stage 2 continues from there with
+    the worker's full max_new_tokens for the answer / tool call; the second call reuses the first one's KV
+    cache (exact prompt-prefix match). If stage 1 ends on its own (no reasoning block at all), or hits the cap
+    while writing a plain long answer, nothing is forced: the first case is returned as is, the second simply
+    continues. Unlike minimax, no opening tag is added back: kimi emits its own, so the combined text is
+    byte-for-byte what a single uncapped call would have produced. Best-effort: raise
+    GIJL_KIMI_THINK_BUDGET if the [llama][kimi] forced-close line shows up on most turns."""
+
+    def chat(self, messages, max_new_tokens=GEN_MAX_NEW_TOKENS, temperature=0.4, tools=None):
+        import requests
+        tool_mode = bool(tools)
+        self.server.ensure_running()
+        base_prompt = self._prepare_prompt(_render_chat(self.tok, messages, tools))
+        try:
+            n_prompt = len(self._tokenize(base_prompt, True))
+        except requests.RequestException:
+            n_prompt = -1
+        if n_prompt >= 0 and n_prompt + 256 > self.ctx:
+            print(f"[llama][kimi] prompt is {n_prompt} tokens with ctx={self.ctx} -- refusing to generate", flush=True)
+            return {"content": "", "error": "context_too_long", "timed_out": False, "prompt_tokens": n_prompt}
+
+        think_budget = _KIMI_THINK_BUDGET
+        if n_prompt >= 0:
+            think_budget = max(256, min(think_budget, self.ctx - n_prompt - max_new_tokens - 64))
+        s1 = self._raw_completion(base_prompt, think_budget, temperature, [_KIMI_THINK_CLOSE], tool_mode)
+
+        def _result(content, s2=None):
+            s2 = s2 or {"cached": 0, "gen": 0, "error": None, "timed_out": False}
+            return {"content": content, "error": s1["error"] or s2["error"],
+                    "timed_out": s1["timed_out"] or s2["timed_out"], "prompt_tokens": n_prompt,
+                    "cached_tokens": s1["cached"] + s2["cached"], "gen_tokens": s1["gen"] + s2["gen"]}
+
+        body = s1["content"]
+        if s1["error"] or s1["timed_out"]:
+            return _result(body)
+        stop_type = s1.get("stop_type")
+        if stop_type == "word" or _KIMI_THINK_CLOSE in body:
+            # the model closed its own reasoning; the stop string itself is not part of the streamed text
+            if _KIMI_THINK_CLOSE not in body:
+                body = body + _KIMI_THINK_CLOSE
+            body = body.rstrip() + "\n"
+        elif stop_type == "limit" and _KIMI_THINK_OPEN in body:
+            body = body.rstrip() + "\n" + _KIMI_THINK_CLOSE + "\n"
+            print(f"[llama][kimi] reasoning hit the {think_budget}-token think-budget before closing itself "
+                  f"-- forced the close tag and continuing to the answer stage", flush=True)
+        elif stop_type != "limit":
+            return _result(body)        # finished on its own (EOS / other): a complete answer, nothing to continue
+        # else: cap reached while writing a plain (non-reasoning) answer -> just continue it in stage 2
+
+        s2 = self._raw_completion(base_prompt + body, max(64, max_new_tokens), temperature, None, tool_mode)
+        return _result(body + s2["content"], s2)
+
+    def _raw_completion(self, prompt, n_predict, temperature, stop, tool_mode):
+        """One /completion call -- same streaming loop, timeouts and SSE parsing as LlamaCppClient.chat();
+        additionally reports llama-server's stop_type ('word' = a stop string matched, 'limit' = n_predict
+        reached, 'eos' = the model finished)."""
+        import requests
+        payload = {"prompt": prompt, "n_predict": n_predict, "temperature": temperature, "top_p": TOP_P, "top_k": TOP_K,
+                   "stream": True, "cache_prompt": True}
+        if stop:
+            payload["stop"] = stop
+        payload.update(self.extra_payload)
+
+        gap = GEN_PER_CHUNK_GAP_TIMEOUT_S.get(self.kind, 300)
+        cap = GEN_MAX_TOTAL_S.get(self.kind, 900)
+        chunks, final, err, timed_out, resp = [], None, None, False, None
+        start = last_data = last_hb = time.time()
+        try:
+            resp = requests.post(self.server.base_url + "/completion", json=payload, stream=True, timeout=(15, gap))
+            if resp.status_code != 200:
+                body = resp.text[:400]
+                err = "context_too_long" if "exceed_context_size" in body else f"HTTP {resp.status_code}: {_redact(body)}"
+            else:
+                for raw in resp.iter_lines(chunk_size=1):
+                    now = time.time()
+                    if now - start > cap:
+                        print(f"[gen][TIMEOUT][kimi] hit the {cap}s hard cap after {len(chunks)} chunks -- cancelling generation.", flush=True)
+                        timed_out = True
+                        break
+                    if now - last_data > gap:
+                        print(f"[gen][TIMEOUT][kimi] no generated text for {gap}s -- cancelling generation.", flush=True)
+                        timed_out = True
+                        break
+                    line = raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw
+                    if not line.startswith("data:"):
+                        continue
+                    try:
+                        d = json.loads(line[5:].strip())
+                    except ValueError:
+                        continue
+                    if "error" in d:
+                        e = d["error"]
+                        err = _redact(str(e.get("message", e)) if isinstance(e, dict) else str(e))
+                        break
+                    last_data = now
+                    if d.get("content"):
+                        chunks.append(d["content"])
+                    if d.get("stop"):
+                        final = d
+                        break
+                    if now - last_hb > 30:
+                        print(f"[gen][kimi] {sum(len(c) for c in chunks)} chars, {now - start:.0f}s elapsed ({self.kind})", flush=True)
+                        last_hb = now
+        except requests.RequestException as e:
+            if isinstance(e, requests.exceptions.Timeout) or "timed out" in str(e).lower():
+                print(f"[gen][TIMEOUT][kimi] llama-server sent nothing for {gap}s -- treating as stuck.", flush=True)
+                timed_out = True
+            else:
+                err = f"{type(e).__name__}: {_redact(str(e))[:200]}"
+        finally:
+            if resp is not None:
+                resp.close()
+
+        cached = gen = 0
+        stop_type = None
+        if final:
+            t = final.get("timings") or {}
+            cached, gen = int(t.get("cache_n") or 0), int(final.get("tokens_predicted") or 0)
+            stop_type = final.get("stop_type")
+            print(f"[llama][kimi] prompt {final.get('tokens_evaluated')} tok (cached {cached}) prefill "
+                  f"{t.get('prompt_per_second') or 0:.0f} tok/s | generated {gen} tok @ "
+                  f"{t.get('predicted_per_second') or 0:.1f} tok/s | stop={stop_type}", flush=True)
+        if err and err != "context_too_long":
+            print(f"[llama][kimi] request failed: {err}", flush=True)
+        text = _strip_special_keep_think(self.tok, "".join(chunks), keep_atem=tool_mode)
+        return {"content": text, "error": err, "timed_out": timed_out, "cached": cached, "gen": gen,
+                "stop_type": stop_type}
+
+
 def start_llama_backend():
     """-> (client, server). Model download, binary build/download and tokenizer load run in parallel."""
     from concurrent.futures import ThreadPoolExecutor
@@ -2062,7 +2229,7 @@ def start_llama_backend():
     if kind == "gpu" and not re.search(r"offloaded|CUDA\d", server.tail(600)):
         print("[llama][WARNING] the server log does not mention CUDA/offloading -- it may be running on CPU. "
               f"Check {server.log_path}", flush=True)
-    client = LlamaCppClient(server, tok, kind)
+    client = KimiThinkBudgetClient(server, tok, kind)
     print(f"[llama] context per request: {client.ctx} tokens | server adds BOS: {client.server_adds_bos}", flush=True)
     if not client.warmup():
         print("[llama][WARNING] warmup produced no text; continuing, but check llama-server.log", flush=True)
@@ -2100,18 +2267,20 @@ def _strip_atem_noise(text: str) -> str:
     return _channel_noise_re().sub(" ", text or "").strip()
 
 
-_THINK_BLOCK_RE = re.compile(r"<think>.*?</think>", re.DOTALL)
-_THINK_DANGLING_RE = re.compile(r"<think>.*\Z", re.DOTALL)   # closing tag never arrived (token cap / timeout)
+_THINK_OPEN_RE = r"(?:<think>|\u25c1think\u25b7)"
+_THINK_CLOSE_RE = r"(?:</think>|\u25c1/think\u25b7)"
+_THINK_BLOCK_RE = re.compile(_THINK_OPEN_RE + r".*?" + _THINK_CLOSE_RE, re.DOTALL)
+_THINK_ORPHAN_CLOSE_RE = re.compile(r"\A.*" + _THINK_CLOSE_RE, re.DOTALL)    # a close with no open: everything before it was reasoning
+_THINK_DANGLING_RE = re.compile(_THINK_OPEN_RE + r".*\Z", re.DOTALL)         # closing tag never arrived (token cap / timeout)
 
 
 def _strip_reasoning_leak(text: str) -> str:
-    """v8: this model's OWN chat template (not this harness) defines a native <think>...</think> convention --
-    see WHAT CHANGED IN v8 for exactly which models do and don't, confirmed per-model via each one's own
-    chat_template.jinja/tokenizer_config.json rather than assumed. Applied at the same call sites
-    _strip_atem_noise already is, so no stored field or trajectory turn keeps one model's thinking-tag residue
-    while another's stays clean -- a no-op for a model with no thinking mode, defense-in-depth for one whose
-    thinking is suppressed by _CHAT_TEMPLATE_EXTRA_KWARGS below, the only backstop for one with no off-switch."""
+    """v8/v10.1: strips a model's own native reasoning block in whichever delimiter convention its chat
+    template uses (ASCII think tags, or kimi's triangle-glyph variant -- confirmed from live output, not from
+    a model card). Order matters: complete pairs first, then an orphan close (reasoning whose opening tag was
+    baked into the prompt or cut off), then a dangling open (reasoning that never closed)."""
     text = _THINK_BLOCK_RE.sub(" ", text or "")
+    text = _THINK_ORPHAN_CLOSE_RE.sub(" ", text)
     return _THINK_DANGLING_RE.sub(" ", text).strip()
 
 
@@ -3015,7 +3184,8 @@ def _validate_fields(fields: dict, language: str, limits: dict):
         if "<atem:" in val:
             return f"{name} contains tool-protocol markup"
         if ("<think" in val or "</think>" in val or "<tool_call" in val or "<minimax:tool_call" in val
-                or "\uff5cDSML\uff5c" in val):
+                or "\uff5cDSML\uff5c" in val
+                or "\u25c1think\u25b7" in val or "\u25c1/think\u25b7" in val):
             return f"{name} contains a model-native reasoning/tool-call leak"   # v8: backstop, see WHAT CHANGED IN v8
         if len(val) < lo:
             return f"{name} too short ({len(val)}<{lo} chars)"
@@ -3381,7 +3551,7 @@ _TASK_TAG_RE = re.compile(r"<task>(.*?)</task>", re.DOTALL | re.IGNORECASE)
 _NEEDS_TOOLS_RE = re.compile(r"NEEDS_TOOLS\s*:\s*(yes|no)", re.IGNORECASE)
 _REVISION_RE = re.compile(r"REVISION\s*:\s*(.*)", re.DOTALL | re.IGNORECASE)
 _RATIONALE_RE = re.compile(r"RATIONALE:\s*(.*)")
-_PY_CODE_BLOCK_RE = re.compile(r"```(?:python|py)?\s*\n(.*?)```", re.DOTALL)
+_PY_CODE_BLOCK_RE = re.compile(r"```(?:python3|python|py)[ \t]*\n(.*?)```", re.DOTALL)   # v10.1: the tag is REQUIRED -- an untagged/other-language fence is never executed as Python
 
 EXECUTE_CODE_TIMEOUT_S = int(os.environ.get("GIJL_EXECUTE_CODE_TIMEOUT_S", "20"))
 EXECUTE_CODE_MAX_OUTPUT_CHARS = 4000
@@ -3454,6 +3624,18 @@ def _grounding_check(proposal: str, research_steps: list) -> dict:
     return {"passed": passed, "overlap_ratio": round(overlap, 3), "reason": reason}
 
 
+_NON_PYTHON_FIRST_LINE_RE = re.compile(r"^\s*(?:<[!/?A-Za-z]|#!/|\$ |@(?:media|import|charset)\b)")
+_CSS_RULE_RE = re.compile(r"^[^\n=()]{1,120}\{\s*[\w-]+\s*:[^}]*;", re.DOTALL)
+
+
+def _block_is_non_python(code: str) -> bool:
+    """v10.1: a block the model fenced as python but that is plainly HTML / CSS / shell. Executing it as Python
+    can only fail, which then poisoned the verification result (live case: a CSS flexbox answer). Such a block
+    is simply not a candidate for code execution; verification falls through to the grounding check."""
+    c = (code or "").strip()
+    return bool(_NON_PYTHON_FIRST_LINE_RE.match(c) or _CSS_RULE_RE.match(c))
+
+
 def _code_syntax_ok(text: str):
     """Cheap, real, non-LLM pre-check (DeepMind's synthetic-data finding: verify with code execution where
     you can). A syntax check only -- used as a fast path so an unparseable block never reaches the heavier
@@ -3532,9 +3714,11 @@ Your own plan: {think}
 
 {research_note}
 
-If your answer includes code, put it in a single fenced ```python code block -- it WILL be executed for real
-to verify it before anything else happens, so make it runnable as-is (no placeholders, no "...", include any
-needed imports), not just a sketch. If it makes factual/technical claims, make sure they are grounded in what
+If your answer includes a complete Python program or script, put it in a single fenced ```python block --
+ONLY ```python blocks are executed for real to verify them, so make that block runnable as-is (no placeholders,
+no "...", include any needed imports), not just a sketch. Code in ANY OTHER language or format (CSS, HTML,
+JavaScript, shell, SQL, YAML/config, plain snippets) must use its own fence tag (```css, ```html, ```bash, ...)
+and must NEVER be labeled ```python -- it is not executed, it is simply part of the answer. If it makes factual/technical claims, make sure they are grounded in what
 you found above, not guessed. Do not pad the answer with meta-commentary about your process -- just the
 answer a real person would receive.
 
@@ -3571,7 +3755,8 @@ RECOVERY_FIX_TEMPLATE = """Based on what you just found, write ONE consolidated
 not conflict with your original approach -- reconcile the candidate fixes you found into a single coherent
 answer rather than patching in every suggestion. Before the fix itself, write one line starting "RATIONALE:"
 explaining briefly why this is the right fix given what you found. Then give the complete, corrected final
-answer (same format rules as before: a single runnable ```python block if there is code, no placeholders).
+answer (same format rules as before: Python goes in one runnable ```python block, any other language in its
+own fence tag and never ```python, no placeholders).
 
 <task>{task_prompt}</task>
 
@@ -3600,9 +3785,10 @@ def _run_verification(client, proposal: str, research_steps: list, task_prompt:
     """Harness-selected, not model-selected: picks the verification method from what `proposal` actually
     contains, runs it for real, and returns {"method", "passed", "evidence"}. This is what makes
     "verify everything you produce" generalize across arbitrary task kinds instead of one fixed routine."""
-    code_blocks = _PY_CODE_BLOCK_RE.findall(proposal)
+    code_blocks = [b for b in _PY_CODE_BLOCK_RE.findall(proposal) if not _block_is_non_python(b)]
     if code_blocks:
-        if _code_syntax_ok(proposal) is False:
+        import ast as _ast
+        if not all(_safe_ast_parse(_ast, b) for b in code_blocks):
             return {"method": "code_execution", "passed": False,
                     "evidence": {"stderr": "SyntaxError: the proposed code does not even parse", "stdout": "",
                                  "exit_code": None, "timed_out": False}}