Something went wrong. Try again.
Parametric learning in your code harness.
Something went wrong. Try again.
12 kB · 368 lines
Python
at letta-code
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369"""Deterministic hidden variants and multilingual sandbox harnesses."""
from __future__ import annotations
import hashlibimport jsonimport randomimport refrom collections.abc import Iterablefrom dataclasses import dataclassfrom typing import Any, Literalfrom urllib.parse import urlparse
VARIANT_SCHEMA = "machine.project-euler.variant.v1"RESULT_SCHEMA = "machine.project-euler.variant-result.v1"RESULT_MARKER = "__MACHINE_EULER_CASE__"
VariantFamily = Literal["sum_multiples", "even_fibonacci"]Split = Literal["train", "eval"]
FAMILIES: tuple[VariantFamily, ...] = ("sum_multiples", "even_fibonacci")LANGUAGES = ("python", "javascript", "julia")SANDBOX_LANGUAGES = { "python": "python", "javascript": "nodejs", "julia": "julia",}
@dataclass(frozen=True)class VariantCase: arguments: tuple[int, ...] expected: int
def to_dict(self) -> dict[str, Any]: return {"arguments": list(self.arguments), "expected": self.expected}
@dataclass(frozen=True)class VariantTask: task_id: str family: VariantFamily split: Split seed: int cases: tuple[VariantCase, ...]
@property def case_digest(self) -> str: payload = json.dumps( [case.to_dict() for case in self.cases], sort_keys=True, separators=(",", ":"), ) return hashlib.sha256(payload.encode()).hexdigest()
@dataclass(frozen=True)class SandboxAssessment: status: Literal["passed", "failed", "submission_error", "infrastructure_error"] passed_cases: int total_cases: int diagnostic: str = ""
@property def passed(self) -> bool: return self.status == "passed"
@property def pass_rate(self) -> float: return self.passed_cases / self.total_cases if self.total_cases else 0.0
def sum_multiples(limit: int, divisor_a: int, divisor_b: int) -> int: return sum( value for value in range(limit) if value % divisor_a == 0 or value % divisor_b == 0 )
def sum_even_fibonacci(limit: int) -> int: total = 0 left, right = 1, 2 while left < limit: if left % 2 == 0: total += left left, right = right, left + right return total
def solve_family(family: VariantFamily, arguments: tuple[int, ...]) -> int: if family == "sum_multiples": if len(arguments) != 3: raise ValueError("sum_multiples expects limit, divisor_a, divisor_b") return sum_multiples(*arguments) if family == "even_fibonacci": if len(arguments) != 1: raise ValueError("even_fibonacci expects limit") return sum_even_fibonacci(arguments[0]) raise ValueError(f"unknown variant family: {family}")
def _derived_seed(seed: int, split: Split, family: VariantFamily, index: int) -> int: payload = f"{VARIANT_SCHEMA}:{seed}:{split}:{family}:{index}".encode() return int.from_bytes(hashlib.sha256(payload).digest()[:8], "big")
def _generate_arguments( family: VariantFamily, rng: random.Random, count: int,) -> list[tuple[int, ...]]: arguments: list[tuple[int, ...]] = [] seen: set[tuple[int, ...]] = set() while len(arguments) < count: if family == "sum_multiples": divisor_a, divisor_b = rng.sample(range(2, 20), 2) candidate = (rng.randint(40, 4000), divisor_a, divisor_b) elif family == "even_fibonacci": candidate = (rng.randint(20, 20_000_000),) else: raise ValueError(f"unknown variant family: {family}") if candidate in seen: continue seen.add(candidate) arguments.append(candidate) return arguments
def build_variant_tasks( families: Iterable[VariantFamily], *, split: Split, tasks_per_family: int, cases_per_task: int, seed: int,) -> list[VariantTask]: if tasks_per_family <= 0: raise ValueError("tasks_per_family must be positive") if cases_per_task <= 0: raise ValueError("cases_per_task must be positive") tasks: list[VariantTask] = [] for family in families: if family not in FAMILIES: raise ValueError(f"unknown variant family: {family}") for index in range(tasks_per_family): task_seed = _derived_seed(seed, split, family, index) rng = random.Random(task_seed) cases = tuple( VariantCase( arguments=arguments, expected=solve_family(family, arguments) ) for arguments in _generate_arguments(family, rng, cases_per_task) ) short_digest = hashlib.sha256( json.dumps( [case.to_dict() for case in cases], sort_keys=True, separators=(",", ":"), ).encode() ).hexdigest()[:12] tasks.append( VariantTask( task_id=f"{split}-{family}-{index:04d}-{short_digest}", family=family, split=split, seed=task_seed, cases=cases, ) ) return tasks
def task_set_digest(tasks: Iterable[VariantTask]) -> str: payload = [ { "task_id": task.task_id, "family": task.family, "split": task.split, "seed": task.seed, "case_digest": task.case_digest, } for task in tasks ] return hashlib.sha256( json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() ).hexdigest()
def assert_split_isolation( train: Iterable[VariantTask], eval_tasks: Iterable[VariantTask]) -> None: train_list = list(train) eval_list = list(eval_tasks) train_ids = {task.task_id for task in train_list} eval_ids = {task.task_id for task in eval_list} if train_ids & eval_ids: raise ValueError("train/eval task IDs overlap") for family in FAMILIES: train_cases = { case.arguments for task in train_list if task.family == family for case in task.cases } eval_cases = { case.arguments for task in eval_list if task.family == family for case in task.cases } if train_cases & eval_cases: raise ValueError(f"train/eval hidden cases overlap for {family}")
def build_variant_prompt(task: VariantTask, language: str) -> str: if language not in LANGUAGES: raise ValueError(f"unsupported language: {language}") if task.family == "sum_multiples": contract = ( "Define solve(limit, divisor_a, divisor_b). Return the sum of all non-negative " "integers below limit that are divisible by divisor_a or divisor_b. Count a value " "only once when both divisors match. Examples: solve(10, 3, 5) = 23 and " "solve(20, 4, 6) = 64." ) elif task.family == "even_fibonacci": contract = ( "Define solve(limit). Return the sum of the even-valued Fibonacci terms strictly " "below limit, with the sequence beginning 1, 2. Examples: solve(10) = 10 and " "solve(100) = 44." ) else: raise ValueError(f"unknown variant family: {task.family}") return ( f"Write a {language} solution for a parameterized Project Euler family. {contract}\n\n" "Use the check_solution tool with the complete function definition. The tool runs hidden " "cases in an isolated sandbox and reports only aggregate results. You may revise once if " "the first check fails. Do not print a memorized Project Euler answer or execute code at " "top level; define the requested solve function, check it, then briefly report completion." )
def build_sandbox_program( task: VariantTask, language: str, submitted_code: str, result_marker: str = RESULT_MARKER,) -> str: if language not in LANGUAGES: raise ValueError(f"unsupported language: {language}") case_arguments = [list(case.arguments) for case in task.cases] cases_json = json.dumps(case_arguments, separators=(",", ":")) if language == "python": harness = f"""_machine_cases = {cases_json}for _machine_index, _machine_arguments in enumerate(_machine_cases): _machine_value = solve(*_machine_arguments) print({result_marker!r} + str(_machine_index) + "\\t" + str(_machine_value))""" elif language == "javascript": harness = f"""const __machineCases = {cases_json};for (let __machineIndex = 0; __machineIndex < __machineCases.length; __machineIndex++) {{ const __machineValue = solve(...__machineCases[__machineIndex]); console.log({json.dumps(result_marker)} + __machineIndex + "\\t" + String(__machineValue));}}""" else: harness = f"""const _machine_cases = {cases_json}for (_machine_index, _machine_arguments) in enumerate(_machine_cases) _machine_value = solve(_machine_arguments...) println( {json.dumps(result_marker)} * string(_machine_index - 1) * "\\t" * string(_machine_value) )end""" return submitted_code.rstrip() + "\n\n" + harness.lstrip()
def ensure_loopback_sandbox_url(url: str) -> str: parsed = urlparse(url) if parsed.scheme not in {"http", "https"}: raise ValueError("sandbox URL must use http or https") if parsed.hostname not in {"127.0.0.1", "localhost", "::1"}: raise ValueError( "sandbox URL must be loopback; hidden cases may not leave the host" ) if not parsed.path.endswith("/run_code"): raise ValueError("sandbox URL must target /run_code") return url
def assess_sandbox_response( task: VariantTask, *, transport_success: bool, response: dict[str, Any], result_marker: str = RESULT_MARKER,) -> SandboxAssessment: total = len(task.cases) run_result = response.get("run_result") compile_result = response.get("compile_result") compile_failed = isinstance(compile_result, dict) and compile_result.get( "return_code" ) not in { None, 0, } run_failed = not isinstance(run_result, dict) or run_result.get("return_code") != 0 has_execution_result = isinstance(compile_result, dict) or isinstance( run_result, dict ) if not transport_success and not has_execution_result: return SandboxAssessment( status="infrastructure_error", passed_cases=0, total_cases=total, diagnostic="sandbox transport failed", ) if compile_failed or run_failed: return SandboxAssessment( status="submission_error", passed_cases=0, total_cases=total, diagnostic="submitted program did not complete", ) if not transport_success: return SandboxAssessment( status="infrastructure_error", passed_cases=0, total_cases=total, diagnostic="sandbox execution service failed", )
stdout = str(run_result.get("stdout", "")) observed: dict[int, int] = {} pattern = re.compile(rf"^{re.escape(result_marker)}(\d+)\t([+-]?\d+)\s*$") for line in stdout.splitlines(): match = pattern.match(line.strip()) if match: index = int(match.group(1)) if index in observed or index not in range(total): return SandboxAssessment( status="submission_error", passed_cases=0, total_cases=total, diagnostic="submitted program emitted an invalid result protocol", ) observed[index] = int(match.group(2)) if set(observed) != set(range(total)): return SandboxAssessment( status="submission_error", passed_cases=0, total_cases=total, diagnostic="submitted program did not return all hidden results", ) passed_cases = sum( observed[index] == case.expected for index, case in enumerate(task.cases) ) return SandboxAssessment( status="passed" if passed_cases == total else "failed", passed_cases=passed_cases, total_cases=total, diagnostic="" if passed_cases == total else f"hidden cases passed {passed_cases}/{total}", )