From 2d695f677033e78565734d6f5b8ceafeaf29feb5 Mon Sep 17 00:00:00 2001 From: song Date: Tue, 3 Nov 2026 11:17:49 +0800 Subject: [PATCH] fix(benchmark): add FrontierCS-2 compatibility v2 recovery patch Signed-off-by: song --- benchmark/README.md | 4 + benchmark/frontiercs2/README.md | 123 ++++++ benchmark/frontiercs2/compatibility/v2.patch | 355 ++++++++++++++++ .../frontiercs2/compatibility/verify_v2.py | 386 ++++++++++++++++++ 4 files changed, 868 insertions(+) create mode 100644 benchmark/frontiercs2/README.md create mode 100644 benchmark/frontiercs2/compatibility/v2.patch create mode 100644 benchmark/frontiercs2/compatibility/verify_v2.py diff --git a/benchmark/README.md b/benchmark/README.md index f4bb9b95ea..cbbc3b932b 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -58,6 +58,10 @@ helper delivery neither launches trials nor changes frozen historical arms. ## Current work +- [FrontierCS-2 compatibility repair v2](frontiercs2/README.md) provides an opt-in + patch and synthetic HTTP/client/Docker checks for the frozen single-host v1 + sources. It preserves historical attribution and does not launch a model or + native-grader trial. - [Execution observations through ordinary Goal planning](runtime/EXECUTION_OBSERVATIONS.md) covers independent local measurements with ordinary Goal/Todo planning and stride selection, complete initial Goal retention, and migration from the diff --git a/benchmark/frontiercs2/README.md b/benchmark/frontiercs2/README.md new file mode 100644 index 0000000000..cd97739c14 --- /dev/null +++ b/benchmark/frontiercs2/README.md @@ -0,0 +1,123 @@ +# FrontierCS-2 single-host compatibility repair v2 + +This opt-in patch repairs the native-client, HTTP bridge and Docker transport +published in [the frozen v1 source](https://github.com/loopx-project/loopx/tree/f3aee4586d87b678581a22d23b61fba7e49d4672/benchmark/frontiercs2-single-host-v1). +Apply it to a new copy. The original archive and its historical scores retain +their identities. This research helper is not installed into LoopX, registered +as a capability, or run by active product CI. + +## Evidence and repaired behavior + +The [published pilot](https://huggingface.co/datasets/songwells/loopxbench/tree/7adf7b7e6828895dee4393506af80afecc843d37/public/frontiercs2-batch-local-counting-20261102-v1) +retains 63 train evaluations and four final evaluations. Its acceptance-time +fields are empty, and final workspace captures are empty or partial. Those are +observed evidence gaps. The export does not establish that a bridge restart, +receipt-write failure or blocked event loop occurred during the pilot; the +defects below are demonstrated separately with synthetic fault injection. + +| Boundary | v2 behavior | +| --- | --- | +| Repeated final submission | Replays the original remote selection and atomically replaces the local receipt; an existing receipt no longer makes a successful retry fail. | +| Local receipt IO failure | Returns `state=accepted` with `receipt_pending_recovery=true` and exit zero. Repeat `final` with the same tar and original trial capability to repair the receipt. Remote acceptance is authoritative. | +| Acceptance observation | The bridge's original Unix timestamp appears as `accepted_at` in final responses/receipts and `final_at` in the inspection projection. It is preserved on replay and restart, not reconstructed from upload/grader timestamps. | +| Bridge restart | A private v2 recovery record retains the original capability, deadline, payload identities, feedback, consumed opportunities, checkpoint/final selections and idempotent responses. Payload bytes and trial identity are checked before recovery. | +| Durable admission | Admission and its replay response are committed before train queue handoff. A failed commit rolls back in-memory admission and consumes no opportunity; a completed orphan upload can be adopted on retry. | +| Interrupted native worker | Recovery reports `controller_recovery_required`, blocks new train admission and automatic queue replay, and preserves measured cost and consumed opportunities. It neither repeats the evaluation nor invents unmeasured cost. | +| Asynchronous Docker transport | Exec and both upload routes await subprocess IO. Timeout/cancellation reaps the Docker client. Error logging no longer prints exec command/stderr fragments. | +| Uploads into writable tmpfs | Streams file bytes or tar through unprivileged `docker exec` in the container's mount namespace. It checks actual content in the Docker fixture; trusted installation retains its root-filesystem copy fallback. | + +The existing HTTP capability, train/final purpose separation, train-only status, +100-call/180,000-second train limits, 512-KiB request limit, final-selection rule +and answering deadline are retained. A checkpoint does not override a later +explicit final choice, and this patch does not select a maximum train score. +Client result schema v1 remains in use with additive local observation fields; +the private recovery record uses version 2. These fields belong only to this +experiment compatibility implementation, not a shared LoopX state vocabulary. + +The tmpfs issue is independently reproduced in the synthetic Docker fixture. +[Docker documents this copy limitation](https://docs.docker.com/reference/cli/docker/container/cp/#corner-cases) +and recommends tar through `docker exec` for mounted paths. This also explains +why a successful copy command alone cannot qualify the historical workspace +captures; it does not prove which files existed before the old capture. + +## Prepare and verify + +From the LoopX repository root, obtain the pinned source commit and materialize +a separate disposable directory: + +```sh +git fetch origin f3aee4586d87b678581a22d23b61fba7e49d4672 +FRONTIER_V2_ROOT=$(mktemp -d) +git archive f3aee4586d87b678581a22d23b61fba7e49d4672 benchmark/frontiercs2-single-host-v1 | + tar -x -C "$FRONTIER_V2_ROOT" --strip-components=2 +git -C "$FRONTIER_V2_ROOT" apply --check "$PWD/benchmark/frontiercs2/compatibility/v2.patch" +git -C "$FRONTIER_V2_ROOT" apply "$PWD/benchmark/frontiercs2/compatibility/v2.patch" +uv run --extra test python benchmark/frontiercs2/compatibility/verify_v2.py \ + --source "$FRONTIER_V2_ROOT" --docker-image python:3.12-slim +``` + +Use the exact source identity above; reject a mismatched base instead of forcing +patch application. The original v1 hash checker still applies only to v1. +Review [v2.patch](compatibility/v2.patch) as the implementation delta. + +The opt-in [verification script](compatibility/verify_v2.py) uses disposable +owner-only directories, loopback HTTP, real client subprocesses and synthetic +archives. `--docker-image` uses an already-local Python image with `/bin/bash`; +it never pulls an image. It creates and removes only its uniquely named +container, with network disabled and no host mounts. The fixture normalizes +the Python interpreter path and checks concurrent exec plus sealed/streamed +file and directory uploads. Omitting the image explicitly skips real Docker +coverage. This is transport validation, not campaign sandbox qualification. +No model, task, native evaluator, reference solution or historical candidate +is executed by these checks. Experiment-specific tests remain opt-in here. + +## Controller integration and remaining limits + +Use a new trial directory for a new experiment. A v1 inspection-only state file +is deliberately rejected as a recovery source. The controller must own a +single-writer process lock and an owner-only trial directory. The new +`trial-recovery.json` contains a live capability and local payload metadata: +keep it private, including in backups; never upload it with public results. +`submission-state.json` is an inspection projection, not recovery authority. +The recovery commit precedes projection publication; an observation-write +failure logs a warning and cannot undo a durably accepted submission. A later +persist refreshes the projection. +Atomic file publication covers process restart; power-loss durability and +multi-process concurrent ownership are not qualified here. + +For a recovered trial, construct `Trial` with its original directory, arm and a +new queue, register the recovered `trial.token`, and preserve its original +deadline. Do not call `start()` again. After stopping/reconciling the old scorer, +call `trial.requeue_pending()` once to hand off queued train work. Repeated calls +on that recovered instance do not enqueue it twice. The controller still owns +phase closure, queued-work cancellation and final-evaluation dispatch. + +An interrupted running worker requires reconciliation against the original +process and native result. Preserve the consumed call; record actual elapsed +cost and its native outcome before clearing the recovery failure and persisting. +Do not clear it to obtain another opportunity or restart the original budget. +The patch cannot determine whether an external native scorer survived a +controller crash, so it fails closed at this boundary. + +The external campaign controller, isolated model execution, mounts/firewall, +workspace capture, native grader and its task-specific prerequisites remain +outside this repair. In particular, the missing historical workspace files +cannot be recovered from submitted tar files, and the new timestamps do not +backfill missing historical acceptance times. Killing a Docker client does not +prove the container command stopped; the controller retains descendant cleanup +and hard-stop ownership. No model/scorer rerun or score benefit is claimed. + +To roll back, stop the disposable controller/containers and discard the patched +source copy. Preserve private v2 recovery data for diagnosis; do not feed it to +v1 or reinterpret the original study as a repaired run. + +## 中文说明 + +这是对冻结 v1 源码的独立修复补丁：保留原始实验、分数和归档字节，修复重复 final、 +回执失败恢复、提交接受时间缺失、桥接重启恢复及异步 Docker 阻塞。以上命令在新目录 +应用补丁并执行合成 HTTP、客户端和真实 Docker 验证，不调用模型或裁判。 + +恢复必须保留原 token、deadline、已消耗次数和实测成本；遇到中断的运行中评测，先核对 +原进程和原生结果，不能重复评分、补发预算或虚构成本。恢复文件含凭据，只能私有保存。 +原始 controller、工作区完整留存和沙箱资格仍未交付；本修复不能补回历史缺失文件和 +时间，也不证明 LoopX 的分数或长程效率提升。 diff --git a/benchmark/frontiercs2/compatibility/v2.patch b/benchmark/frontiercs2/compatibility/v2.patch new file mode 100644 index 0000000000..d5fb7f168e --- /dev/null +++ b/benchmark/frontiercs2/compatibility/v2.patch @@ -0,0 +1,355 @@ +diff --git a/docker_environment.py b/docker_environment.py +--- a/docker_environment.py ++++ b/docker_environment.py +@@ -1,4 +1,5 @@ + """Private Docker transport for the existing shared benchmark installer.""" ++import asyncio + import io + from pathlib import Path + import shlex +@@ -15,6 +16,25 @@ + self.security_bootstrap = security_bootstrap + self.sealed = sealed + ++ @staticmethod ++ async def _run(argv, *, timeout, data=None): ++ process = await asyncio.create_subprocess_exec( ++ *argv, stdin=asyncio.subprocess.PIPE if data is not None else None, ++ stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) ++ try: ++ stdout, stderr = await asyncio.wait_for(process.communicate(data), timeout) ++ except (TimeoutError, asyncio.CancelledError) as error: ++ if process.returncode is None: ++ try: ++ process.kill() ++ except ProcessLookupError: ++ pass ++ await process.wait() ++ if isinstance(error, TimeoutError): ++ raise subprocess.TimeoutExpired(argv, timeout) from error ++ raise ++ return SimpleNamespace(returncode=process.returncode, stdout=stdout, stderr=stderr) ++ + async def exec(self, command, *, user=None, env=None, cwd=None, timeout_sec=None): + effective_user = self.default_user if self.sealed else user or self.default_user + argv = ['docker', 'exec', '--user', effective_user] +@@ -29,19 +49,20 @@ + else: + argv += [self.name, '/usr/bin/python3', '-c', self.security_bootstrap, + '/bin/bash', '-c', command] +- result = subprocess.run(argv, capture_output=True, text=True, timeout=timeout_sec or 300) ++ result = await self._run(argv, timeout=timeout_sec or 300) ++ stdout, stderr = result.stdout.decode(), result.stderr.decode() + if result.returncode: +- print('environment command exit', result.returncode, 'user', user, +- 'command', command[:200], 'stderr', result.stderr[-1200:], flush=True) +- return SimpleNamespace(return_code=result.returncode, stdout=result.stdout, stderr=result.stderr) ++ print('environment command exit', result.returncode, flush=True) ++ return SimpleNamespace(return_code=result.returncode, stdout=stdout, stderr=stderr) + + async def upload_file(self, source, target): +- if self.sealed: +- code = 'import pathlib,sys; p=pathlib.Path(sys.argv[1]); p.write_bytes(sys.stdin.buffer.read()); p.chmod(0o644)' +- result = subprocess.run(['docker', 'exec', '-i', '--user', self.default_user, ++ writable = await self.exec('test -w ' + shlex.quote(str(Path(target).parent))) ++ if self.sealed or writable.return_code == 0: ++ mode = 0o644 if self.sealed else Path(source).stat().st_mode & 0o777 ++ code = 'import pathlib,sys; p=pathlib.Path(sys.argv[1]); p.write_bytes(sys.stdin.buffer.read()); p.chmod(int(sys.argv[2]))' ++ result = await self._run(['docker', 'exec', '-i', '--user', self.default_user, + self.name, '/usr/bin/python3', '-c', self.security_bootstrap, +- '/usr/bin/python3', '-c', code, target], input=Path(source).read_bytes(), +- capture_output=True, timeout=120) ++ '/usr/bin/python3', '-c', code, target, str(mode)], data=Path(source).read_bytes(), timeout=120) + if result.returncode: + raise RuntimeError(result.stderr.decode(errors='replace')[-1500:]) + return +@@ -54,15 +75,18 @@ + info.uname = info.gname = '' + return info + archive.add(source, arcname=Path(target).name, filter=owned) +- result = subprocess.run(['docker', 'cp', '-', self.name + ':' + str(Path(target).parent)], +- input=stream.getvalue(), capture_output=True, timeout=120) ++ result = await self._run(['docker', 'cp', '-', self.name + ':' + str(Path(target).parent)], ++ data=stream.getvalue(), timeout=120) + if result.returncode: + raise RuntimeError(result.stderr.decode(errors='replace')[-1500:]) + + async def upload_dir(self, source, target): +- result = await self.exec('mkdir -p ' + shlex.quote(target), user='root') ++ result = await self.exec('mkdir -p ' + shlex.quote(target)) ++ if result.return_code and not self.sealed: ++ result = await self.exec('mkdir -p ' + shlex.quote(target), user='root') + if result.return_code: + raise RuntimeError(result.stderr) ++ writable = await self.exec('test -w ' + shlex.quote(target)) + stream = io.BytesIO() + with tarfile.open(fileobj=stream, mode='w') as archive: + def owned(info): +@@ -70,7 +94,15 @@ + info.uname = info.gname = '' + return info + archive.add(source, arcname='.', filter=owned) +- result = subprocess.run(['docker', 'cp', '-', self.name + ':' + target], +- input=stream.getvalue(), capture_output=True, timeout=240) ++ if self.sealed or writable.return_code == 0: ++ # docker cp cannot reliably populate tmpfs mounts. Extract in the ++ # container's mount namespace, keeping the unprivileged owner. ++ result = await self._run(['docker', 'exec', '-i', '--user', self.default_user, ++ self.name, '/usr/bin/python3', '-c', self.security_bootstrap, ++ '/bin/tar', '-xpf', '-', '--no-same-owner', '-C', target], ++ data=stream.getvalue(), timeout=240) ++ else: ++ result = await self._run(['docker', 'cp', '-', self.name + ':' + target], ++ data=stream.getvalue(), timeout=240) + if result.returncode: + raise RuntimeError(result.stderr.decode(errors='replace')[-1500:]) +diff --git a/frontier_submit.py b/frontier_submit.py +--- a/frontier_submit.py ++++ b/frontier_submit.py +@@ -14,6 +14,7 @@ + import stat + import sys + import tarfile ++import tempfile + import time + from urllib.parse import urlsplit + +@@ -236,11 +237,27 @@ + raise SubmitError("Driver submission response is invalid") + result = {"schema_version": SCHEMA, "operation": "final", "state": "accepted", + "payload_id": payload_id, "verification_id": verification_id} +- FINAL_RECEIPT.parent.mkdir(mode=0o700, parents=True, exist_ok=True) +- flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL +- descriptor = os.open(FINAL_RECEIPT, flags, 0o600) +- with os.fdopen(descriptor, "wb") as destination: +- destination.write(canonical(result)); destination.flush(); os.fsync(destination.fileno()) ++ if "accepted_at" in view: ++ result["accepted_at"] = view["accepted_at"] ++ temporary = None ++ try: ++ FINAL_RECEIPT.parent.mkdir(mode=0o700, parents=True, exist_ok=True) ++ descriptor, temporary = tempfile.mkstemp(dir=FINAL_RECEIPT.parent, prefix=".final-") ++ with os.fdopen(descriptor, "wb") as destination: ++ destination.write(canonical(result)) ++ destination.flush() ++ os.fsync(destination.fileno()) ++ os.replace(temporary, FINAL_RECEIPT) ++ except OSError: ++ # Remote acceptance is the authority even when this local cache fails. ++ # Repeating final with the same tar replays the remote intent and repairs it. ++ result["receipt_pending_recovery"] = True ++ finally: ++ if temporary is not None: ++ try: ++ Path(temporary).unlink(missing_ok=True) ++ except OSError: ++ pass + return result + + +diff --git a/submission_bridge.py b/submission_bridge.py +--- a/submission_bridge.py ++++ b/submission_bridge.py +@@ -3,12 +3,17 @@ + This is a single-host experiment adapter, not the upstream Management/Driver + implementation. Every accepted train request uses its real verifier output. + """ ++import copy + import hashlib + from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + import json ++import logging ++import os + from pathlib import Path ++import re + import secrets + import threading ++import tempfile + import time + import uuid + +@@ -34,21 +39,102 @@ + self.final = None + self.final_at = None + self.rejected_requests = 0 ++ self._pending_requeued = False ++ self.root.mkdir(parents=True, mode=0o700, exist_ok=True) ++ if self.root.stat().st_mode & 0o077: ++ raise ValueError('trial directory must be owner-only') + (self.root/'payloads').mkdir(parents=True, mode=0o700, exist_ok=True) ++ self.recovery_path = self.root/'trial-recovery.json' ++ if self.recovery_path.exists(): ++ self._restore() ++ elif (self.root/'submission-state.json').exists(): ++ raise ValueError('v1 observation is not a recoverable trial; use a new directory') + + def snapshot(self): + return {'arm': self.arm, 'phase': self.phase, 'deadline': self.deadline, + 'train_calls': self.train_calls, 'train_wall_seconds': self.train_wall_seconds, + 'saved_payload': self.saved, 'final_payload': self.final, ++ 'final_at': self.final_at, + 'jobs': {k: {x: v[x] for x in ['mode', 'state', 'payload_id']} + for k, v in self.jobs.items()}, + 'rejected_requests': self.rejected_requests} + ++ def _recovery(self): ++ return {'version': 2, 'arm': self.arm, **{key: copy.deepcopy(getattr(self, key)) ++ for key in ('token', 'deadline', 'phase', 'payloads', 'jobs', 'idempotency', ++ 'train_calls', 'train_wall_seconds', 'saved', 'final', ++ 'final_at', 'rejected_requests')}} ++ ++ def _restore(self): ++ path = self.recovery_path ++ if path.is_symlink() or path.stat().st_mode & 0o077: ++ raise ValueError('recovery record must be an owner-only regular file') ++ record = json.loads(path.read_text()) ++ if record.get('version') != 2 or record.get('arm') != self.arm: ++ raise ValueError('recovery identity mismatch') ++ if set(record) != set(self._recovery()): ++ raise ValueError('incomplete recovery record') ++ for identifier, payload in record['payloads'].items(): ++ digest = payload['sha256'] ++ if (not re.fullmatch(r'sha256:[0-9a-f]{64}', digest) ++ or payload['purpose'] not in ('train', 'final') ++ or identifier != 'payload_'+digest.removeprefix('sha256:')+'_'+payload['purpose']): ++ raise ValueError('recovery payload identity mismatch') ++ source = self.root/'payloads'/(identifier+'.tar') ++ if source.is_symlink() or 'sha256:'+hashlib.sha256(source.read_bytes()).hexdigest() != digest: ++ raise ValueError('recovery payload bytes mismatch') ++ payload['path'] = str(source) ++ for key, value in record.items(): ++ if key not in ('version', 'arm'): ++ setattr(self, key, value) ++ # Never replay an uncertain native evaluation or refund its opportunity. ++ # The controller must reconcile its original worker/result separately. ++ for job in self.jobs.values(): ++ if job['state'] == 'running': ++ job.update(state='failed', failure={'code': 'controller_recovery_required', ++ 'message': 'Reconcile the original native worker and result.', ++ 'retryable': False}) ++ ++ @staticmethod ++ def _write(path, value): ++ descriptor, temporary = tempfile.mkstemp(dir=path.parent, prefix='.'+path.name+'-') ++ try: ++ with os.fdopen(descriptor, 'w') as output: ++ json.dump(value, output, indent=2, allow_nan=False) ++ output.flush() ++ os.fsync(output.fileno()) ++ os.replace(temporary, path) ++ finally: ++ try: ++ Path(temporary).unlink(missing_ok=True) ++ except OSError: ++ pass ++ + def persist(self): +- p = self.root/'submission-state.json' +- temporary = p.with_suffix('.pending') +- temporary.write_text(json.dumps(self.snapshot(), indent=2)) +- temporary.replace(p) ++ with self.lock: ++ self._write(self.recovery_path, self._recovery()) ++ try: ++ self._write(self.root/'submission-state.json', self.snapshot()) ++ except OSError: ++ # An observation failure cannot undo an accepted durable intent. ++ logging.getLogger(__name__).warning('recovery committed; inspection projection unavailable') ++ ++ def requeue_pending(self): ++ """Call once after controller recovery, with the old scorer stopped.""" ++ with self.lock: ++ if self._pending_requeued: ++ return ++ if self._needs_reconciliation(): ++ raise Rejected(409, 'original native work requires reconciliation') ++ self.persist() ++ for identifier, job in self.jobs.items(): ++ if job['state'] == 'queued' and job['mode'] == 'train': ++ self.queue.put((self, identifier)) ++ self._pending_requeued = True ++ ++ def _needs_reconciliation(self): ++ return any(job.get('failure', {}).get('code') == 'controller_recovery_required' ++ for job in self.jobs.values()) + + def start(self, seconds): + with self.lock: +@@ -73,10 +159,20 @@ + if old != fingerprint: + raise Rejected(409, 'idempotency key is bound to different bytes') + return status, response +- status, result = self._request(method, path, body, headers) +- if key: +- self.idempotency[key] = (fingerprint, status, result) +- self.persist() ++ before = self._recovery() ++ try: ++ status, result = self._request(method, path, body, headers) ++ if key: ++ self.idempotency[key] = (fingerprint, status, copy.deepcopy(result)) ++ self.persist() ++ except BaseException: ++ for name, value in before.items(): ++ if name not in ('version', 'arm'): ++ setattr(self, name, value) ++ raise ++ # Admission and the replay response are durable before worker handoff. ++ if method == 'POST' and path == '/v1/verifier-train': ++ self.queue.put((self, result['verification_id'])) + return status, result + + def _request(self, method, path, body, headers): +@@ -99,7 +195,14 @@ + identifier = 'payload_'+digest.split(':')[1]+'_'+purpose + if identifier not in self.payloads: + target = self.root/'payloads'/(identifier+'.tar') +- target.write_bytes(body) ++ if target.exists(): ++ if target.is_symlink() or target.read_bytes() != body: ++ raise Rejected(409, 'existing payload bytes mismatch') ++ else: ++ with target.open('xb') as output: ++ output.write(body) ++ output.flush() ++ os.fsync(output.fileno()) + # The enclosing host directory remains private; the native + # unprivileged verifier receives this one file read-only. + target.chmod(0o444) +@@ -112,6 +215,8 @@ + except (ValueError, KeyError, TypeError): + raise Rejected(400, 'unknown trial payload') from None + if method == 'POST' and path == '/v1/verifier-train': ++ if self._needs_reconciliation(): ++ raise Rejected(409, 'original native work requires reconciliation') + if payload['purpose'] != 'train': + raise Rejected(400, 'payload purpose mismatch') + if self.train_calls >= 100 or self.train_wall_seconds >= 180000: +@@ -121,7 +226,6 @@ + 'mode': 'train', 'state': 'queued', 'metrics': {}} + self.jobs[verification] = job + self.train_calls += 1 +- self.queue.put((self, verification)) + return 202, self.view(job) + if method == 'POST' and path == '/v1/submissions/checkpoint': + if payload['purpose'] != 'final': +@@ -139,7 +243,7 @@ + self.phase = 'final_selected' + self.jobs[verification] = {'verification_id': verification, + 'payload_id': identifier, 'mode': 'eval', 'state': 'queued', 'metrics': {}} +- return 202, {'verification_id': verification} ++ return 202, {'verification_id': verification, 'accepted_at': self.final_at} + raise Rejected(404, 'unsupported operation') + + +@@ -178,7 +282,9 @@ + status, result = error.status, {'error': error.reason} + if trial: + trial.rejected_requests += 1 +- except (ValueError, OSError): ++ except OSError: ++ status, result = 503, {'error': 'trial state commit unavailable'} ++ except ValueError: + status, result = 400, {'error': 'malformed request'} + raw = json.dumps(result, separators=(',', ':'), allow_nan=False).encode() + self.send_response(status) diff --git a/benchmark/frontiercs2/compatibility/verify_v2.py b/benchmark/frontiercs2/compatibility/verify_v2.py new file mode 100644 index 0000000000..9565a6b080 --- /dev/null +++ b/benchmark/frontiercs2/compatibility/verify_v2.py @@ -0,0 +1,386 @@ +#!/usr/bin/env python3 +"""Opt-in compatibility checks with synthetic payloads; no model or native grader.""" + +from __future__ import annotations + +import argparse +import asyncio +import importlib.util +import io +import json +import os +from pathlib import Path +import queue +import shlex +import subprocess +import sys +import tarfile +import tempfile +import threading +import time +import unittest +from unittest.mock import patch +import uuid + +SOURCE: Path +DOCKER_IMAGE: str | None = None + + +def load(name): + spec = importlib.util.spec_from_file_location(name, SOURCE / (name + '.py')) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def archive(path, content=b'synthetic candidate'): + with tarfile.open(path, 'w', format=tarfile.USTAR_FORMAT) as output: + member = tarfile.TarInfo('payload/solution.patch') + member.size = len(content) + output.addfile(member, io.BytesIO(content)) + return path + + +class Compatibility(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory(prefix='frontiercs2-contract-') + self.addCleanup(self.temporary.cleanup) + self.root = Path(self.temporary.name) + self.bridge_module = load('submission_bridge') + self.client_module = load('frontier_submit') + self.client_module.REQUEST_ATTEMPTS = 1 + self.jobs = queue.Queue() + self.trial = self.bridge_module.Trial(self.root / 'trial', 'fixture', self.jobs) + self.trial.start(60) + self.payload = archive(self.root / 'candidate.tar') + self.client_module.FINAL_RECEIPT = self.root / 'receipt' / 'final.json' + self.token_file = self.root / 'capability' + self.token_file.write_text(self.trial.token) + self.token_file.chmod(0o600) + self.server = None + self.start_server() + self.addCleanup(self.stop_server) + environment = patch.dict(os.environ, { + 'FRONTIER_DRIVER_API_URL': self.origin, + 'FRONTIER_DRIVER_TOKEN_FILE': str(self.token_file), + }) + environment.start() + self.addCleanup(environment.stop) + self.client = self.client_module.Client() + + def start_server(self, port=0): + self.server = self.bridge_module.Bridge(('127.0.0.1', port)) + self.server.trials[self.trial.token] = self.trial + self.thread = threading.Thread(target=self.server.serve_forever, + kwargs={'poll_interval': 0.02}, daemon=True) + self.thread.start() + self.origin = f'http://127.0.0.1:{self.server.server_address[1]}' + + def stop_server(self): + if self.server is not None: + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=3) + self.server = None + + def restart(self): + port = self.server.server_address[1] + self.stop_server() + self.jobs = queue.Queue() + self.trial = self.bridge_module.Trial(self.root / 'trial', 'fixture', self.jobs) + self.start_server(port) + + def test_repeat_final_reuses_receipt_and_remote_selection(self): + first = self.client_module.run_final(self.client, self.payload) + second = self.client_module.run_final(self.client, self.payload) + self.assertEqual(first, second) + self.assertEqual(first['state'], 'accepted') + self.assertEqual(len(self.trial.jobs), 1) + self.assertEqual(first['accepted_at'], self.trial.snapshot()['final_at']) + self.assertEqual(json.loads(self.client_module.FINAL_RECEIPT.read_text()), first) + self.assertEqual(self.client_module.FINAL_RECEIPT.stat().st_mode & 0o777, 0o600) + + def test_receipt_io_failure_keeps_acceptance_and_recovers(self): + blocked = self.root / 'not-a-directory' + blocked.write_text('fixture') + self.client_module.FINAL_RECEIPT = blocked / 'final.json' + accepted = self.client_module.run_final(self.client, self.payload) + self.assertEqual(accepted['state'], 'accepted') + self.assertTrue(accepted['receipt_pending_recovery']) + self.client_module.FINAL_RECEIPT = self.root / 'recovered.json' + recovered = self.client_module.run_final(self.client, self.payload) + self.assertEqual(recovered['verification_id'], accepted['verification_id']) + self.assertEqual(recovered['accepted_at'], accepted['accepted_at']) + self.assertNotIn('receipt_pending_recovery', recovered) + self.assertEqual(len(self.trial.jobs), 1) + + def test_client_cli_repeats_without_error_or_duplicate_effect(self): + code = ( + 'import importlib.util,pathlib,sys; ' + 's=importlib.util.spec_from_file_location("client",sys.argv[1]); ' + 'm=importlib.util.module_from_spec(s); s.loader.exec_module(m); ' + 'm.FINAL_RECEIPT=pathlib.Path(sys.argv[2]); ' + 'sys.exit(m.main(["final",sys.argv[3]]))' + ) + argv = [sys.executable, '-c', code, str(SOURCE / 'frontier_submit.py'), + str(self.client_module.FINAL_RECEIPT), str(self.payload)] + rows = [subprocess.run(argv, capture_output=True, text=True, timeout=10) + for _ in range(2)] + self.assertEqual([r.returncode for r in rows], [0, 0]) + self.assertEqual(json.loads(rows[0].stdout), json.loads(rows[1].stdout)) + self.assertEqual(len(self.trial.jobs), 1) + + def test_restart_preserves_capability_deadline_cost_and_feedback(self): + result, _ = self.client_module.run_train(self.client, self.payload) + job_id = result['verification_id'] + self.trial.jobs[job_id].update(state='passed', raw_score=7, metrics={'score_unbounded': 7}) + self.trial.train_wall_seconds = 13.5 + self.trial.persist() + before_token, before_deadline = self.trial.token, self.trial.deadline + self.restart() + self.assertEqual(self.trial.token, before_token) + self.assertEqual(self.trial.deadline, before_deadline) + self.assertEqual(self.trial.train_calls, 1) + self.assertEqual(self.trial.train_wall_seconds, 13.5) + feedback, status = self.client_module.run_status(self.client, job_id) + self.assertEqual((status, feedback['raw_score']), (0, 7)) + self.client_module.run_train(self.client, self.payload) + self.assertEqual(self.trial.train_calls, 1) + self.assertTrue(self.jobs.empty()) + record = self.root / 'trial' / 'trial-recovery.json' + self.assertEqual(record.stat().st_mode & 0o777, 0o600) + self.assertNotIn(before_token, (record.parent / 'submission-state.json').read_text()) + + def test_final_replay_after_restart_and_deadline_is_readback(self): + first = self.client_module.run_final(self.client, self.payload) + self.trial.deadline = time.time() - 1 + self.trial.persist() + self.restart() + second = self.client_module.run_final(self.client, self.payload) + self.assertEqual(second, first) + self.assertEqual(len(self.trial.jobs), 1) + self.assertTrue(self.jobs.empty()) + other = archive(self.root / 'other.tar', b'new candidate') + with self.assertRaisesRegex(self.client_module.SubmitError, '409'): + self.client_module.run_final(self.client, other) + + def test_queued_handoff_is_explicit_and_not_repeated(self): + result, _ = self.client_module.run_train(self.client, self.payload) + self.restart() + self.assertTrue(self.jobs.empty()) + self.trial.requeue_pending() + self.trial.requeue_pending() + self.assertEqual(self.jobs.get_nowait(), (self.trial, result['verification_id'])) + self.assertTrue(self.jobs.empty()) + self.assertEqual(self.trial.train_calls, 1) + + def test_interrupted_native_worker_requires_reconciliation(self): + result, _ = self.client_module.run_train(self.client, self.payload) + self.trial.jobs[result['verification_id']]['state'] = 'running' + self.trial.train_wall_seconds = 3.5 + self.trial.persist() + self.restart() + feedback, status = self.client_module.run_status(self.client, result['verification_id']) + self.assertEqual(status, 2) + self.assertEqual(feedback['failure']['code'], 'controller_recovery_required') + self.assertEqual(self.trial.train_wall_seconds, 3.5) + self.assertEqual(self.trial.train_calls, 1) + self.assertTrue(self.jobs.empty()) + with self.assertRaises(self.bridge_module.Rejected): + self.trial.requeue_pending() + other = archive(self.root / 'other.tar', b'new candidate') + with self.assertRaisesRegex(self.client_module.SubmitError, '409'): + self.client_module.run_train(self.client, other) + + def test_failed_payload_commit_can_adopt_its_immutable_orphan(self): + real_write = self.trial._write + + def fail_recovery(path, value): + if path == self.trial.recovery_path: + raise OSError('synthetic failed commit') + real_write(path, value) + + with patch.object(self.trial, '_write', side_effect=fail_recovery): + with self.assertRaises(self.client_module.SubmitError): + self.client_module.run_train(self.client, self.payload) + self.assertEqual(self.trial.train_calls, 0) + self.assertEqual(self.trial.payloads, {}) + self.assertTrue(self.jobs.empty()) + # Reuse the immutable orphan left by the failed upload commit. + self.client_module.run_train(self.client, self.payload) + self.assertEqual(self.trial.train_calls, 1) + self.assertEqual(self.jobs.qsize(), 1) + + def test_failed_train_commit_does_not_enqueue_or_consume_an_opportunity(self): + self.client_module.upload(self.client, 'train', self.payload) + real_write = self.trial._write + + def fail_admission(path, value): + if path == self.trial.recovery_path and value['train_calls']: + raise OSError('synthetic failed admission') + real_write(path, value) + + with patch.object(self.trial, '_write', side_effect=fail_admission): + with self.assertRaisesRegex(self.client_module.SubmitError, '503'): + self.client_module.run_train(self.client, self.payload) + self.assertEqual(self.trial.train_calls, 0) + self.assertEqual(self.trial.jobs, {}) + self.assertTrue(self.jobs.empty()) + self.restart() + self.assertEqual(self.trial.train_calls, 0) + self.client_module.run_train(self.client, self.payload) + self.assertEqual(self.trial.train_calls, 1) + self.assertEqual(self.jobs.qsize(), 1) + + def test_checkpoint_and_explicit_final_selection_survive_restart(self): + first = self.client_module.run_store_final(self.client, self.payload) + other = archive(self.root / 'other.tar', b'other candidate') + second = self.client_module.run_store_final(self.client, other) + self.restart() + self.assertEqual(self.trial.saved, second['payload_id']) + # Preserve an explicit final choice even when it differs from the checkpoint. + final = self.client_module.run_final(self.client, self.payload) + self.assertEqual(final['payload_id'], first['payload_id']) + self.assertEqual(self.trial.saved, second['payload_id']) + self.assertEqual(self.trial.final, first['payload_id']) + + def test_projection_failure_does_not_reject_a_committed_final(self): + real_write = self.trial._write + + def fail_projection(path, value): + if path.name == 'submission-state.json': + raise OSError('synthetic observation failure') + real_write(path, value) + + with patch.object(self.trial, '_write', side_effect=fail_projection): + final = self.client_module.run_final(self.client, self.payload) + self.assertEqual(final['state'], 'accepted') + self.restart() + replay = self.client_module.run_final(self.client, self.payload) + self.assertEqual(replay, final) + self.assertEqual(len(self.trial.jobs), 1) + + def test_wrong_capability_purpose_and_final_status_stay_rejected(self): + auth = self.client.auth + self.client.auth = 'wrong-fixture-capability' + with self.assertRaisesRegex(self.client_module.SubmitError, '401'): + self.client_module.upload(self.client, 'train', self.payload) + self.client.auth = auth + payload, _ = self.client_module.upload(self.client, 'train', self.payload) + with self.assertRaisesRegex(self.client_module.SubmitError, '400'): + self.client.request('POST', '/v1/submissions', + body=json.dumps({'payload_id': payload}).encode()) + final = self.client_module.run_final(self.client, self.payload) + self.trial.jobs[final['verification_id']].update(state='passed', raw_score=99) + with self.assertRaisesRegex(self.client_module.SubmitError, '404'): + self.client_module.run_status(self.client, final['verification_id']) + + def test_restart_rejects_wrong_identity_and_tampered_payload(self): + self.client_module.upload(self.client, 'train', self.payload) + with self.assertRaises(ValueError): + self.bridge_module.Trial(self.root / 'trial', 'other-arm', queue.Queue()) + payload = next(iter(self.trial.payloads.values())) + path = Path(payload['path']) + path.chmod(0o600) + path.write_bytes(b'tampered fixture') + with self.assertRaises(ValueError): + self.bridge_module.Trial(self.root / 'trial', 'fixture', queue.Queue()) + + def test_original_quota_and_deadline_remain_enforced(self): + self.trial.train_calls = 100 + with self.assertRaisesRegex(self.client_module.SubmitError, '429'): + self.client_module.run_train(self.client, self.payload) + self.trial.train_calls = 0 + self.trial.train_wall_seconds = 180000 + with self.assertRaisesRegex(self.client_module.SubmitError, '429'): + self.client_module.run_train(self.client, self.payload) + self.trial.deadline = time.time() - 1 + other = archive(self.root / 'other.tar', b'other candidate') + with self.assertRaisesRegex(self.client_module.SubmitError, '409'): + self.client_module.run_train(self.client, other) + + +class DockerTransport(unittest.TestCase): + def test_event_loop_can_release_a_real_child_process(self): + module = load('docker_environment') + + async def check(root): + gate = root / 'gate' + child = 'import pathlib,sys,time; p=pathlib.Path(sys.argv[1]);\nwhile not p.exists(): time.sleep(.01)\nprint("released")' + task = asyncio.create_task(module.DockerEnvironment._run( + [sys.executable, '-c', child, str(gate)], timeout=3)) + await asyncio.sleep(0.02) + gate.touch() + result = await task + self.assertEqual(result.stdout.strip(), b'released') + with self.assertRaises(subprocess.TimeoutExpired): + await module.DockerEnvironment._run( + [sys.executable, '-c', 'import time; time.sleep(30)'], timeout=0.05) + + with tempfile.TemporaryDirectory() as temporary: + asyncio.run(check(Path(temporary))) + + def test_real_docker_exec_and_both_upload_paths(self): + if not DOCKER_IMAGE: + self.skipTest('pass --docker-image to exercise a real isolated container') + module = load('docker_environment') + name = 'frontiercs2-contract-' + uuid.uuid4().hex + # This synthetic transport fixture does not qualify the campaign sandbox. + bootstrap = 'import os,sys; os.execv(sys.argv[1],sys.argv[1:])' + subprocess.run(['docker', 'run', '-d', '--rm', '--pull=never', '--name', name, + '--network=none', '--cap-drop=ALL', + '--memory=128m', '--pids-limit=64', + '--tmpfs', '/workspace:rw,uid=10001,gid=10001,mode=0777', + DOCKER_IMAGE, 'sleep', '120'], + capture_output=True, check=True, timeout=15) + try: + subprocess.run(['docker', 'exec', '--user=0', name, '/bin/sh', '-c', + 'test -x /usr/bin/python3 || ln -s /usr/local/bin/python3 /usr/bin/python3'], + capture_output=True, check=True, timeout=10) + async def check(root): + env = module.DockerEnvironment(name, bootstrap, sealed=True) + wait = 'import pathlib,time; p=pathlib.Path("/workspace/gate");\nwhile not p.exists(): time.sleep(.01)\nprint("ready")' + first, second = await asyncio.gather( + env.exec('python3 -c ' + shlex.quote(wait), timeout_sec=4), + env.exec('touch /workspace/gate', timeout_sec=4)) + self.assertEqual((first.return_code, first.stdout.strip(), second.return_code), (0, 'ready', 0)) + user = await env.exec('id -u', user='root') + self.assertEqual(user.stdout.strip(), '10001') + data = root / 'fixture.txt' + data.write_text('synthetic transfer') + await env.upload_file(data, '/workspace/sealed.txt') + unsealed = module.DockerEnvironment(name, bootstrap) + await unsealed.upload_file(data, '/workspace/streamed.txt') + await unsealed.upload_dir(root, '/workspace/tree') + await env.upload_dir(root, '/workspace/sealed-tree') + await unsealed.upload_file(data, '/usr/local/bin/frontier-fixture') + await unsealed.upload_dir(root, '/opt/frontier-fixture') + result = await env.exec('cat /workspace/sealed.txt /workspace/streamed.txt ' + '/workspace/tree/fixture.txt /workspace/sealed-tree/fixture.txt ' + '/usr/local/bin/frontier-fixture /opt/frontier-fixture/fixture.txt') + self.assertEqual(result.return_code, 0) + self.assertEqual(result.stdout, 'synthetic transfer' * 6) + with tempfile.TemporaryDirectory() as temporary: + asyncio.run(check(Path(temporary))) + finally: + subprocess.run(['docker', 'rm', '-f', name], capture_output=True, timeout=15) + + +def main(): + global SOURCE, DOCKER_IMAGE + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--source', required=True, type=Path) + parser.add_argument('--docker-image', help='already-local image with python3 and /bin/bash') + args = parser.parse_args() + SOURCE = args.source.resolve() + DOCKER_IMAGE = args.docker_image + # The fixtures run against a copy; do not write bytecode into a frozen source. + sys.dont_write_bytecode = True + suite = unittest.TestSuite([unittest.defaultTestLoader.loadTestsFromTestCase(Compatibility), + unittest.defaultTestLoader.loadTestsFromTestCase(DockerTransport)]) + result = unittest.TextTestRunner(verbosity=2).run(suite) + return 0 if result.wasSuccessful() else 1 + + +if __name__ == '__main__': + raise SystemExit(main())