code/finetune/common_ft.py

"""Shared: build the exact jev-ultrafast systemone request for a page + goal, with the server-side compaction."""
import importlib.util, json, sys, types

JEV = "/home/ckl/projects/S/jev-ultrafast/jev_ultrafast"
# load model.py / questions.py without running the package __init__ (which pulls in browser-harness)
if "jev_ultrafast" not in sys.modules:
    _pkg = types.ModuleType("jev_ultrafast"); _pkg.__path__ = [JEV]; sys.modules["jev_ultrafast"] = _pkg
    for _name in ("questions", "model"):
        _spec = importlib.util.spec_from_file_location(f"jev_ultrafast.{_name}", f"{JEV}/{_name}.py")
        _m = importlib.util.module_from_spec(_spec); sys.modules[_spec.name] = _m; _spec.loader.exec_module(_m)
from jev_ultrafast.model import action_space  # noqa: E402
from jev_ultrafast.questions import GOAL_DONE, NEXT_ACTION, TARGET  # noqa: E402

import os
import re
FMT = os.environ.get("LAYA_FMT", "v1")
# v1: jev's state verbatim (page text up to 6000 chars + the whole element table as JSON) -- the 1024-token budget truncates
#     most of it, so the model often never sees the candidates' context.  3000 chars was tried (v7): -0.04 top-1.
# v2: elements live only in the option list (full label + role + value); state keeps title/url/history and 1500 chars of text.
# v3: v2 + option labels capped at 50 chars and 1200 chars of text (~30% fewer tokens; for the 322M base to hit ~20 ms/step)
# v4: v3 + no duplicate "[key] " before each option label, and <select> options rendered as just "Field → Option"
# v5: v4 + "fields" (every form field with its current value) first in the state
# v6: v5 + longer context -- the last 20 actions, each with its RESULT (the page it led to: URL path/query + title), placed
#     before the page text so they are never truncated; 3000 chars of page text; trained / served at max_len 2048
PAGE_TEXT_CHARS = {"v2": 1500, "v3": 1200, "v4": 1200, "v5": 1200, "v6": 3000}.get(FMT, 6000)
LABEL_CHARS = 50 if FMT in ("v3", "v4", "v5", "v6") else 10000

LABELS = {
    "CLICK": "Click an element, button, menu option, autocomplete suggestion, or calendar day.",
    "TYPE_TEXT": "Enter or replace text in an editable field. A small LLM will supply the value from the goal.",
    "SELECT": "Select an observed dropdown value.",
}


def _cut(el, n):
    """Truncate an element label to n chars; for a <select> option ("[i:k] Field label → Option") keep the option name."""
    el = str(el)
    if len(el) <= n:
        return el
    if " → " in el:
        head, opt = el.rsplit(" → ", 1)
        opt = " ".join(opt.split())[:40]
        keep = max(12, n - len(opt) - 3)
        return " ".join(head.split())[:keep] + " → " + opt
    return el[:n]


def compact(v):
    if isinstance(v, dict) and "element" in v:
        el = str(v["element"])
        if FMT in ("v4", "v5", "v6"):
            # v4: the option key is already rendered by laya ("<key>: ..."), so drop the duplicate "[key] "; a <select>
            # option is only "Field → Option" (role and current value repeated on every option ate the head budget)
            el = re.sub(r"^\[[^\]]*\]\s*", "", el)
            if " → " in el:
                return _cut(el, 50)
        s = _cut(el, LABEL_CHARS)
        if v.get("role"):
            s += f" ({v['role']})"
        if v.get("current_value"):
            s += f" = {str(v['current_value'])[:30]!r}"
        for k in ("checked", "selected", "expanded"):
            if k in v:
                s += f" {k}={v[k]}"
        return s
    return v


def fields_summary(elements):
    """v5: the form's fields and their CURRENT values, first in the state, so the policy sees what is still empty
    (v4's option lists no longer repeat a dropdown's current value).  Same code in apps/systemone_server.py."""
    out = []
    for e in elements or []:
        ops, role = e.get("operations") or [], e.get("role")
        if "TYPE_TEXT" in ops or "SELECT" in ops or role == "combobox":
            v = str(e.get("value") or "").strip()
            out.append(f"{str(e.get('label', ''))[:40]} = {v[:30]!r}" if v else f"{str(e.get('label', ''))[:40]} = (empty)")
        elif role in ("checkbox", "radio", "switch") and "checked" in e:
            out.append(f"{str(e.get('label', ''))[:40]}: checked={e['checked']}")
        if len(out) >= 14:
            break
    return "; ".join(out)


def goal_done_question(goal):
    """The yes/no (noul) completion check asked next to the operation question (same text in jev_ultrafast.model)."""
    return {"type": "noul", "instructions": {"goal": goal, "statement": GOAL_DONE}}



def short_url(u):
    """path + query of a URL (the host is in the page's own url), capped -- what an action led to."""
    u = re.sub(r"^https?://[^/]+", "", str(u or ""))
    return u[:90] or "/"


def history_v6(history):
    """v6 history: the last 20 actions, each with its result (URL path/query and title of the page it led to).
    Same code in apps/systemone_server.py and jev_ultrafast.model."""
    out = []
    for h in list(history)[-20:]:
        e = {"action": str(h.get("action", ""))[:60], "kind": h.get("kind")}
        if h.get("text"):
            e["text"] = str(h["text"])[:40]
        if h.get("url") or h.get("title"):
            e["result"] = (short_url(h.get("url")) + " | " + str(h.get("title") or "")[:40]).strip(" |")
        elif h.get("page_changed") is False:
            e["result"] = "no change"
        out.append(e)
    return out


def build_request(page, goal, history=()):
    """Mirror of jev_ultrafast.model.choose() up to the HTTP call. Returns (state, questions, targets, controls)."""
    elements, targets, controls = action_space(page["actions"])
    operations = {key: LABELS[key] for key in targets}
    operations.update({key: value["label"] for key, value in controls.items()})
    operations.update(DONE="Every requirement is visibly satisfied.", BLOCKED="No supported operation can progress.")
    questions = {"operation": {"type": "choice", "criteria": operations, "instructions": {"goal": goal, "rules": NEXT_ACTION}}}
    for operation, candidates in targets.items():
        questions[operation.lower() + "_target"] = {
            "type": "choice",
            "criteria": {index: {"element": f"[{index}] {a['label']}", "current_value": a.get("current_value", a.get("value", "")),
                                 **{k: a[k] for k in ("role", "checked", "selected", "expanded") if k in a}} for index, a in candidates.items()},
            "instructions": {"goal": goal, "operation": operation, "rules": [NEXT_ACTION, TARGET]},
        }
    state = {"page": {"url": page["url"], "title": page["title"], "text": page["text"][:PAGE_TEXT_CHARS]},
             "recent_actions": [{k: h.get(k) for k in ("action", "kind", "text", "page_changed")} for h in list(history)[-10:]]}
    if FMT == "v5":
        state = {"fields": fields_summary(elements), **state}
    if FMT == "v6":
        state = {"fields": fields_summary(elements), "recent_actions": history_v6(history), "page": state["page"]}
    if FMT not in ("v2", "v3", "v4", "v5", "v6"):
        state["elements"] = elements
    for q in questions.values():
        q["criteria"] = {k: compact(v) for k, v in q["criteria"].items()}
    return state, questions, targets, controls


def gold_for(case, targets, controls):
    """(gold operation key, gold target index or None) for a case in the operation/target question vocab."""
    op = case["gold_op"]
    if op == "DONE":
        return "DONE", None
    if op in ("SCROLL_DOWN", "SCROLL_UP", "WAIT", "PRESS_ENTER"):   # page-level controls: operation question only
        return (op, None) if op in {k.upper() for k in controls} else (None, None)
    for index, a in targets.get(op, {}).items():
        if a["id"] == case["gold_id"]:
            return op, index
    return None, None