"""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// 2. writes tasks/filesystem////{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: # Negative check: untouched environment should fail verification. 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)" # Positive check: oracle solution should pass verification. 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()