"""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": , "files": {"": "", ...}} """ 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()