AILang/experiments/2026-05-12-cross-model-authoring/qwen_agentic.py

#!/usr/bin/env python3
"""Agentic oracle loop — can the model DRIVE the compiler, the way Claude Code does?

The single-shot ablation asks "can the model write it blind?". This asks the
fairer question: given the compiler as a tool, can the model ITERATE to green?
The model never touches `ail` directly (it only emits text); this harness runs
`ail check`/`ail run` after each attempt and feeds the RAW, unedited compiler
output back into the dialogue, up to MAX_TURNS. No hints, no hand-holding — the
only new information each turn is what the compiler actually said. That is
exactly the loop Claude Code runs.

Plain Form-A surface (no converter, so no converter risk: the model's text goes
straight to `ail`). Same SERIES few-shot + SMA task as qwen_ablation.py L7.

Reads $IONOS_API_TOKEN (never printed). Live IONOS — run with consent.
"""
from __future__ import annotations
import json, os, re, subprocess, urllib.request
from pathlib import Path

ENDPOINT = "https://openai.inference.de-txl.ionos.com/v1/chat/completions"
MODEL    = os.environ.get("PROBE_MODEL", "Qwen/Qwen3-Coder-Next")
TAG      = os.environ.get("PROBE_TAG", "qwen")
MAX_TURNS = int(os.environ.get("PROBE_TURNS", "8"))
REPO = Path(__file__).resolve().parents[2]
AIL  = os.environ.get("AIL_BIN", str(REPO / "target/debug/ail"))
DOC  = REPO / f"experiments/2026-05-12-cross-model-authoring/{TAG}-agentic-sma.md"

DEMO_A = """(module demo_a
  (fn double
    (type (fn-type (params (own (con Int))) (ret (own (con Int)))))
    (params x)
    (body (app + x x)))
  (fn main
    (type (fn-type (params) (ret (own (con Unit))) (effects IO)))
    (params)
    (body (seq (app print (app double 21)) (do io/print_str "\\n")))))"""

DEMO_B = """(module demo_b
  (data IntList (ctor Nil) (ctor Cons (con Int) (con IntList)))
  (fn sum
    (type (fn-type (params (own (con IntList))) (ret (own (con Int)))))
    (params xs)
    (body
      (match xs
        (case (pat-ctor Nil) 0)
        (case (pat-ctor Cons h t) (app + h (app sum t))))))
  (fn main
    (type (fn-type (params) (ret (own (con Unit))) (effects IO)))
    (params)
    (body
      (let xs (term-ctor IntList Cons 1 (term-ctor IntList Cons 2 (term-ctor IntList Nil)))
        (seq (app print (app sum xs)) (do io/print_str "\\n"))))))"""

DEMO_S = """(module demo_s
  (fn main
    (type (fn-type (params) (ret (own (con Unit))) (effects IO)))
    (params)
    (body
      (let s (new Series (con Float) 3)
        (let s2 (app Series.push s 1.0)
          (let s3 (app Series.push s2 2.0)
            (seq (app print (app Series.len s3)) (do io/print_str "\\n"))))))))"""

INTRO = ("AILang is a fully-parenthesised S-expression language (\"Form-A\"). "
         "Study these complete, valid example modules, then write the requested "
         "module. Return ONLY the AILang source for the module — no prose, no "
         "markdown fences.\n\n")
BASE   = INTRO + "Example 1:\n" + DEMO_A + "\n\nExample 2:\n" + DEMO_B + "\n"
SERIES = (BASE + "\nExample 3 (uses the Series ring-buffer library — "
          "`(new Series (con Float) N)` allocates one of capacity N; "
          "`Series.push : (own (Series a)) (own a) -> (own (Series a))` appends "
          "and returns the new series; `Series.at : (borrow (Series a)) (own Int) "
          "-> (own a)` reads index i back from newest (0 = newest); "
          "`Series.len`/`Series.total_count : (borrow (Series a)) -> (own Int)`):\n"
          + DEMO_S + "\n")

SMA_OUT = "3.0\n5.33333\n5.66667\n5.33333\n"
SMA_TASK = ("Write a module `m` computing a simple moving average, window 3, over "
            "the float stream 1.0, 5.0, 3.0, 8.0, 6.0, 2.0 using the Series library: "
            "create a Float Series lookback 3, push each value in turn, and after each "
            "push once at least 3 values have been pushed, print the average of the "
            "most recent 3 (their sum / 3.0) on its own line. Expected:\n" + SMA_OUT +
            "\nYou will be given the compiler's exact output after each attempt and "
            "may correct your module. Return ONLY the module source each turn.")

def call(messages):
    body = json.dumps({"model": MODEL, "temperature": 0.2, "max_tokens": 1500,
        "messages": messages}).encode()
    req = urllib.request.Request(ENDPOINT, data=body, headers={
        "Authorization": f"Bearer {os.environ['IONOS_API_TOKEN']}", "Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=300) as r:
        d = json.loads(r.read())
    return d["choices"][0]["message"]["content"], d.get("usage", {})

def extract(resp):
    m = re.search(r"```(?:[a-zA-Z]*)?\n(.*?)```", resp, re.S)
    code = m.group(1) if m else resp
    i = code.find("(module")
    if i < 0: return code.strip()
    depth = 0
    for j in range(i, len(code)):
        if code[j] == "(": depth += 1
        elif code[j] == ")":
            depth -= 1
            if depth == 0: return code[i:j+1]
    return code[i:].strip()

def evaluate(code, expected):
    """Returns (ok, raw_feedback). raw_feedback is the compiler's exact text."""
    m = re.search(r"\(module\s+([A-Za-z_][A-Za-z0-9_]*)", code)
    name = m.group(1) if m else "m"
    d = Path("/tmp/agentic"); d.mkdir(exist_ok=True)
    f = d / f"{name}.ail"; f.write_text(code + "\n")
    chk = subprocess.run([AIL, "check", str(f)], capture_output=True, text=True)
    if chk.returncode != 0:
        return False, "`ail check` failed:\n" + (chk.stderr or chk.stdout).strip()
    run = subprocess.run([AIL, "run", str(f)], capture_output=True, text=True)
    if run.returncode != 0:
        return False, "`ail check` passed; `ail run` failed:\n" + (run.stderr or run.stdout).strip()
    if run.stdout != expected:
        return False, (f"`ail check` and `ail run` succeeded but the output is wrong.\n"
                       f"Your program printed:\n{run.stdout!r}\nExpected:\n{expected!r}")
    return True, run.stdout

def main():
    messages = [{"role": "system", "content": SERIES},
                {"role": "user", "content": SMA_TASK}]
    log = [f"# Agentic oracle loop — can {MODEL} drive the compiler to a green SMA?\n",
           f"**Model:** {MODEL}; multi-turn (≤{MAX_TURNS} turns); raw `ail` output fed "
           "back each turn; plain Form-A, no converter.\n", "---\n"]
    solved_at = None
    for turn in range(1, MAX_TURNS + 1):
        try:
            resp, usage = call(messages)
        except Exception as e:
            log += [f"## Turn {turn} — ❌ api-error\n", f"```\n{str(e)[:300]}\n```\n"]; break
        code = extract(resp)
        ok, fb = evaluate(code, SMA_OUT)
        mark = "✅ GREEN" if ok else "❌ not yet"
        print(f"turn {turn}: {mark}  (completion={usage.get('completion_tokens','?')})")
        log += [f"## Turn {turn} — {mark}",
                f"\ntokens: prompt={usage.get('prompt_tokens','?')} completion={usage.get('completion_tokens','?')}\n",
                "model output (extracted module):\n```\n" + code + "\n```\n",
                "compiler feedback returned to the model:\n```\n" + fb[:1500] + "\n```\n"]
        if ok:
            solved_at = turn; break
        # feed the raw compiler output back, exactly as Claude Code's loop would
        messages.append({"role": "assistant", "content": resp})
        messages.append({"role": "user", "content":
            fb + "\n\nFix the module so it compiles and prints exactly the expected "
            "output. Return ONLY the corrected module source."})
    verdict = (f"**Solved at turn {solved_at}/{MAX_TURNS}.**" if solved_at
               else f"**Not solved within {MAX_TURNS} turns.**")
    print(verdict)
    log += ["\n---\n## Verdict\n", verdict + "\n"]
    DOC.write_text("\n".join(log)); print("doc →", DOC)

if __name__ == "__main__":
    main()