243 lines
12 KiB
Python
243 lines
12 KiB
Python
"""Task generator: a cloud model writes a task bundle; the harness validates it (oracle = 100, null = 0).
|
|
|
|
python3 -m harness.generator --id G0001 --pool eval --object-type CLAS --category C --difficulty 2
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import time
|
|
import urllib.request
|
|
|
|
from .adt_client import load_env
|
|
from .agents import NullAgent, OracleAgent
|
|
from .ledger import add_usage, check_budget
|
|
from .runner import Runner
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
EXAMPLE_FOR = {"CLAS": "T01", "INTF": "T01", "FUNC": "T13", "PROG": "T14", "DDLS": "T15", "TABL": "T15"}
|
|
CATEGORIES = {
|
|
"A": "pure logic in a new class (language, OO design, Clean ABAP)",
|
|
"B": "database access (ABAP SQL, CDS) with test doubles",
|
|
"C": "internal table craft (table kind, keys, access path)",
|
|
"D": "exception design (class-based exceptions, messages)",
|
|
"E": "refactoring legacy code (seed contains legacy code; behavior must stay the same)",
|
|
"F": "use an unknown object: the spec names a seed object but not its signature; the model must read it",
|
|
"G": "old release target (v702 or v740sp05): no newer syntax",
|
|
"I": "bug fix: the seed contains a faulty object; the spec describes the symptom",
|
|
}
|
|
|
|
SYSTEM = """You write evaluation tasks for an ABAP developer model. Each task is a bundle of files.
|
|
The model gets only spec.md and works on an SAP ABAP Platform 2025 system (SAP_BASIS 816, client 001)
|
|
through ADT tools. The harness installs the seed objects, runs the model, then checks the result with
|
|
hidden ABAP Unit tests, ATC, and abaplint.
|
|
|
|
Rules for the bundle:
|
|
- Use the placeholder {{P}} (upper case) and {{p}} (lower case) at the start of EVERY object name.
|
|
The harness replaces it with a run prefix of 9 characters (for example Z005P001_).
|
|
Name length after replacement: classes, programs, function modules, CDS entities max 30;
|
|
database tables max 16; function groups max 26.
|
|
- All objects are in package $TMP. Do not use transports.
|
|
- Do not use SAP application module data (no FI, SD, MM tables). Use generic business domains and
|
|
only objects that exist in every ABAP Platform system (language, ABAP SQL, CDS, CL_ABAP_*, CL_SALV_TABLE,
|
|
CL_OSQL_TEST_ENVIRONMENT, CL_CDS_TEST_ENVIRONMENT). Seed your own tables and data if needed.
|
|
- No dynpro (CALL SCREEN), no SmartForms, no BAdI, no RAP behavior definitions.
|
|
- spec.md uses Simplified Technical English and these sections in this order:
|
|
1. Goal, 2. Open questions (write "None."), 3. Context, 4. Contract, 5. Business rules,
|
|
6. Constraints (release target, coding standards, out of scope), 7. Acceptance.
|
|
The Contract fixes every public name the hidden tests use: object names, method signatures,
|
|
function module parameters, report parameters, ALV column names, CDS element names.
|
|
The Business rules are complete and unambiguous. Every rule is checked by at least one hidden test.
|
|
Do not tell the model HOW to implement (no table kinds, no SQL). Craft decisions belong to the model.
|
|
- Seed objects: TABL as DDL source ("define table ..."), FUGR without source, FUNC with "functionGroup"
|
|
and full source including the signature in the FUNCTION statement, classes with full source.
|
|
Seed table data: a seed class that implements IF_OO_ADT_CLASSRUN, with "run": true.
|
|
- Hidden tests: one global class, "FOR TESTING DURATION SHORT RISK LEVEL HARMLESS", 5 to 12 test
|
|
methods. Use only the public contract. Function modules: CALL FUNCTION with EXCEPTIONS.
|
|
Reports: SUBMIT ... AND RETURN with cl_salv_bs_runtime_info=>set( display = abap_false
|
|
metadata = abap_false data = abap_true ) and get_data_ref. CDS: cl_cds_test_environment.
|
|
- Reference solution: correct, Clean ABAP, methods below 40 statements, passes all hidden tests,
|
|
no ATC priority 1 or 2 findings (for example: pass large parameters by reference).
|
|
Include the model's expected own tests: for classes a "testclasses_file" (local test classes);
|
|
for reports local test classes inside the program; for function modules and CDS a global test class.
|
|
- task.json keys: id, category, object_type, difficulty, release_target, expected_outcome ("implement"),
|
|
budget {max_tool_calls, max_activations}, seed[], contract[], out_of_scope[], hidden_tests[],
|
|
reference[], craft_checks[]. Contract entries: CLAS {"implements"} optional; FUNC {"functionGroup",
|
|
"params":[{"name","type"}]}; PROG {"parameters":[...]}; DDLS {"fields":[...]}.
|
|
|
|
Return ONLY one JSON object, no markdown fence:
|
|
{"task": <task.json object>, "files": {"<relative path>": "<file content>", ...}}
|
|
"""
|
|
|
|
|
|
def bundle_of(task_dir):
|
|
files = {}
|
|
for base, _, names in os.walk(task_dir):
|
|
for n in names:
|
|
if n.startswith(".") or n == "generation.json":
|
|
continue
|
|
path = os.path.join(base, n)
|
|
rel = os.path.relpath(path, task_dir)
|
|
if rel != "task.json":
|
|
files[rel] = open(path).read()
|
|
return {"task": json.load(open(os.path.join(task_dir, "task.json"))), "files": files}
|
|
|
|
|
|
def chat(model, messages, base_url):
|
|
body = {"model": model, "messages": messages, "temperature": 0.7}
|
|
req = urllib.request.Request(f"{base_url}/chat/completions", json.dumps(body).encode(),
|
|
{"Content-Type": "application/json", "Authorization": "Bearer none"})
|
|
for attempt in range(4):
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=1800) as r:
|
|
data = json.loads(r.read().decode())
|
|
add_usage(model, data.get("usage", {}), kind="generate")
|
|
return data["choices"][0]["message"].get("content") or ""
|
|
except Exception as e: # noqa: BLE001
|
|
if getattr(e, "code", 500) < 500 and getattr(e, "code", 500) != 429:
|
|
raise
|
|
time.sleep(15 * (attempt + 1))
|
|
raise RuntimeError("generation request failed")
|
|
|
|
|
|
def parse_bundle(text):
|
|
text = text.strip()
|
|
text = re.sub(r"^```(json)?\s*|\s*```$", "", text)
|
|
start, end = text.find("{"), text.rfind("}")
|
|
return json.loads(text[start:end + 1])
|
|
|
|
|
|
def check_bundle(b):
|
|
errs = []
|
|
t, files = b.get("task", {}), b.get("files", {})
|
|
for k in ("seed", "contract", "hidden_tests", "reference"):
|
|
if k not in t:
|
|
errs.append(f"task.json misses '{k}'")
|
|
if "spec.md" not in files:
|
|
errs.append("spec.md missing")
|
|
for k in ("seed", "hidden_tests", "reference"):
|
|
for o in t.get(k, []):
|
|
for fk in ("file", "testclasses_file"):
|
|
if fk in o and o[fk] not in files:
|
|
errs.append(f"{k}: file {o[fk]} missing")
|
|
if not o.get("name", "").startswith("{{P}}"):
|
|
errs.append(f"{k}: name {o.get('name')} does not start with {{{{P}}}}")
|
|
limit = 16 if o.get("type") == "TABL" else 26 if o.get("type") == "FUGR" else 30
|
|
if len(o.get("name", "").replace("{{P}}", "Z0000000_")) > limit:
|
|
errs.append(f"{k}: name {o.get('name')} too long (max {limit})")
|
|
if not t.get("hidden_tests"):
|
|
errs.append("no hidden test class")
|
|
return errs
|
|
|
|
|
|
def write_bundle(b, task_dir, task_id):
|
|
os.makedirs(task_dir, exist_ok=True)
|
|
b["task"]["id"] = task_id
|
|
json.dump(b["task"], open(os.path.join(task_dir, "task.json"), "w"), indent=2)
|
|
for rel, content in b["files"].items():
|
|
path = os.path.join(task_dir, rel)
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
open(path, "w").write(content)
|
|
|
|
|
|
def validate(pool_root, task_id, run_base):
|
|
r = Runner(pool_root, os.path.join(ROOT, "runs", "gen"))
|
|
os.makedirs(os.path.join(ROOT, "runs", "gen"), exist_ok=True)
|
|
rep_o, dir_o = r.run(task_id, OracleAgent(), run_base)
|
|
rep_o["_dir"] = dir_o
|
|
rep_n, _ = r.run(task_id, NullAgent(), run_base + 1)
|
|
return rep_o, rep_n
|
|
|
|
|
|
def write_errors(run_dir):
|
|
"""Failed writes and activation messages of the oracle run."""
|
|
out = []
|
|
path = os.path.join(run_dir or "", "trajectory.jsonl")
|
|
if not os.path.exists(path):
|
|
return out
|
|
for line in open(path):
|
|
e = json.loads(line)
|
|
if e.get("tool") in ("sap_create_object", "sap_push_source") and (
|
|
e["is_error"] or '"success":false' in e["result"].replace(" ", "")):
|
|
out.append({"tool": e["tool"], "object": e["args"].get("objectName"),
|
|
"include": e["args"].get("includeType", "main"), "result": e["result"][:1500]})
|
|
return out
|
|
|
|
|
|
def failure_summary(rep):
|
|
out = {"reference_write_errors": write_errors(rep.get("_dir")),"gates": rep.get("gates"), "score": rep.get("score"), "setup": rep.get("setup"),
|
|
"atc": rep.get("atc"), "abaplint": rep.get("abaplint"), "own_tests": rep.get("own_tests"),
|
|
"hidden_failed": [d for d in rep.get("hidden_tests", {}).get("detail", []) if not d["ok"]],
|
|
"hidden_install": rep.get("hidden_tests", {}).get("install")}
|
|
return json.dumps(out)[:6000]
|
|
|
|
|
|
def generate(task_id, pool, object_type, category, difficulty, model, base_url, run_base, topic=None,
|
|
max_repairs=3):
|
|
check_budget()
|
|
pool_root = os.path.join(ROOT, "tasks_gen", pool)
|
|
task_dir = os.path.join(pool_root, task_id)
|
|
example = bundle_of(os.path.join(ROOT, "tasks", EXAMPLE_FOR[object_type]))
|
|
ask = (f"Write one new task.\nObject type of the main contract object: {object_type}.\n"
|
|
f"Skill category {category}: {CATEGORIES[category]}.\nDifficulty {difficulty} of 3.\n"
|
|
+ (f"Topic idea: {topic}\n" if topic else "Choose a new, realistic business topic.\n")
|
|
+ "Here is an example bundle of a different task (same format):\n" + json.dumps(example))
|
|
messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": ask}]
|
|
log = {"id": task_id, "pool": pool, "object_type": object_type, "category": category, "attempts": []}
|
|
for attempt in range(max_repairs + 1):
|
|
text = chat(model, messages, base_url)
|
|
messages.append({"role": "assistant", "content": text})
|
|
try:
|
|
b = parse_bundle(text)
|
|
errs = check_bundle(b)
|
|
except Exception as e: # noqa: BLE001
|
|
b, errs = None, [f"invalid JSON: {e}"]
|
|
if errs:
|
|
log["attempts"].append({"stage": "bundle", "errors": errs})
|
|
messages.append({"role": "user", "content": "Fix these problems and return the full bundle again:\n"
|
|
+ "\n".join(errs)})
|
|
continue
|
|
b["task"].setdefault("object_type", object_type)
|
|
b["task"].setdefault("category", category)
|
|
write_bundle(b, task_dir, task_id)
|
|
rep_o, rep_n = validate(pool_root, task_id, run_base + 2 * attempt)
|
|
so, sn = (rep_o.get("score") or {}).get("total"), (rep_n.get("score") or {}).get("total")
|
|
log["attempts"].append({"stage": "validate", "oracle": so, "null": sn})
|
|
if so == 100 and sn == 0:
|
|
log["accepted"] = True
|
|
break
|
|
messages.append({"role": "user", "content":
|
|
"The harness ran your reference solution (oracle) and an empty solution (null). "
|
|
f"Required: oracle 100, null 0. Result: oracle {so}, null {sn}.\n"
|
|
f"Oracle report: {failure_summary(rep_o)}\n"
|
|
"Fix the bundle (reference, hidden tests, seed, or contract) and return the full "
|
|
"bundle again."})
|
|
else:
|
|
log["accepted"] = False
|
|
json.dump(log, open(os.path.join(task_dir if os.path.isdir(task_dir) else pool_root, "generation.json"), "w"),
|
|
indent=1)
|
|
return log
|
|
|
|
|
|
def main():
|
|
load_env(os.path.join(ROOT, ".env"))
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--id", required=True)
|
|
ap.add_argument("--pool", default="eval", choices=["eval", "train"])
|
|
ap.add_argument("--object-type", required=True, choices=sorted(EXAMPLE_FOR))
|
|
ap.add_argument("--category", required=True, choices=sorted(CATEGORIES))
|
|
ap.add_argument("--difficulty", type=int, default=2)
|
|
ap.add_argument("--topic")
|
|
ap.add_argument("--model", default="deepseek-v4.1-flash:cloud")
|
|
ap.add_argument("--base-url", default=os.environ.get("LLM_BASE_URL", "http://127.0.0.1:11434/v1"))
|
|
ap.add_argument("--run-base", type=int, required=True)
|
|
a = ap.parse_args()
|
|
os.makedirs(os.path.join(ROOT, "tasks_gen", a.pool), exist_ok=True)
|
|
log = generate(a.id, a.pool, a.object_type, a.category, a.difficulty, a.model, a.base_url,
|
|
a.run_base, a.topic)
|
|
print(json.dumps(log))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|