| """Synthesize MCPMark filesystem tasks. |
| |
| Usage (from repo root, inside the mcpmark conda env): |
| |
| python -m synth.generate --n 8 --seed 1 |
| python -m synth.generate --n 4 --types smallest_merge,duplicate_finder |
| python -m synth.generate --n 4 --no-llm # offline, no API calls |
| |
| For each task it: |
| 1. builds a test environment under test_environments/<category>/ |
| 2. writes tasks/filesystem/<suite>/<category>/<task_id>/{meta.json,description.md,verify.py} |
| 3. self-checks: solves the task with a built-in oracle and asserts verify.py |
| exits 0 on the correct answer and non-zero on the untouched environment. |
| |
| Run a generated task with the normal pipeline, e.g.: |
| |
| python -m pipeline --mcp filesystem --k 1 --models deepseek-v3.2-instruct \\ |
| --tasks synth_size_classification_01 --exp-name synth-run |
| """ |
|
|
| import argparse |
| import datetime |
| import json |
| import random |
| import shutil |
| import subprocess |
| import sys |
| import tempfile |
| from pathlib import Path |
|
|
| from synth.generators import REGISTRY |
| from synth.llm import LLMClient |
|
|
| REPO_ROOT = Path(__file__).resolve().parent.parent |
|
|
|
|
| def _state_content(env_dir: Path) -> str: |
| """A short relative listing of the test environment for meta.json.""" |
| items = sorted( |
| str(p.relative_to(env_dir)) + ("/" if p.is_dir() else "") |
| for p in env_dir.rglob("*") |
| ) |
| return "\n".join(items[:40]) |
|
|
|
|
| def write_task(gen, spec, category, task_id, suite, tasks_root, author, env_dir): |
| task_dir = tasks_root / "filesystem" / suite / category / task_id |
| task_dir.mkdir(parents=True, exist_ok=True) |
|
|
| (task_dir / "description.md").write_text(gen.description(spec), encoding="utf-8") |
| (task_dir / "verify.py").write_text(gen.verify_src(spec), encoding="utf-8") |
|
|
| meta = { |
| "task_id": task_id, |
| "task_name": gen.CATEGORY_NAME, |
| "category_id": category, |
| "category_name": gen.CATEGORY_NAME, |
| "description": gen.description(spec).split("\n\n")[1].strip()[:200], |
| "author": author, |
| "created_at": datetime.date.today().isoformat(), |
| "difficulty": gen.DIFFICULTY, |
| "tags": gen.TAGS, |
| "mcp": ["filesystem"], |
| "meta_data": { |
| "stateType": "synthetic", |
| "stateContent": _state_content(env_dir), |
| "stateUrl": None, |
| "stateOriginalUrl": None, |
| }, |
| } |
| (task_dir / "meta.json").write_text(json.dumps(meta, indent=2), encoding="utf-8") |
| return task_dir |
|
|
|
|
| def self_check(gen, spec, env_dir) -> tuple[bool, str]: |
| """Solve in a temp copy and confirm verify.py accepts it (and rejects unsolved).""" |
| verify_src = gen.verify_src(spec) |
| tmp = Path(tempfile.mkdtemp(prefix="synth_check_")) |
| try: |
| |
| neg = tmp / "neg" |
| shutil.copytree(env_dir, neg) |
| vfile = tmp / "verify.py" |
| vfile.write_text(verify_src, encoding="utf-8") |
| neg_rc = _run_verify(vfile, neg) |
| if neg_rc == 0: |
| return False, "verifier passes on the UNSOLVED environment (too weak)" |
|
|
| |
| pos = tmp / "pos" |
| shutil.copytree(env_dir, pos) |
| gen.solve(pos, spec) |
| pos_rc = _run_verify(vfile, pos) |
| if pos_rc != 0: |
| return False, "verifier REJECTS the oracle solution (rc=%d)" % pos_rc |
| return True, "oracle passes, unsolved fails" |
| finally: |
| shutil.rmtree(tmp, ignore_errors=True) |
|
|
|
|
| def _run_verify(verify_file: Path, test_dir: Path) -> int: |
| proc = subprocess.run( |
| [sys.executable, str(verify_file)], |
| env={"FILESYSTEM_TEST_DIR": str(test_dir), "PATH": __import__("os").environ.get("PATH", "")}, |
| capture_output=True, |
| text=True, |
| ) |
| return proc.returncode |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser(description="Synthesize MCPMark filesystem tasks.") |
| ap.add_argument("--n", type=int, default=4, help="number of tasks to generate") |
| ap.add_argument( |
| "--types", |
| default=",".join(REGISTRY), |
| help="comma-separated task types (default: all): " + ", ".join(REGISTRY), |
| ) |
| ap.add_argument("--suite", default="standard", choices=["standard", "easy"]) |
| ap.add_argument( |
| "--difficulty", |
| default="medium", |
| choices=["easy", "medium", "hard", "none", "all"], |
| help="difficulty tier for generators that support it (e.g. duplicate_finder)", |
| ) |
| ap.add_argument("--seed", type=int, default=1) |
| ap.add_argument("--author", default="synth-pipeline") |
| ap.add_argument("--no-llm", action="store_true", help="offline content, no API calls") |
| ap.add_argument("--model", default="deepseek/deepseek-chat") |
| ap.add_argument("--tasks-dir", default=str(REPO_ROOT / "tasks")) |
| ap.add_argument("--env-dir", default=str(REPO_ROOT / "test_environments")) |
| ap.add_argument("--prefix", default="synth", help="category-id prefix") |
| args = ap.parse_args() |
|
|
| types = [t.strip() for t in args.types.split(",") if t.strip()] |
| for t in types: |
| if t not in REGISTRY: |
| ap.error(f"unknown type '{t}'. Choices: {', '.join(REGISTRY)}") |
|
|
| rng = random.Random(args.seed) |
| llm = LLMClient(model=args.model, enabled=not args.no_llm, seed=args.seed) |
| print(f"LLM content generation: {'ON (' + args.model + ')' if llm.enabled else 'OFF (offline)'}") |
|
|
| tasks_root = Path(args.tasks_dir) |
| env_root = Path(args.env_dir) |
|
|
| ok_count, fail_count = 0, 0 |
| for i in range(args.n): |
| key = types[i % len(types)] |
| gen = REGISTRY[key](difficulty=args.difficulty) |
| category = f"{args.prefix}_{key}_{i + 1:02d}" |
| env_dir = env_root / category |
| if env_dir.exists(): |
| shutil.rmtree(env_dir) |
| env_dir.mkdir(parents=True, exist_ok=True) |
|
|
| spec = gen.build(env_dir, llm, rng) |
| passed, detail = self_check(gen, spec, env_dir) |
| if not passed: |
| fail_count += 1 |
| print(f" ✗ {category}: SELF-CHECK FAILED — {detail}") |
| shutil.rmtree(env_dir, ignore_errors=True) |
| continue |
|
|
| task_dir = write_task( |
| gen, spec, category, key, args.suite, tasks_root, args.author, env_dir |
| ) |
| ok_count += 1 |
| rel = task_dir.relative_to(REPO_ROOT) |
| n_items = sum(1 for _ in env_dir.rglob("*") if _.is_file()) |
| print(f" ✓ {category}: {n_items} files | {detail}") |
| print(f" task: {rel}") |
|
|
| print(f"\nGenerated {ok_count}/{args.n} tasks ({fail_count} failed self-check).") |
| if ok_count: |
| print("Run one with:") |
| print( |
| " python -m pipeline --mcp filesystem --k 1 " |
| "--models deepseek-v3.2-instruct " |
| f"--tasks {args.prefix}_{types[0]}_01 --exp-name synth-run" |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
