Generator training mode: train pool, eval overlap check, category mix, error-targeted tasks

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

harness/overlap.py

@@ -0,0 +1,100 @@
+"""Overlap check between a new training task and the eval tasks (spec text and object names).
+
+A training task is rejected when its spec or its object names are too close to an eval task.
+"""
+import glob
+import json
+import math
+import os
+import re
+from collections import Counter
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+STOP = set("the a an of to in is are be and or for with that this it as on by at from not no if then each "
+           "all any must shall can will when which its into than one two use used using value values "
+           "object objects method methods class classes table tables field fields return returns".split())
+SPEC_LIMIT = 0.55   # cosine of word counts of the whole spec
+GOAL_LIMIT = 0.60   # cosine of the Goal and Business rules sections
+NAME_LIMIT = 0.60   # Jaccard of name tokens (placeholder and Z prefix removed)
+
+
+def words(text):
+    text = re.sub(r"\{\{[Pp]\}\}", " ", text)
+    return [w for w in re.findall(r"[a-z][a-z0-9]{2,}", text.lower()) if w not in STOP]
+
+
+def cosine(a, b):
+    ca, cb = Counter(a), Counter(b)
+    dot = sum(ca[k] * cb[k] for k in ca if k in cb)
+    na = math.sqrt(sum(v * v for v in ca.values()))
+    nb = math.sqrt(sum(v * v for v in cb.values()))
+    return dot / (na * nb) if na and nb else 0.0
+
+
+def section(spec, start, stop=None):
+    m = re.search(rf"(?:^|\n)#*\s*\d*\.?\s*{start}.*?\n(.*?)(?=\n#*\s*\d*\.?\s*(?:{stop})|\Z)" if stop
+                  else rf"(?:^|\n)#*\s*\d*\.?\s*{start}.*?\n(.*)", spec, re.S | re.I)
+    return m.group(1) if m else ""
+
+
+def core_text(spec):
+    return section(spec, "Goal", "Open questions|Context") + "\n" + section(spec, "Business rules", "Constraints")
+
+
+def name_tokens(name):
+    n = re.sub(r"\{\{[Pp]\}\}", "", name).lower()
+    n = re.sub(r"^(?:zcl_|zif_|zcx_|z|/\w+/)", "", n)
+    return {t for t in re.split(r"[_\W]+", n) if len(t) > 2 and t not in ("test", "tab", "cls", "intf")}
+
+
+def task_names(task):
+    return [o["name"] for k in ("contract", "seed", "reference") for o in task.get(k, []) if o.get("name")]
+
+
+def load_task(d):
+    task = json.load(open(os.path.join(d, "task.json")))
+    spec = open(os.path.join(d, "spec.md")).read()
+    return {"id": task.get("id", os.path.basename(d)), "spec": words(spec), "core": words(core_text(spec)),
+            "names": [name_tokens(n) for n in task_names(task)],
+            "main": [name_tokens(o["name"]) for o in task.get("contract", []) if o.get("name")]}
+
+
+def load_pool(pool):
+    out = []
+    for d in sorted(glob.glob(os.path.join(ROOT, "tasks_gen", pool, "G*")) + glob.glob(os.path.join(ROOT, "tasks", "T*"))):
+        try:
+            out.append(load_task(d))
+        except (OSError, ValueError):
+            pass
+    return out
+
+
+def jaccard(a, b):
+    return len(a & b) / len(a | b) if a and b else 0.0
+
+
+def compare(x, y):
+    """Scores of two loaded tasks: spec cosine, core cosine, best name Jaccard (contract names)."""
+    name = max([jaccard(a, b) for a in x["main"] for b in y["main"]] or [0.0])
+    return {"spec": cosine(x["spec"], y["spec"]), "core": cosine(x["core"], y["core"]), "name": name}
+
+
+def check(task_dir_or_loaded, pool_tasks):
+    """Return a list of problems (empty = no overlap) against a list of loaded tasks."""
+    x = task_dir_or_loaded if isinstance(task_dir_or_loaded, dict) else load_task(task_dir_or_loaded)
+    problems = []
+    for y in pool_tasks:
+        if y["id"] == x["id"]:
+            continue
+        s = compare(x, y)
+        if s["spec"] >= SPEC_LIMIT or s["core"] >= GOAL_LIMIT or s["name"] >= NAME_LIMIT:
+            problems.append((y["id"], s))
+    return problems
+
+
+def load_bundle(b):
+    """Loaded form of a generated bundle ({"task", "files"}) before it is written to disk."""
+    task, spec = b["task"], b["files"].get("spec.md", "")
+    return {"id": task.get("id", "(new)"), "spec": words(spec), "core": words(core_text(spec)),
+            "names": [name_tokens(n) for n in task_names(task)],
+            "main": [name_tokens(o["name"]) for o in task.get("contract", []) if o.get("name")]}