diff --git a/.gitattributes b/.gitattributes
index 27961f90ea..7d61c379ae 100644
--- a/.gitattributes
+++ b/.gitattributes
@@ -1,5 +1,6 @@
 # Preserve the immutable archive's bytes even with core.autocrlf=true.
 benchmark/deepswe-gptxhigh-v1/** text eol=lf
+benchmark/frontiercs2-single-host-v1/** text eol=lf
 
 # Published study archives bind sha256 of their own LF bytes in index.json, so
 # a CRLF checkout must not rewrite them either.
diff --git a/benchmark/README.md b/benchmark/README.md
index f4bb9b95ea..d0fc1fb3ef 100644
--- a/benchmark/README.md
+++ b/benchmark/README.md
@@ -133,6 +133,11 @@ preserved for inspection; the companion notice governs their current evidence
 status. Neither archive category is the architecture for new work or grants
 runner, verifier, submission, or publication authority.
 
+The [FrontierCS-2 single-host v1 archive note](frontiercs2-single-host-versions.md)
+registers the frozen native-client, HTTP bridge, Docker transport and verifier
+wrapper used by a one-task four-arm pilot. The companion note links the separate
+data release, dependencies, read-only identity check and qualification limits.
+
 ## Matched model provider routing
 
 For a custom Responses-compatible model endpoint, set `OPENAI_BASE_URL` and
diff --git a/benchmark/check_frontiercs2_v1.py b/benchmark/check_frontiercs2_v1.py
new file mode 100644
index 0000000000..7c6a642b3a
--- /dev/null
+++ b/benchmark/check_frontiercs2_v1.py
@@ -0,0 +1,77 @@
+#!/usr/bin/env python3
+"""Verify frozen compatibility sources without importing or executing them."""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import hashlib
+import json
+from pathlib import Path
+
+
+ARCHIVE = Path(__file__).resolve().with_name("frontiercs2-single-host-v1")
+EXPECTED = {
+    "docker_environment.py": "d6e22a5f894e801d3736657a69b700f16b83f008240235121f5ff40361914b64",
+    "frontier_submit.py": "f4cbc551acd10a729e5cade5cc3a70b653d9aee1dee4b1cbfeb20b3208df6f00",
+    "local_runner.py": "5f407c8c9d800eb110cc453a9815b571861d0d2fca6412ffe1c83e9918938cad",
+    "source-manifest.json": "d0c096b7e1b67630174990f9813e16f7904d27b11be9edf7892c9fe8ae579cbc",
+    "submission_bridge.py": "cd1be73411bbb94f1a7ee0df9e49f0d26aa5d6e23412be233c5e483a1728b639",
+}
+
+
+def check_archive(archive: Path) -> dict:
+    observed = set()
+    python_count = embedded_count = 0
+    for path in archive.iterdir():
+        if path.name == "__pycache__" and path.is_dir() and not path.is_symlink():
+            if any(p.is_symlink() or not p.is_file() or p.suffix != ".pyc" for p in path.iterdir()):
+                raise ValueError("unexpected non-bytecode cache entry")
+            continue
+        if path.is_symlink() or not path.is_file() or path.name not in EXPECTED:
+            raise ValueError(f"unexpected archive entry: {path.name}")
+        if path.stat().st_mode & 0o111:
+            raise ValueError(f"executable mode changed: {path.name}")
+        raw = path.read_bytes()
+        if hashlib.sha256(raw).hexdigest() != EXPECTED[path.name]:
+            raise ValueError(f"archive bytes changed: {path.name}")
+        observed.add(path.name)
+        if path.suffix == ".py":
+            compile(raw, path.name, "exec")
+            python_count += 1
+            for node in ast.parse(raw).body:
+                if (isinstance(node, ast.Assign) and isinstance(node.value, ast.Constant)
+                        and isinstance(node.value.value, str)
+                        and any(isinstance(t, ast.Name) and t.id == "NNP_BOOTSTRAP" for t in node.targets)):
+                    compile(node.value.value, f"{path.name}:NNP_BOOTSTRAP", "exec")
+                    embedded_count += 1
+        else:
+            json.loads(raw)
+    if observed != set(EXPECTED):
+        raise ValueError("archive source files are missing")
+    return {
+        "original_snapshot_preserved": True,
+        "files_checked": len(observed),
+        "python_syntax_checks": python_count,
+        "embedded_python_checks": embedded_count,
+        "benchmark_executed": False,
+        "standalone_runnable": False,
+        "runtime_validation": "not_performed",
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--archive", type=Path, default=ARCHIVE)
+    args = parser.parse_args()
+    try:
+        result = check_archive(args.archive)
+    except (OSError, ValueError, SyntaxError) as error:
+        print(json.dumps({"ok": False, "error": str(error)}))
+        return 1
+    print(json.dumps({"ok": True, **result}, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/benchmark/frontiercs2-single-host-v1/docker_environment.py b/benchmark/frontiercs2-single-host-v1/docker_environment.py
new file mode 100644
index 0000000000..3ab1b2bf1c
--- /dev/null
+++ b/benchmark/frontiercs2-single-host-v1/docker_environment.py
@@ -0,0 +1,76 @@
+"""Private Docker transport for the existing shared benchmark installer."""
+import io
+from pathlib import Path
+import shlex
+import subprocess
+import tarfile
+from types import SimpleNamespace
+
+
+class DockerEnvironment:
+    default_user = '10001:10001'
+
+    def __init__(self, name, security_bootstrap, *, sealed=False):
+        self.name = name
+        self.security_bootstrap = security_bootstrap
+        self.sealed = sealed
+
+    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]
+        if cwd:
+            argv += ['--workdir', cwd]
+        for key, value in (env or {}).items():
+            argv += ['--env', key + '=' + str(value)]
+        # Trusted installation runs as root. Every solver-side process enters
+        # the verified no-new-privileges bootstrap before executing its argv.
+        if not self.sealed and user in ('root', '0', '0:0', 0):
+            argv += [self.name, '/bin/bash', '-c', command]
+        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)
+        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)
+
+    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,
+                 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)
+            if result.returncode:
+                raise RuntimeError(result.stderr.decode(errors='replace')[-1500:])
+            return
+        # Snap Docker has its own /tmp. Stream archive bytes rather than
+        # asking its client to open a host temporary pathname.
+        stream = io.BytesIO()
+        with tarfile.open(fileobj=stream, mode='w') as archive:
+            def owned(info):
+                info.uid = info.gid = 10001
+                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)
+        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')
+        if result.return_code:
+            raise RuntimeError(result.stderr)
+        stream = io.BytesIO()
+        with tarfile.open(fileobj=stream, mode='w') as archive:
+            def owned(info):
+                info.uid = info.gid = 10001
+                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 result.returncode:
+            raise RuntimeError(result.stderr.decode(errors='replace')[-1500:])
diff --git a/benchmark/frontiercs2-single-host-v1/frontier_submit.py b/benchmark/frontiercs2-single-host-v1/frontier_submit.py
new file mode 100644
index 0000000000..23c9aa631b
--- /dev/null
+++ b/benchmark/frontiercs2-single-host-v1/frontier_submit.py
@@ -0,0 +1,274 @@
+#!/usr/bin/env python3
+"""Submit task-defined tar archives to the run-local Frontier Driver."""
+
+from __future__ import annotations
+
+import uuid
+import argparse
+import hashlib
+import http.client
+import json
+import os
+from pathlib import Path, PurePosixPath
+import re
+import stat
+import sys
+import tarfile
+import time
+from urllib.parse import urlsplit
+
+
+SCHEMA = "frontier.candidate-submit-result/v1"
+TOKEN_LIMIT = 16 * 1024
+RESPONSE_LIMIT = 64 * 1024 * 1024
+REQUEST_ATTEMPTS = 16
+MAX_MEMBERS = 100_000
+FINAL_RECEIPT = Path("/workspace/.frontier/final-submission.json")
+TERMINAL = {"passed", "failed"}
+PENDING = {"queued", "running"}
+VERIFICATION_ID = re.compile(r"verification_[A-Za-z0-9]{1,64}")
+
+
+class SubmitError(RuntimeError):
+    pass
+
+
+def validate_tar(path: Path) -> tuple[int, str]:
+    if not path.is_absolute():
+        raise SubmitError("archive path must be absolute")
+    metadata = path.lstat()
+    if not stat.S_ISREG(metadata.st_mode) or metadata.st_size <= 0:
+        raise SubmitError("archive must be a nonempty regular file")
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    try:
+        with tarfile.open(path, mode="r:") as archive:
+            members = archive.getmembers()
+            if not members or len(members) > MAX_MEMBERS:
+                raise SubmitError("archive has an invalid member count")
+            seen: dict[str, bool] = {}
+            for member in members:
+                name = PurePosixPath(member.name)
+                normalized = name.as_posix()
+                if (
+                    not normalized
+                    or normalized == "."
+                    or member.name.startswith("/")
+                    or "//" in member.name
+                    or ".." in name.parts
+                    or member.isdev()
+                    or member.isfifo()
+                ):
+                    raise SubmitError(f"unsafe archive member: {member.name}")
+                if normalized in seen:
+                    raise SubmitError(f"duplicate archive member: {member.name}")
+                for parent in name.parents:
+                    parent_name = parent.as_posix()
+                    if parent_name != "." and parent_name in seen and not seen[parent_name]:
+                        raise SubmitError(f"conflicting archive member: {member.name}")
+                if not member.isdir() and any(existing.startswith(f"{normalized}/") for existing in seen):
+                    raise SubmitError(f"conflicting archive member: {member.name}")
+                seen[normalized] = member.isdir()
+                if member.issym() or member.islnk():
+                    target = PurePosixPath(member.linkname)
+                    if member.linkname.startswith("/") or ".." in target.parts:
+                        raise SubmitError(f"unsafe archive link: {member.name}")
+    except tarfile.TarError as error:
+        raise SubmitError("archive must be an uncompressed POSIX tar") from error
+    return metadata.st_size, f"sha256:{digest.hexdigest()}"
+
+
+def token() -> str:
+    path = Path(os.environ.get("FRONTIER_DRIVER_TOKEN_FILE", ""))
+    if not path.is_absolute():
+        raise SubmitError("FRONTIER_DRIVER_TOKEN_FILE must be absolute")
+    value = path.read_bytes()
+    if not value or len(value) > TOKEN_LIMIT:
+        raise SubmitError("Driver token file has an invalid size")
+    return value.decode("utf-8").strip()
+
+
+class Client:
+    def __init__(self) -> None:
+        parsed = urlsplit(os.environ.get("FRONTIER_DRIVER_API_URL", ""))
+        if parsed.scheme != "http" or not parsed.hostname or parsed.path not in ("", "/"):
+            raise SubmitError("FRONTIER_DRIVER_API_URL must be a plain HTTP origin")
+        self.host = parsed.hostname
+        self.port = parsed.port or 80
+        self.auth = token()
+
+    def request(self, method: str, path: str, *, body: bytes = b"", headers: dict[str, str] | None = None) -> dict[str, object]:
+        request_headers = {"Authorization": f"Bearer {self.auth}"}
+        request_headers.update(headers or {})
+        for attempt in range(REQUEST_ATTEMPTS):
+            connection = http.client.HTTPConnection(self.host, self.port, timeout=30)
+            try:
+                connection.request(method, path, body=body, headers=request_headers)
+                response = connection.getresponse()
+                data = response.read(RESPONSE_LIMIT + 1)
+                if len(data) > RESPONSE_LIMIT:
+                    raise SubmitError("Driver response exceeded its size limit")
+                if response.status in {408, 429, 500, 502, 503, 504} and attempt + 1 < REQUEST_ATTEMPTS:
+                    time.sleep(min(10, 0.25 * (2 ** attempt)))
+                    continue
+                if response.status < 200 or response.status >= 300:
+                    raise SubmitError(f"Driver returned HTTP {response.status}")
+                value = json.loads(data)
+                if not isinstance(value, dict):
+                    raise SubmitError("Driver response must be a JSON object")
+                return value
+            except (OSError, http.client.HTTPException) as error:
+                if attempt + 1 == REQUEST_ATTEMPTS:
+                    raise SubmitError("Driver request failed") from error
+                time.sleep(min(10, 0.25 * (2 ** attempt)))
+            finally:
+                connection.close()
+        raise SubmitError("Driver request failed")
+
+
+def canonical(value: object) -> bytes:
+    return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8")
+
+
+def idempotency(operation: str, digest: str) -> str:
+    return f"frontier-submit-{operation}-{digest.removeprefix('sha256:')}"
+
+
+def upload(client: Client, operation: str, path: Path) -> tuple[str, str]:
+    size, digest = validate_tar(path)
+    value = client.request("POST", "/v1/payloads", body=path.read_bytes(), headers={
+        "Content-Type": "application/x-tar", "Content-Length": str(size),
+        "X-Content-SHA256": digest, "X-Frontier-Purpose": operation,
+        "Idempotency-Key": idempotency(f"payload-{operation}", digest),
+    })
+    payload_id = value.get("payload_id")
+    if not isinstance(payload_id, str) or not payload_id.startswith("payload_"):
+        raise SubmitError("Driver payload response is invalid")
+    return payload_id, digest
+
+
+def train_result(view: dict[str, object], *, payload_id: str, verification_id: str,
+                 remaining: int) -> tuple[dict[str, object], int]:
+    """The Agent-facing JSON and exit status for one train verification view."""
+    state = view.get("state")
+    if state not in TERMINAL and state not in PENDING:
+        raise SubmitError("Driver train response has an invalid state")
+    result: dict[str, object] = {
+        "schema_version": SCHEMA, "operation": "train", "state": state,
+        "payload_id": payload_id, "verification_id": verification_id,
+        "remaining_train_calls": remaining,
+    }
+    if state in PENDING:
+        return result, 0
+    metrics = view.get("metrics", {})
+    if not isinstance(metrics, dict):
+        raise SubmitError("Driver train response has invalid metrics")
+    result["metrics"] = metrics
+    failure = view.get("failure")
+    if state == "failed" and failure is not None:
+        if (not isinstance(failure, dict)
+                or not isinstance(failure.get("code"), str)
+                or not isinstance(failure.get("message"), str)
+                or not isinstance(failure.get("retryable"), bool)):
+            raise SubmitError("Driver train response has invalid failure feedback")
+        result["failure"] = {key: failure[key] for key in ("code", "message", "retryable")}
+
+    if isinstance(view.get("raw_score"), (int, float)):
+        result["raw_score"] = view["raw_score"]
+    return result, 0 if state == "passed" else 2
+
+
+def run_train(client: Client, archive: Path) -> tuple[dict[str, object], int]:
+    payload_id, digest = upload(client, "train", archive)
+    view = client.request("POST", "/v1/verifier-train", body=canonical({"payload_id": payload_id}), headers={
+        "Content-Type": "application/json",
+        "Idempotency-Key": idempotency("train", digest),
+    })
+    verification_id = view.get("verification_id")
+    remaining = view.get("remaining_train_calls")
+    if not isinstance(verification_id, str) or not isinstance(remaining, int) or remaining < 0:
+        raise SubmitError("Driver train response lacks remaining_train_calls")
+    return train_result(view, payload_id=payload_id, verification_id=verification_id, remaining=remaining)
+
+
+def run_status(client: Client, verification_id: str) -> tuple[dict[str, object], int]:
+    """Read one train verification once, in the same JSON shape as `train`."""
+    if not VERIFICATION_ID.fullmatch(verification_id):
+        raise SubmitError("invalid verification id")
+    view = client.request("GET", f"/v1/verifications/{verification_id}")
+    if view.get("verification_id") != verification_id or view.get("mode") != "train":
+        raise SubmitError("status reports train verifications only")
+    payload_id = view.get("payload_id")
+    remaining = view.get("remaining_train_calls")
+    if not isinstance(payload_id, str) or type(remaining) is not int or remaining < 0:
+        raise SubmitError("Driver verification response lacks payload_id or remaining_train_calls")
+    return train_result(view, payload_id=payload_id, verification_id=verification_id, remaining=remaining)
+
+
+def run_store_final(client: Client, archive: Path) -> dict[str, object]:
+    payload_id, _ = upload(client, "final", archive)
+    # One intent per invocation, retained across the Client's transport retries.
+    # A deliberate later invocation may select the same tar again.
+    view = client.request("POST", "/v1/submissions/checkpoint", body=canonical({"payload_id": payload_id}), headers={
+        "Content-Type": "application/json",
+        "Idempotency-Key": f"frontier-store-final-{uuid.uuid4()}",
+    })
+    runtime = view.get("remaining_runtime_s")
+    remaining = view.get("remaining_checkpoint_submissions")
+    if (type(runtime) is not int or runtime < 0 or "remaining_checkpoint_submissions" not in view
+            or (remaining is not None and (type(remaining) is not int or remaining < 0))):
+        raise SubmitError("Driver store-final response has invalid remaining runtime or submission count")
+    return {"schema_version": SCHEMA, "operation": "store-final", "state": "stored",
+            "payload_id": payload_id, "remaining_runtime_s": runtime,
+            "remaining_checkpoint_submissions": remaining}
+
+
+def run_final(client: Client, archive: Path) -> dict[str, object]:
+    payload_id, digest = upload(client, "final", archive)
+    view = client.request("POST", "/v1/submissions", body=canonical({"payload_id": payload_id}), headers={
+        "Content-Type": "application/json",
+        "Idempotency-Key": idempotency("final", digest),
+    })
+    verification_id = view.get("verification_id")
+    if not isinstance(verification_id, str):
+        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())
+    return result
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    operations = parser.add_subparsers(dest="operation", required=True)
+    train = operations.add_parser("train", help="request train feedback for a tar")
+    train.add_argument("archive", type=Path)
+    for name in ("final", "store-final"):
+        operations.add_parser(name).add_argument("archive", type=Path)
+    operations.add_parser("status", help="read one train verification").add_argument("verification_id")
+    args = parser.parse_args(argv)
+    try:
+        client = Client()
+        if args.operation == "train":
+            result, status = run_train(client, args.archive)
+        elif args.operation == "status":
+            result, status = run_status(client, args.verification_id)
+        elif args.operation == "store-final":
+            result, status = run_store_final(client, args.archive), 0
+        else:
+            result, status = run_final(client, args.archive), 0
+        sys.stdout.buffer.write(canonical(result))
+        return status
+    except (SubmitError, OSError, UnicodeError, json.JSONDecodeError, ValueError) as error:
+        print(f"frontier-submit: ERROR: {error}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/benchmark/frontiercs2-single-host-v1/local_runner.py b/benchmark/frontiercs2-single-host-v1/local_runner.py
new file mode 100644
index 0000000000..bf6ae9f831
--- /dev/null
+++ b/benchmark/frontiercs2-single-host-v1/local_runner.py
@@ -0,0 +1,287 @@
+"""Private single-host qualification using unchanged FrontierCS-2 images.
+
+This drives the native verifier directly. It is not Management, Driver, a
+frontier-submit API implementation, or a scored Agent experiment.
+"""
+import argparse
+import base64
+import datetime
+import hashlib
+import io
+import json
+import os
+from pathlib import Path
+import signal
+import subprocess
+import sys
+import tarfile
+import time
+
+ROOT = Path(__file__).resolve().parent
+MANIFEST = json.loads((ROOT / 'source-manifest.json').read_text())
+TASK = MANIFEST['task']
+ACTIVE = set()
+NNP_BOOTSTRAP = '''import ctypes,json,os,pathlib,sys
+libc=ctypes.CDLL(None,use_errno=True)
+if libc.prctl(38,1,0,0,0)!=0:
+ raise OSError(ctypes.get_errno(),"PR_SET_NO_NEW_PRIVS failed")
+status=dict(line.split(":",1) for line in pathlib.Path("/proc/self/status").read_text().splitlines() if ":" in line)
+proof={key:status[key].strip() for key in ["NoNewPrivs","CapEff","Seccomp"]}
+proof["uid"]=os.getuid()
+proof["apparmor"]=pathlib.Path("/proc/self/attr/current").read_text().strip()
+assert proof["NoNewPrivs"]=="1" and int(proof["CapEff"],16)==0 and proof["Seccomp"]=="2"
+assert "docker-default" in proof["apparmor"]
+if os.environ.get("FRONTIER_LOCAL_SECURITY_RECEIPT"):
+ pathlib.Path(os.environ["FRONTIER_LOCAL_SECURITY_RECEIPT"]).write_text(json.dumps(proof))
+os.execv(sys.argv[1],sys.argv[1:])
+'''
+
+
+def digest(data):
+    return 'sha256:' + hashlib.sha256(data).hexdigest()
+
+
+def write_json(path, value):
+    temporary = path.with_suffix('.tmp')
+    temporary.write_text(json.dumps(value, indent=2, allow_nan=False))
+    temporary.replace(path)
+
+
+def image(key):
+    value = TASK[key]
+    return value['repository'] + '@' + value['digest']
+
+
+def cleanup():
+    for name in list(ACTIVE):
+        subprocess.run(['docker', 'rm', '-f', name], capture_output=True, timeout=20)
+        ACTIVE.discard(name)
+
+
+def interrupted(signum, frame):
+    cleanup()
+    raise SystemExit(128 + signum)
+
+
+def base_container(name, key, uid, memory):
+    # All timed implementations share one fixed logical CPU. The workspace
+    # tmpfs is bounded independently; its pages also count toward memory.
+    return ['docker', 'run', '--name', name, '--pull=never', '--network=none',
+            '--read-only', '--cap-drop=ALL',
+            '--cpus=1', '--cpuset-cpus=' + str(int(os.environ.get('FRONTIER_LOCAL_GRADER_CPU',max(os.sched_getaffinity(0))))),
+            '--memory=' + memory, '--memory-swap=' + memory, '--pids-limit=512',
+            '--user=' + str(uid) + ':' + str(uid)]
+
+
+def prepare_workspace():
+    reference = ROOT / 'operator-only/reference.patch'
+    assert digest(reference.read_bytes()) == TASK['reference']['payload_sha256']
+    name = 'fcs2-blc-workspace-20261102-r02'
+    code = r'''
+import subprocess,pathlib,json,hashlib,base64,sys
+def run(args):
+ r=subprocess.run(args,capture_output=True,text=True,timeout=120)
+ return r
+r=run(['/frontier/task/prepare_workspace.sh'])
+assert r.returncode==0,r.stderr
+empty=run(['/frontier/task/make_submission.sh'])
+assert empty.returncode==2 and 'empty' in empty.stderr,empty.stderr
+r=run(['git','-C','/workspace/repo','apply','/qualification/reference.patch'])
+assert r.returncode==0,r.stderr
+r=run(['/frontier/task/make_submission.sh'])
+assert r.returncode==0,r.stderr
+r=run(['tar','-C','/workspace','--format=ustar','--sort=name',
+ '--mtime=UTC 1970-01-01','--owner=0','--group=0','-cf','/workspace/generated.tar','payload'])
+assert r.returncode==0,r.stderr
+assert not pathlib.Path('/opt/frontier/task/evaluator.py').exists()
+assert not pathlib.Path('/frontier/task/judge').exists()
+body=pathlib.Path('/workspace/generated.tar').read_bytes()
+patch=pathlib.Path('/workspace/payload/solution.patch').read_bytes()
+assert 0<len(body)<=524288
+status=dict(l.split(':',1) for l in pathlib.Path('/proc/self/status').read_text().splitlines() if ':' in l)
+assert status['NoNewPrivs'].strip()=='1' and int(status['CapEff'].strip(),16)==0
+print(json.dumps({'empty_patch_rejected':True,'task_image_has_no_judge':True,
+ 'no_new_privs':status['NoNewPrivs'].strip(),'effective_capabilities':status['CapEff'].strip(),
+ 'patch_sha256':'sha256:'+hashlib.sha256(patch).hexdigest(),
+ 'archive_sha256':'sha256:'+hashlib.sha256(body).hexdigest(),
+ 'archive_base64':base64.b64encode(body).decode()}))
+'''
+    cmd = base_container(name, 'agent_image', 10001, '4g') + [
+        '--tmpfs', '/workspace:rw,exec,nosuid,nodev,size=4294967296,uid=10001,gid=10001,mode=0700',
+        '--mount', 'type=bind,src=' + str(reference) + ',dst=/qualification/reference.patch,readonly',
+        '-i', '--entrypoint=/usr/bin/python3', image('agent_image'),
+        '-c', NNP_BOOTSTRAP, '/usr/bin/python3', '-']
+    ACTIVE.add(name)
+    try:
+        result = subprocess.run(cmd, input=code.encode(), capture_output=True, timeout=180)
+        (ROOT / 'logs/workspace.stderr').write_bytes(result.stderr)
+        assert result.returncode == 0, result.stderr.decode(errors='replace')[-2500:]
+        receipt = json.loads(result.stdout)
+        archive = base64.b64decode(receipt.pop('archive_base64'))
+        (ROOT / 'inputs/generated.tar').write_bytes(archive)
+        receipt['matches_frozen_reference_patch'] = receipt['patch_sha256'] == TASK['reference']['payload_sha256']
+        write_json(ROOT / 'workspace-receipt.json', receipt)
+        return receipt
+    finally:
+        cleanup()
+
+
+def make_archive(path, patch, symlink=False):
+    with tarfile.open(path, 'w', format=tarfile.USTAR_FORMAT) as archive:
+        directory = tarfile.TarInfo('payload')
+        directory.type = tarfile.DIRTYPE
+        directory.mode = 0o755
+        archive.addfile(directory)
+        member = tarfile.TarInfo('payload/solution.patch')
+        member.mode = 0o644
+        if symlink:
+            member.type = tarfile.SYMTYPE
+            member.linkname = '/etc/passwd'
+            archive.addfile(member)
+        else:
+            member.size = len(patch)
+            archive.addfile(member, io.BytesIO(patch))
+    assert path.stat().st_size <= 524288
+
+
+def decode_evidence(path):
+    code = ('import io,json,tarfile,zstandard; '
+            'raw=zstandard.ZstdDecompressor().decompress(open("/evidence.tar.zst","rb").read()); '
+            't=tarfile.open(fileobj=io.BytesIO(raw)); '
+            'print(t.extractfile("metrics.json").read().decode())')
+    result = subprocess.run([
+        'docker', 'run', '--rm', '--pull=never', '--network=none', '--read-only',
+        '--cap-drop=ALL', '--memory=256m', '--cpus=1',
+        '--mount', 'type=bind,src=' + str(path) + ',dst=/evidence.tar.zst,readonly',
+        '--entrypoint=/usr/bin/python3', image('verifier_image'),
+        '-c', NNP_BOOTSTRAP, '/opt/frontier/runtime-python', '-c', code],
+        capture_output=True, timeout=30, check=True)
+    return json.loads(result.stdout)
+
+
+def verify(archive, mode, label):
+    assert mode in ('train', 'eval')
+    dataset = ROOT / ('inputs/' + mode + '.json')
+    assert digest(dataset.read_bytes()) == TASK['datasets'][mode + '_sha256']
+    out = ROOT / 'runs' / label
+    out.mkdir(mode=0o700)
+    os.chown(out, 65532, 65532)
+    name = 'fcs2-blc-' + label + '-20261102-r02'
+    cmd = base_container(name, 'verifier_image', 65532, '4g') + [
+        '--tmpfs', '/frontier:rw,exec,nosuid,nodev,size=4294967296,uid=65532,gid=65532,mode=0700',
+        '--mount', 'type=bind,src=' + str(archive.resolve()) + ',dst=/inputs/payload.tar,readonly',
+        '--mount', 'type=bind,src=' + str(dataset) + ',dst=/inputs/dataset.json,readonly',
+        '--mount', 'type=bind,src=' + str(out) + ',dst=/out',
+        '-e', 'FRONTIER_PAYLOAD_PATH=/inputs/payload.tar',
+        '-e', 'FRONTIER_DATASET_PATH=/inputs/dataset.json',
+        '-e', 'FRONTIER_VERIFICATION_MODE=' + mode,
+        '-e', 'FRONTIER_RESULT_PATH=/out/result.json',
+        '-e', 'FRONTIER_EVIDENCE_PATH=/out/evidence.tar.zst',
+        '-e', 'FRONTIER_LOCAL_SECURITY_RECEIPT=/out/launch-security.json',
+        '--entrypoint=/usr/bin/python3', image('verifier_image'),
+        '-c', NNP_BOOTSTRAP, '/opt/frontier/verifier']
+    write_json(out / 'invocation.json', {'argv': cmd, 'payload_sha256': digest(archive.read_bytes()),
+                                        'dataset_sha256': digest(dataset.read_bytes())})
+    started = time.monotonic()
+    ACTIVE.add(name)
+    try:
+        with (out / 'process.log').open('wb') as log:
+            process = subprocess.run(cmd, stdout=log, stderr=subprocess.STDOUT, timeout=1800)
+        inspect = subprocess.run(['docker', 'inspect', name], capture_output=True, check=True, timeout=15)
+        container = json.loads(inspect.stdout)[0]
+        write_json(out / 'container.json', {k: container[k] for k in ['Id', 'State', 'HostConfig', 'Image']})
+        result = json.loads((out / 'result.json').read_text())
+        row = {'label': label, 'mode': mode, 'process_exit_code': process.returncode,
+               'elapsed_seconds': round(time.monotonic() - started, 3),
+               'result': result, 'oom_killed': container['State']['OOMKilled']}
+        if (out / 'evidence.tar.zst').exists():
+            evidence = decode_evidence(out / 'evidence.tar.zst')
+            write_json(out / 'operator-metrics.json', evidence)
+            row['evidence_sha256'] = digest((out / 'evidence.tar.zst').read_bytes())
+            row['operator_metrics'] = evidence['metrics']
+        write_json(out / 'receipt.json', row)
+        return row
+    finally:
+        cleanup()
+
+
+def qualify():
+    completed = []
+    def progress(stage):
+        write_json(ROOT / 'qualification-state.json', {'stage': stage, 'completed': completed,
+                   'observed_at': datetime.datetime.now(datetime.timezone.utc).isoformat()})
+    try:
+        assert json.loads((ROOT / 'pull-state.json').read_text())['stage'] == 'pulled'
+        progress('preparing_workspace')
+        workspace = prepare_workspace()
+        reference = ROOT / 'operator-only/reference.patch'
+        make_archive(ROOT / 'inputs/reference.tar', reference.read_bytes())
+        make_archive(ROOT / 'inputs/empty.tar', b'')
+        make_archive(ROOT / 'inputs/symlink.tar', b'', symlink=True)
+        forbidden = (b'diff --git a/README.md b/README.md\n--- a/README.md\n+++ b/README.md\n'
+                     b'@@ -1 +1 @@\n-# Batch Local Counting\n+# Forbidden qualification edit\n')
+        make_archive(ROOT / 'inputs/forbidden.tar', forbidden)
+        archive = ROOT / 'inputs/generated.tar'
+        # Retain the authored byte identity for exact-reference deduplication.
+        # If native diff formatting differs, verify both paths explicitly.
+        for mode in ['train', 'eval']:
+            label = 'reference-' + mode
+            progress(label)
+            row = verify(ROOT / 'inputs/reference.tar', mode, label)
+            completed.append(row)
+            metrics = row.get('operator_metrics', {})
+            assert row['process_exit_code'] == 0 and row['result']['status'] == 'passed', row
+            assert metrics.get('valid_patch') == 1, row
+            assert len(metrics.get('beat_spec_workload_passes', [])) == 6, row
+            assert all(r['metrics']['exact_counts'] for r in metrics['beat_spec_workload_passes']), row
+            assert metrics['beats_reference'] == 0, 'Exact reference must not beat itself'
+            assert metrics['margin'] == 0, 'Deduplicated reference scores must match'
+            if mode == 'train':
+                assert 'beat_spec_progress' in row['result']['metrics'], row
+                assert 'private_metrics' not in row['result'], row
+            else:
+                assert 'beat_spec_progress' not in row['result']['metrics'], row
+        if not workspace['matches_frozen_reference_patch']:
+            progress('generated-train')
+            row = verify(archive, 'train', 'generated-train')
+            completed.append(row)
+            assert row['process_exit_code'] == 0 and row.get('operator_metrics', {}).get('valid_patch') == 1, row
+            assert all(r['metrics']['exact_counts'] for r in row['operator_metrics']['beat_spec_workload_passes']), row
+        for label in ['empty', 'symlink', 'forbidden']:
+            progress('negative-' + label)
+            row = verify(ROOT / ('inputs/' + label + '.tar'), 'train', 'negative-' + label)
+            completed.append(row)
+            if label in ['empty', 'symlink']:
+                assert row['process_exit_code'] != 0 and row['result']['status'] == 'failed', row
+                assert row['result']['failure']['code'] == 'invalid_submission', row
+            else:
+                assert row['process_exit_code'] == 0 and row['result']['status'] == 'passed', row
+                assert row['result']['raw_score'] == 0 and row['operator_metrics'].get('valid_patch') != 1, row
+                assert all(not x['metrics']['exact_counts'] for x in row['operator_metrics']['beat_spec_workload_passes']), row
+        progress('qualified')
+    except BaseException as error:
+        write_json(ROOT / 'qualification-state.json', {'stage': 'failed', 'completed': completed,
+                   'error': type(error).__name__ + ': ' + str(error)[:3500]})
+        raise
+
+
+if __name__ == '__main__':
+    signal.signal(signal.SIGTERM, interrupted)
+    signal.signal(signal.SIGINT, interrupted)
+    parser = argparse.ArgumentParser()
+    sub = parser.add_subparsers(dest='command', required=True)
+    sub.add_parser('qualify')
+    grading = sub.add_parser('verify')
+    grading.add_argument('--archive', type=Path, required=True)
+    grading.add_argument('--mode', choices=['train', 'eval'], required=True)
+    grading.add_argument('--label', required=True)
+    args = parser.parse_args()
+    try:
+        if args.command == 'qualify':
+            qualify()
+        else:
+            if not args.label.replace('-', '').isalnum():
+                raise SystemExit('label must contain only letters, numbers and hyphens')
+            print(json.dumps(verify(args.archive, args.mode, args.label), indent=2))
+    finally:
+        cleanup()
diff --git a/benchmark/frontiercs2-single-host-v1/source-manifest.json b/benchmark/frontiercs2-single-host-v1/source-manifest.json
new file mode 100644
index 0000000000..c3c1b36b57
--- /dev/null
+++ b/benchmark/frontiercs2-single-host-v1/source-manifest.json
@@ -0,0 +1,62 @@
+{
+  "source_commit": "d247b3f15b73805e455ff93c6d043d84603ec341",
+  "task": {
+    "agent_image": {
+      "digest": "sha256:023cf56e4be5ca17fa3ac58d30bac45b3a765bcbf32635bb900d9051efef762d",
+      "repository": "docker.io/joyemang33/fcs2-task-batch_local_counting-agent",
+      "tag": "source-563816fc066098b1af7c91b97484322e2960f503e3c1c2f33536df1d46f6b747"
+    },
+    "baseline": {
+      "agreement_rule": {
+        "abs": 0.01,
+        "name": "cross-site-agreement/v1",
+        "rel": 0.05,
+        "replays": 1
+      },
+      "comparability_note": "result.json does not attest that these scores measured the exported input identity",
+      "comparable": false,
+      "datasets": {
+        "eval": "sha256:b1c6e93f03c972a4bf63d54b453a0e8519efaa51651a4f572d7334d173a501d7",
+        "train": "sha256:192f72eac119f17c4f866fe46e653d3ccc730e7c848ebed35e77c06461847868"
+      },
+      "evaluator_sha256": "sha256:cc0848f4d810f76319a3ef656281c0ce39953c03703ac36507f682ac4afad507",
+      "historical_mean": 12.805318426021563,
+      "historical_scores": [
+        12.500607571875047,
+        13.016857446054395,
+        13.011021729803632,
+        12.71547318684792,
+        12.782632195526816
+      ],
+      "historical_spread": 0.5162498741793478,
+      "provenance": "historical-authoring",
+      "reference_patch_sha256": "sha256:1cb3d6c02baec5454052d297fe7b29fed30982bce1fc3c605783676684eb0614",
+      "result_json_source": "judge/result.json",
+      "source_set_sha256": "sha256:563816fc066098b1af7c91b97484322e2960f503e3c1c2f33536df1d46f6b747",
+      "verifier_adapter_sha256": "sha256:e278eb9db3817ddc011c88b1af885c433b6c34f5d7a29d2eb93bd316013581a3"
+    },
+    "dataset_image": {
+      "digest": "sha256:b171b15d65c8b39e798361cf8ad5b35dcab279e3131623247718abcc6ed80a72",
+      "repository": "docker.io/joyemang33/fcs2-task-batch_local_counting-datasets",
+      "tag": "source-563816fc066098b1af7c91b97484322e2960f503e3c1c2f33536df1d46f6b747"
+    },
+    "datasets": {
+      "eval_sha256": "sha256:b1c6e93f03c972a4bf63d54b453a0e8519efaa51651a4f572d7334d173a501d7",
+      "train_sha256": "sha256:192f72eac119f17c4f866fe46e653d3ccc730e7c848ebed35e77c06461847868"
+    },
+    "name": "batch_local_counting",
+    "reference": {
+      "digest": "sha256:69837c04100abfd6817c4d9f6957b78423ad7a20f51c083d83a10a3e8b1eb060",
+      "launcher_sha256": "sha256:ce7234cdc7cd3a274df81e5d23cc6c96e38c2e10c5418511940928e98aa3e9ec",
+      "payload_sha256": "sha256:1cb3d6c02baec5454052d297fe7b29fed30982bce1fc3c605783676684eb0614",
+      "repository": "docker.io/joyemang33/fcs2-task-batch_local_counting-reference",
+      "tag": "source-563816fc066098b1af7c91b97484322e2960f503e3c1c2f33536df1d46f6b747"
+    },
+    "verifier_image": {
+      "digest": "sha256:2655a86ab2e228ac1ddb5ea9e47554f3d1178601ed6df85fb2be555df18b2ee9",
+      "repository": "docker.io/joyemang33/fcs2-task-verifier-batch_local_counting",
+      "tag": "source-563816fc066098b1af7c91b97484322e2960f503e3c1c2f33536df1d46f6b747"
+    }
+  },
+  "qualification": "native-local-reference-pipeline; no model calls or comparative claim"
+}
\ No newline at end of file
diff --git a/benchmark/frontiercs2-single-host-v1/submission_bridge.py b/benchmark/frontiercs2-single-host-v1/submission_bridge.py
new file mode 100644
index 0000000000..5c6fa8420a
--- /dev/null
+++ b/benchmark/frontiercs2-single-host-v1/submission_bridge.py
@@ -0,0 +1,188 @@
+"""Private native-client transport; scoring stays in the pinned verifier.
+
+This is a single-host experiment adapter, not the upstream Management/Driver
+implementation. Every accepted train request uses its real verifier output.
+"""
+import hashlib
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+from pathlib import Path
+import secrets
+import threading
+import time
+import uuid
+
+
+class Rejected(Exception):
+    def __init__(self, status, reason):
+        self.status, self.reason = status, reason
+
+
+class Trial:
+    def __init__(self, root, arm, queue):
+        self.root, self.arm, self.queue = Path(root), arm, queue
+        self.token = secrets.token_hex(32)
+        self.lock = threading.RLock()
+        self.deadline = None
+        self.phase = 'prepared'
+        self.payloads = {}
+        self.jobs = {}
+        self.idempotency = {}
+        self.train_calls = 0
+        self.train_wall_seconds = 0.0
+        self.saved = None
+        self.final = None
+        self.final_at = None
+        self.rejected_requests = 0
+        (self.root/'payloads').mkdir(parents=True, mode=0o700, exist_ok=True)
+
+    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,
+                '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 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)
+
+    def start(self, seconds):
+        with self.lock:
+            assert self.phase == 'prepared'
+            self.deadline = time.time()+seconds
+            self.phase = 'solving'
+            self.persist()
+
+    def admit(self):
+        if self.phase != 'solving' or self.deadline is None or time.time() >= self.deadline:
+            raise Rejected(409, 'answering phase is closed')
+
+    def view(self, job):
+        return {**job, 'remaining_train_calls': max(0, 100-self.train_calls)}
+
+    def request(self, method, path, body, headers):
+        with self.lock:
+            fingerprint = hashlib.sha256(method.encode()+path.encode()+body).hexdigest()
+            key = headers.get('Idempotency-Key')
+            if key and key in self.idempotency:
+                old, status, response = self.idempotency[key]
+                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()
+            return status, result
+
+    def _request(self, method, path, body, headers):
+        if method == 'GET' and path.startswith('/v1/verifications/'):
+            identifier = path.rsplit('/', 1)[-1]
+            job = self.jobs.get(identifier)
+            if not job or job['mode'] != 'train':
+                raise Rejected(404, 'train verification not found')
+            return 200, self.view(job)
+        self.admit()
+        if method == 'POST' and path == '/v1/payloads':
+            if not 0 < len(body) <= 524288:
+                raise Rejected(413, 'task submission size limit exceeded')
+            digest = 'sha256:'+hashlib.sha256(body).hexdigest()
+            if headers.get('X-Content-SHA256') != digest:
+                raise Rejected(400, 'payload digest mismatch')
+            purpose = headers.get('X-Frontier-Purpose')
+            if purpose not in ('train', 'final'):
+                raise Rejected(400, 'unsupported payload purpose')
+            identifier = 'payload_'+digest.split(':')[1]+'_'+purpose
+            if identifier not in self.payloads:
+                target = self.root/'payloads'/(identifier+'.tar')
+                target.write_bytes(body)
+                # The enclosing host directory remains private; the native
+                # unprivileged verifier receives this one file read-only.
+                target.chmod(0o444)
+                self.payloads[identifier] = {'path': str(target), 'sha256': digest, 'purpose': purpose}
+            return 201, {'payload_id': identifier}
+        try:
+            document = json.loads(body)
+            identifier = document['payload_id']
+            payload = self.payloads[identifier]
+        except (ValueError, KeyError, TypeError):
+            raise Rejected(400, 'unknown trial payload') from None
+        if method == 'POST' and path == '/v1/verifier-train':
+            if payload['purpose'] != 'train':
+                raise Rejected(400, 'payload purpose mismatch')
+            if self.train_calls >= 100 or self.train_wall_seconds >= 180000:
+                raise Rejected(429, 'authored train budget exhausted')
+            verification = 'verification_'+uuid.uuid4().hex
+            job = {'verification_id': verification, 'payload_id': identifier,
+                   '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':
+                raise Rejected(400, 'payload purpose mismatch')
+            # The latest explicitly chosen immutable snapshot wins. A rejected
+            # replacement never reaches this mutation.
+            self.saved = identifier
+            return 200, {'remaining_runtime_s': max(0, int(self.deadline-time.time())),
+                         'remaining_checkpoint_submissions': None}
+        if method == 'POST' and path == '/v1/submissions':
+            if payload['purpose'] != 'final':
+                raise Rejected(400, 'payload purpose mismatch')
+            verification = 'verification_'+uuid.uuid4().hex
+            self.final, self.final_at = identifier, time.time()
+            self.phase = 'final_selected'
+            self.jobs[verification] = {'verification_id': verification,
+                'payload_id': identifier, 'mode': 'eval', 'state': 'queued', 'metrics': {}}
+            return 202, {'verification_id': verification}
+        raise Rejected(404, 'unsupported operation')
+
+
+class Bridge(ThreadingHTTPServer):
+    daemon_threads = True
+
+    def __init__(self, address):
+        self.trials = {}
+        super().__init__(address, Handler)
+
+
+class Handler(BaseHTTPRequestHandler):
+    def log_message(self, fmt, *args):
+        pass
+
+    def do_GET(self):
+        self.handle_request()
+
+    def do_POST(self):
+        self.handle_request()
+
+    def handle_request(self):
+        trial = next((t for token, t in self.server.trials.items()
+                      if secrets.compare_digest(self.headers.get('Authorization', ''), 'Bearer '+token)), None)
+        try:
+            if trial is None:
+                raise Rejected(401, 'trial capability required')
+            length = int(self.headers.get('Content-Length', 0))
+            if length < 0 or length > 524288:
+                raise Rejected(413, 'task submission size limit exceeded')
+            body = self.rfile.read(length)
+            if len(body) != length:
+                raise Rejected(400, 'incomplete request body')
+            status, result = trial.request(self.command, self.path, body, self.headers)
+        except Rejected as error:
+            status, result = error.status, {'error': error.reason}
+            if trial:
+                trial.rejected_requests += 1
+        except (ValueError, OSError):
+            status, result = 400, {'error': 'malformed request'}
+        raw = json.dumps(result, separators=(',', ':'), allow_nan=False).encode()
+        self.send_response(status)
+        self.send_header('Content-Type', 'application/json')
+        self.send_header('Content-Length', str(len(raw)))
+        self.end_headers()
+        self.wfile.write(raw)
diff --git a/benchmark/frontiercs2-single-host-versions.md b/benchmark/frontiercs2-single-host-versions.md
new file mode 100644
index 0000000000..e18584403b
--- /dev/null
+++ b/benchmark/frontiercs2-single-host-versions.md
@@ -0,0 +1,115 @@
+# FrontierCS-2 single-host v1 compatibility archive
+
+This is the frozen compatibility-source snapshot behind a four-arm,
+one-task `batch_local_counting` pilot. It follows the
+[archive placement rules](README.md#archive-placement). It is an inert historical
+reference: no product imports, automatic CI execution, job launch or model call
+is added. Scores, submissions and usage observations are published separately in
+the [Hugging Face release](https://huggingface.co/datasets/songwells/loopxbench/tree/main/public/frontiercs2-batch-local-counting-20261102-v1).
+
+The study has one run per arm and a five-hour answering cap, with earlier final
+submissions. It uses the native verifier through a local single-host compatibility
+bridge, rather than the official Management/Driver deployment. Its terminal board
+still records integrity qualification as false, treatment fidelity as pending for
+the LoopX arms, and score countability as false. The published observations do
+not establish a qualified matched study, leaderboard result or LoopX benefit.
+
+## Source identity
+
+The evaluated LoopX source is `e9eda527afd5fcb262ee7c199e1822b1a936cc24`.
+The task export records source revision
+`d247b3f15b73805e455ff93c6d043d84603ec341`; its manifest binds the task,
+dataset, reference and verifier image digests. The image/source metadata is
+retained, while task inputs, reference patch and evaluator source are omitted.
+
+The [five frozen files](frontiercs2-single-host-v1/) retain the collected source
+bytes and non-executable Git modes. Their SHA-256 identities are independently
+pinned in [the read-only checker](check_frontiercs2_v1.py) and the HF release.
+Original docstrings saying “private” describe the experiment's former transport
+boundary; these selected source files are now released. The containing directory
+name changes, while the native verifier wrapper retains its original basename.
+
+| File | Historical role |
+| --- | --- |
+| `frontier_submit.py` | Candidate-side train, train-status, checkpoint and final client |
+| `submission_bridge.py` | Per-trial HTTP capability, immutable payloads, idempotency and grader queue |
+| `docker_environment.py` | Existing benchmark installer's Docker exec/upload transport, including streamed copies for Snap Docker |
+| `local_runner.py` | Operator-side native verifier invocation, resource limits and deferred no-new-privileges bootstrap |
+| `source-manifest.json` | Original exported image and input identities, including the non-comparable historical baseline notice |
+
+The campaign controller, authentication homes, private configuration, raw
+trajectories and verification logs are outside this source snapshot. No private
+host, account or local operating paths are needed to inspect it. A later repair
+must have a new version and validation; it cannot rewrite these historical bytes
+or inherit their score attribution.
+
+## Check the snapshot
+
+From the source checkout:
+
+```sh
+uv run --extra test python benchmark/check_frontiercs2_v1.py
+```
+
+The check verifies all five files' hashes and Git executable modes, rejects extra
+source files and symlinks, parses four Python modules and the JSON manifest, and
+compiles the named `NNP_BOOTSTRAP` string. Bytecode-only `__pycache__` is ignored.
+It imports none of the archived modules, executes no embedded program and makes
+no Docker, network, task or model request. Passing means identity and syntax
+checks passed; the report explicitly says `standalone_runnable: false` and
+`runtime_validation: not_performed`. Other dynamically constructed programs are
+outside this syntax check.
+
+## Runtime prerequisites and authority
+
+For an owner-authorized replay, first assemble a **new isolated experiment
+workspace**. Preserve this archive and copy the selected files into that workspace.
+Keep the candidate client and HTTP bridge separate from the operator-only grader.
+
+| Requirement | Historical interface |
+| --- | --- |
+| Host runtime | Python 3.12, Linux Docker and the recorded native image digests |
+| Candidate client | `FRONTIER_DRIVER_API_URL` is an HTTP origin on the isolated trial network; `FRONTIER_DRIVER_TOKEN_FILE` names the independently provisioned capability file |
+| Final acknowledgement | The candidate client writes `/workspace/.frontier/final-submission.json`; this is a container protocol path |
+| Bridge controller | An external controller supplies `Trial`, its queue, start/deadline and native result mapping; the module alone does not launch a service |
+| Existing installer | A compatible checkout of the evaluated LoopX source supplies the benchmark runtime using `DockerEnvironment` |
+| Native grader inputs | The manifest, `inputs/train.json` or `inputs/eval.json`, native images, submitted tar and a writable output directory, supplied by the operator |
+| Reference qualification | `qualify` additionally needs the original pull receipt and reference patch; these operator-only prerequisites are not supplied |
+
+The no-new-privileges bootstrap assumes Linux `prctl`, default Docker seccomp,
+zero effective capabilities and the `docker-default` AppArmor profile. It is not
+a portable sandbox. Trusted installation can run as root before `sealed=True`;
+the original controller, which is not included, owns sealing, mounts, firewall,
+network isolation, token distribution, grader workers, final-stop and cleanup.
+Source inspection or the hash check does not qualify those external boundaries.
+
+## Known limits
+
+The local API is a partial compatibility implementation, not a replacement for
+the official driver. It stores control state for inspection but does not rebuild
+in-memory tokens, payload ownership and idempotency after restart. Train limits
+and the archive-size limit are frozen experiment assumptions. Final evaluation
+is deliberately unavailable through the candidate's train-status operation.
+
+The native grader retains task-specific names, CPU-affinity defaults, output
+directory assumptions and privileged host setup needs. The Docker adapter uses
+blocking subprocesses inside its asynchronous interface and can print command
+and stderr fragments on errors; use only synthetic or separately reviewed
+inputs. A failed final-receipt write after remote acceptance also needs explicit
+controller recovery. These historical limitations are preserved rather than
+silently repaired in an archive publication.
+
+This PR performs identity and syntax validation only. The separate historical
+results are observations from the original pinned environment; they are not a
+new end-to-end replay, portability proof or long-horizon qualification.
+
+## 中文说明
+
+这里保留本次单机四组实验所用的五个兼容文件原始字节及权限，便于检查原生提交
+客户端、HTTP 桥接、Docker 传输和裁判调用方式。实验数据单独存入上方 HF 目录；
+仓库不包含题目输入、参考补丁、裁判源码、凭据、原始轨迹或私有运行配置。
+
+只读检查验证文件身份与语法，不运行模型或裁判。重跑仍需在新的隔离工作区准备
+外部 controller、原生镜像、数据与独立授权。数据源中的完整性资格尚未完成，
+LoopX 处理忠实度仍为 pending，不能把此次归档称为正式配对收益或榜单结果。
+后续兼容修复应另立版本，保留这次冻结源码及历史失败。