diff --git a/docs/en/reference/judge-api.md b/docs/en/reference/judge-api.md
index 14a1efc8..6c694332 100644
--- a/docs/en/reference/judge-api.md
+++ b/docs/en/reference/judge-api.md
@@ -12,8 +12,9 @@ The judge server exposes a REST API for registering runs, submitting archives, p
 |--------|------------|-----------|
 | `sforge run` (host) | admin secret | `register`, auto-eval `submit`, full `history`, `usage` |
 | Agent in the work container | session token | `submit`, `result`, agent `history`, `llm/...`, game endpoints |
+| Feedback reader | feedback relay ID | read-only registration `context` and that registration's agent `result` |
 
-The admin secret (`SFORGE_ADMIN_SECRET`) never enters a container. The work container gets only its session token (`SFORGE_TOKEN`) and the judge URL (`SFORGE_JUDGE_URL`).
+The admin secret (`SFORGE_ADMIN_SECRET`) never enters a container. The work container gets its session token (`SFORGE_TOKEN`), the judge URL (`SFORGE_JUDGE_URL`) and a read-only feedback URL (`SFORGE_FEEDBACK_URL`). Treat the feedback URL as a credential for reading feedback: it cannot submit, read auto-evaluations or access another run/task. Results with a recorded registration binding are further restricted to that registration.
 
 ## Session Registration
 
@@ -35,11 +36,23 @@ Optional fields: `judge_cpu_limit`, `judge_mem_limit`, `max_agent_submissions`,
 **Response:**
 
 ```json
-{"token": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4"}
+{
+  "token": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4",
+  "feedback_relay_id": "<opaque-read-capability>",
+  "evaluation_context": {
+    "schema_version": "sforge_run_evaluation_context_v1",
+    "task_id": "ad_placement_optimization",
+    "run_id": "run-001",
+    "registration_id": "<server-assigned-id>",
+    "evaluation_conditions": null
+  }
+}
 ```
 
 The token identifies the task and run and counts agent rounds (`agent-1`, `agent-2`, ...) and auto-eval rounds (`auto-1`, ...) separately.
 
+When evaluation conditions are known, registering the same run/task with identical conditions and registration options returns the original token, relay and counters. Changed conditions or options return `409`; use a fresh `run_id`. A retry cannot reset the submission budget or switch the run's model gateway. Registrations with unknown conditions retain the previous registration behavior and carry no verified attribution claim.
+
 ## Submit Archive
 
 ```http
@@ -57,6 +70,7 @@ kind: agent
 | `archive` | yes | `.tar.gz` of the workspace; filtered by the task's submission paths |
 | `kind` | no | `agent` (default) or `auto`; `auto` also needs `admin_secret` |
 | `admin_secret` | for `auto` | Admin secret |
+| `include_receipt` | no | `true` adds the accepted archive receipt; default `false` preserves the existing response |
 
 **Response:**
 
@@ -109,6 +123,63 @@ The report is the grader's `result.json` plus timing. The 0–100 score is not i
 | `completed` | Graded; `report` is available (it may still be `valid: false`) |
 | `error` | Grading failed outright; see `error` |
 
+## Evaluation Conditions and Read-Only Feedback
+
+Read the original registration context once before accepting measurements:
+
+```http
+GET /api/v1/feedback/{feedback_relay_id}/context
+```
+
+It returns the same `sforge_run_evaluation_context_v1` object as registration. `evaluation_conditions`, when available, is an object with exactly these fields:
+
+| Field | Meaning |
+|-------|---------|
+| `schema_version` | `sforge_evaluation_conditions_v1` |
+| `task_sha256` | SHA-256 of the `task.toml` bytes parsed by the task loader |
+| `benchmark_sha256` | SHA-256 of the `benchmark.toml` bytes parsed by the benchmark loader |
+| `task_spec_sha256` | SHA-256 of the effective logical task, including instructions, judge command, score direction/rescaling and declared Blob path/digest inventory; no host paths |
+| `native_source_sha256` | SHA-256 inventory of the service's installed Python framework source, observed once at boot |
+| `judge_image_id` | Actual local Docker content ID (`sha256:...`), resolved at registration and used for grading |
+| `execution_config_sha256` | SHA-256 of effective judge CPU/memory/environment/network, timeout and platform |
+| `score_direction` | `maximize` or `minimize` from that frozen task |
+| `selection` | `valid_then_score`: existing valid/completed filtering, raw-score direction and first-tie selection |
+
+The server freezes the loaded task, effective config and backend for accepted work. Changing files or retagging an image after registration does not change its context or the image used by subsequent submissions. Blob downloads and SHA checks remain owned by the normal image-build path; the Docker ID identifies the evaluator and data actually inside the image. The digest is not a claim of deterministic or hermetic grading and does not attest mutable external services/data. No evaluator contents or raw environment values are exposed by this contract.
+
+Poll one result with both its original feedback and provenance:
+
+```http
+GET /api/v1/feedback/{feedback_relay_id}/result/{submission_id}?include_receipt=true
+```
+
+This is the ordinary result response plus:
+
+```json
+{
+  "receipt": {
+    "schema_version": "sforge_submission_receipt_v1",
+    "task_id": "ad_placement_optimization",
+    "round_id": "agent-1",
+    "archive_sha256": "<sha256-of-accepted-upload-bytes>",
+    "archive_bytes": 1234,
+    "evaluation_context": {
+      "schema_version": "sforge_run_evaluation_context_v1",
+      "task_id": "ad_placement_optimization",
+      "run_id": "run-001",
+      "registration_id": "<original-server-assigned-id>",
+      "evaluation_conditions": null
+    }
+  }
+}
+```
+
+The context above shows the unknown-conditions case; known conditions contain the object described in the table. The archive digest covers the accepted compressed upload before submission-path filtering. The same optional receipt is available on `GET /api/v1/result/{submission_id}?include_receipt=true` and `POST /api/v1/submit` with the form field `include_receipt=true`. Omitting the flag preserves the original feedback body, including the raw-score-only report.
+
+A reader can compare this result's context to its installation snapshot with one GET per result. A missing context, unknown conditions or a mismatch must stay unknown for measurement attribution; keep the native feedback usable. Context/result reads never rescan source, inspect the current image, build an image or download a Blob. Unsupported backends, game sessions, unprepared images and unobserved/manual task specifications return `evaluation_conditions: null`. Historical results missing provenance return a null context/digest instead of borrowing current conditions. This interface adds no checkpoint/restore mechanism.
+
+The relay returns `404` for foreign runs/tasks/recorded registrations, auto-evaluations and unknown IDs. Legacy results without a registration binding remain visible only within their matching run/task and carry no recovered attribution. The relay has no submit, history, model-gateway or admin route. Existing session-token APIs and generated submit-client behavior remain available; retained-upload and asynchronous-client options are separate client features.
+
 ## Run History
 
 ```http
diff --git a/docs/zh/reference/judge-api.md b/docs/zh/reference/judge-api.md
index c66c0a7a..6580f189 100644
--- a/docs/zh/reference/judge-api.md
+++ b/docs/zh/reference/judge-api.md
@@ -12,8 +12,9 @@ Judge server 提供 REST API，用于登记运行、提交 archive、轮询结
 |--------|------|------|
 | `sforge run`（宿主机） | admin secret | `register`、auto-eval 的 `submit`、完整 `history`、`usage` |
 | work 容器里的 Agent | session token | `submit`、`result`、Agent 视角的 `history`、`llm/...`、游戏端点 |
+| 反馈读取方 | feedback relay ID | 只读登记 `context` 与该登记的 Agent `result` |
 
-Admin secret（`SFORGE_ADMIN_SECRET`）从不进入容器。Work 容器只拿到自己的 session token（`SFORGE_TOKEN`）和 judge URL（`SFORGE_JUDGE_URL`）。
+Admin secret（`SFORGE_ADMIN_SECRET`）从不进入容器。Work 容器拿到自己的 session token（`SFORGE_TOKEN`）、judge URL（`SFORGE_JUDGE_URL`）和只读反馈 URL（`SFORGE_FEEDBACK_URL`）。反馈 URL 是读取反馈的凭据，不能用于提交、读取 auto-eval 或访问其他 run/task。已记录登记绑定的结果进一步限定在该登记内。
 
 ## 登记会话
 
@@ -35,11 +36,23 @@ Content-Type: application/json
 **响应：**
 
 ```json
-{"token": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4"}
+{
+  "token": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4",
+  "feedback_relay_id": "<opaque-read-capability>",
+  "evaluation_context": {
+    "schema_version": "sforge_run_evaluation_context_v1",
+    "task_id": "ad_placement_optimization",
+    "run_id": "run-001",
+    "registration_id": "<server-assigned-id>",
+    "evaluation_conditions": null
+  }
+}
 ```
 
 Token 标识任务和运行，并分别计数 Agent 轮次（`agent-1`、`agent-2`、...）和 auto-eval 轮次（`auto-1`、...）。
 
+评测条件已知时，同一 run/task 在条件和登记选项完全相同时返回原 token、relay 与计数器；条件或选项变化返回 `409`，需要新的 `run_id`。重试不能重置提交预算或切换运行的模型网关。条件未知的登记保留原有登记行为，不声明归因已验证。
+
 ## 提交 Archive
 
 ```http
@@ -57,6 +70,7 @@ kind: agent
 | `archive` | 是 | 工作区的 `.tar.gz`，按该题的提交路径过滤 |
 | `kind` | 否 | `agent`（默认）或 `auto`；`auto` 还需要 `admin_secret` |
 | `admin_secret` | `auto` 时 | Admin secret |
+| `include_receipt` | 否 | `true` 附带已接受归档的 receipt；默认 `false` 保留原响应 |
 
 **响应：**
 
@@ -109,6 +123,63 @@ GET /api/v1/result/{submission_id}
 | `completed` | 评分完成，`report` 可用（仍可能是 `valid: false`） |
 | `error` | 评分直接失败，见 `error` |
 
+## 评测条件与只读反馈
+
+接受测量前，先读取一次原登记上下文：
+
+```http
+GET /api/v1/feedback/{feedback_relay_id}/context
+```
+
+响应与登记返回的 `sforge_run_evaluation_context_v1` 对象相同。条件可用时，`evaluation_conditions` 对象恰好包含以下字段：
+
+| 字段 | 含义 |
+|------|------|
+| `schema_version` | `sforge_evaluation_conditions_v1` |
+| `task_sha256` | Task loader 实际解析的 `task.toml` 字节的 SHA-256 |
+| `benchmark_sha256` | Benchmark loader 实际解析的 `benchmark.toml` 字节的 SHA-256 |
+| `task_spec_sha256` | 有效逻辑任务的 SHA-256，包括指令、judge 命令、分数方向/缩放及 Blob 路径和摘要清单；不含宿主机路径 |
+| `native_source_sha256` | 服务启动时一次观察的已安装 Python 框架源码清单的 SHA-256 |
+| `judge_image_id` | 登记时解析、评分时实际使用的本地 Docker 内容 ID（`sha256:...`） |
+| `execution_config_sha256` | 有效 judge CPU/内存/环境/网络、超时及平台的 SHA-256 |
+| `score_direction` | 冻结任务的 `maximize` 或 `minimize` |
+| `selection` | `valid_then_score`：已有的有效且已完成筛选、原始分数方向比较与并列取首条规则 |
+
+服务为已接受的工作冻结已加载任务、有效配置及 backend。登记后修改文件或镜像标签，不会改变原上下文或后续提交实际使用的镜像。Blob 下载与 SHA 校验仍归原生镜像构建路径负责；Docker ID 标识镜像中实际的评测器与数据。摘要不代表评分确定性或封闭性，也不证明外部可变服务/数据未变。协议不暴露评测器内容或原始环境变量值。
+
+一次请求读取结果原文与来源：
+
+```http
+GET /api/v1/feedback/{feedback_relay_id}/result/{submission_id}?include_receipt=true
+```
+
+响应在原结果之上增加：
+
+```json
+{
+  "receipt": {
+    "schema_version": "sforge_submission_receipt_v1",
+    "task_id": "ad_placement_optimization",
+    "round_id": "agent-1",
+    "archive_sha256": "<sha256-of-accepted-upload-bytes>",
+    "archive_bytes": 1234,
+    "evaluation_context": {
+      "schema_version": "sforge_run_evaluation_context_v1",
+      "task_id": "ad_placement_optimization",
+      "run_id": "run-001",
+      "registration_id": "<original-server-assigned-id>",
+      "evaluation_conditions": null
+    }
+  }
+}
+```
+
+上例展示条件未知的情况；条件已知时包含表中描述的对象。归档摘要覆盖按提交路径过滤之前收到的压缩上传字节。`GET /api/v1/result/{submission_id}?include_receipt=true` 与带 `include_receipt=true` 表单字段的 `POST /api/v1/submit` 也提供同一可选 receipt。省略开关时保留原反馈正文，报告仍只含原始分数。
+
+读取方每个结果只需一次 GET，即可将其上下文与安装时快照比较。上下文缺失、条件未知或不匹配时，测量归因必须保留未知，原生反馈仍可使用。读取 context/result 不会重扫源码、查询当前镜像、构建镜像或下载 Blob。不支持的 backend、游戏会话、未准备镜像及未经 loader 观察的手工 TaskSpec 返回 `evaluation_conditions: null`。旧结果缺少来源时返回 null 上下文/摘要，不借用当前条件。本接口不增加 checkpoint/restore 机制。
+
+对其他 run/task/已记录登记、auto-eval 和未知 ID，relay 返回 `404`。没有登记绑定的旧结果仍只在匹配的 run/task 内可见，归因信息不会被恢复。Relay 没有提交、历史、模型网关或 admin 路由。已有 session-token API 和生成的提交客户端保持可用；保留上传归档与异步客户端选项属于独立的客户端功能。
+
 ## 运行历史
 
 ```http
diff --git a/pyproject.toml b/pyproject.toml
index 23208b17..d69418a3 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -22,6 +22,7 @@ dependencies = [
     "tqdm",
     "rich",
     "fastapi",
+    "pydantic>=2.0",
     "python-multipart",
     "uvicorn[standard]",
     "jinja2",
diff --git a/sforge/harness/evaluation_conditions.py b/sforge/harness/evaluation_conditions.py
new file mode 100644
index 00000000..c563b073
--- /dev/null
+++ b/sforge/harness/evaluation_conditions.py
@@ -0,0 +1,129 @@
+# Copyright (c) 2026 ByteDance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Service-observed evaluation inputs, distinct from a caller's version labels."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from sforge.harness.backend import ContainerBackend
+from sforge.harness.config import SForgeConfig
+from sforge.harness.run_evaluation import judge_container_options
+from sforge.harness.task_spec import TaskSpec, task_spec_sha256
+
+
+class EvaluationConditions(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    schema_version: Literal["sforge_evaluation_conditions_v1"] = "sforge_evaluation_conditions_v1"
+    task_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
+    benchmark_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
+    task_spec_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
+    native_source_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
+    judge_image_id: str = Field(pattern=r"^sha256:[0-9a-f]{64}$")
+    execution_config_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
+    score_direction: Literal["maximize", "minimize"]
+    selection: str
+
+
+class RunEvaluationContext(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    schema_version: Literal["sforge_run_evaluation_context_v1"] = "sforge_run_evaluation_context_v1"
+    task_id: str
+    run_id: str
+    registration_id: str
+    evaluation_conditions: EvaluationConditions | None
+
+
+@dataclass(frozen=True)
+class RegisteredEvaluator:
+    """Keep accepted work on its own registration's executable inputs."""
+
+    task: TaskSpec
+    config: SForgeConfig
+    backend: ContainerBackend
+    context: RunEvaluationContext
+
+
+def native_source_sha256(source_root: Path | None = None) -> str | None:
+    """Observe the installed Python source once at service boot, not per read."""
+    if source_root is None:
+        import sforge
+
+        source_root = Path(sforge.__file__).parent
+    digest = hashlib.sha256()
+    try:
+        paths = sorted(source_root.rglob("*.py"))
+        if not paths:
+            return None
+        for path in paths:
+            digest.update(path.relative_to(source_root).as_posix().encode() + b"\0")
+            digest.update(hashlib.sha256(path.read_bytes()).digest())
+    except OSError:
+        return None
+    return digest.hexdigest()
+
+
+def capture_conditions(
+    task: TaskSpec, backend: ContainerBackend, source_sha256: str | None, config: SForgeConfig,
+) -> EvaluationConditions | None:
+    """Unknown image/data identity preserves feedback without claiming attribution.
+
+    The Docker content ID identifies the evaluator and data in that image. It
+    does not attest mutable external services or deterministic/hermetic grading.
+    Reading context never builds an image or downloads a task blob.
+    """
+    revision = getattr(task, "_source_revision", None)
+    if revision is None or source_sha256 is None or task.game_mode or backend.backend_name != "docker":
+        return None
+    try:
+        effective = task_spec_sha256(task)
+    except (TypeError, ValueError):
+        return None
+    if effective != revision.task_spec_sha256 or task.judge.direction not in ("maximize", "minimize"):
+        return None
+    try:
+        image_id = backend.client.images.get(task.judge_image_key).id
+    except Exception:
+        return None
+    if not isinstance(image_id, str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", image_id):
+        return None
+    execution = {**judge_container_options(task, config),
+                 "timeout": task.judge.timeout, "platform": task.platform}
+    try:
+        raw = json.dumps(execution, sort_keys=True, separators=(",", ":"),
+                         ensure_ascii=False, allow_nan=False).encode()
+    except (TypeError, ValueError):
+        return None
+    return EvaluationConditions(
+        task_sha256=revision.task_sha256,
+        benchmark_sha256=revision.benchmark_sha256,
+        task_spec_sha256=effective,
+        native_source_sha256=source_sha256,
+        judge_image_id=image_id,
+        execution_config_sha256=hashlib.sha256(raw).hexdigest(),
+        score_direction=task.judge.direction,
+        # Modern grading.select_best filters valid completed entries, compares
+        # their raw score in this direction and retains the first tie.
+        selection="valid_then_score",
+    )
diff --git a/sforge/harness/judge_server.py b/sforge/harness/judge_server.py
index 95b3f003..62371c2f 100644
--- a/sforge/harness/judge_server.py
+++ b/sforge/harness/judge_server.py
@@ -16,6 +16,8 @@
 
 from __future__ import annotations
 
+import copy
+import hashlib
 import json
 import logging
 import secrets
@@ -26,6 +28,7 @@
 from enum import Enum
 from pathlib import Path
 from typing import Any
+from typing import Literal
 
 import requests
 from fastapi import FastAPI, File, Form, HTTPException, Query, UploadFile
@@ -41,6 +44,9 @@
 )
 from sforge.harness.constants import CONTAINER_DIR, get_admin_secret
 from sforge.harness.docker_build import BuildImageError, build_judge_image
+from sforge.harness.evaluation_conditions import (
+    RegisteredEvaluator, RunEvaluationContext, capture_conditions, native_source_sha256,
+)
 from sforge.harness.llm_gateway import UsageLog, add_gateway_routes, make_client
 from sforge.harness.run_evaluation import judge_submission
 from sforge.harness.grading import select_best
@@ -60,6 +66,10 @@ class SubmissionCooldownActive(Exception):
     pass
 
 
+class RegistrationConditionsDiffer(ValueError):
+    pass
+
+
 # --- Models ---
 
 
@@ -70,11 +80,21 @@ class SubmissionStatus(str, Enum):
     ERROR = "error"
 
 
+class SubmissionReceipt(BaseModel):
+    schema_version: Literal["sforge_submission_receipt_v1"] = "sforge_submission_receipt_v1"
+    task_id: str
+    round_id: str | None
+    archive_sha256: str | None
+    archive_bytes: int | None
+    evaluation_context: RunEvaluationContext | None = None
+
+
 class ResultResponse(BaseModel):
     submission_id: str
     status: SubmissionStatus
     report: dict | None = None
     error: str | None = None
+    receipt: SubmissionReceipt | None = None
 
 
 class GameNewRequest(BaseModel):
@@ -144,6 +164,8 @@ class RegisterRequest(BaseModel):
 
 class RegisterResponse(BaseModel):
     token: str
+    feedback_relay_id: str
+    evaluation_context: RunEvaluationContext | None = None
 
 
 class SubmitResponse(BaseModel):
@@ -151,6 +173,7 @@ class SubmitResponse(BaseModel):
     round_id: str
     status: SubmissionStatus
     remaining_submissions: int | None = None
+    receipt: SubmissionReceipt | None = None
 
 
 # --- Game session state ---
@@ -202,7 +225,12 @@ def __init__(self, config: SForgeConfig):
         self._game_lock = threading.Lock()
         self._history_lock = threading.Lock()
         self._tokens_lock = threading.Lock()
+        self._registration_lock = threading.Lock()
+        self._evaluators: dict[str, RegisteredEvaluator] = {}
+        self._run_evaluators: dict[str, RegisteredEvaluator] = {}
+        self._feedback_relays: dict[str, str] = {}
         self.backend = create_backend_from_config(config)
+        self.native_source_sha256 = native_source_sha256()
 
         self._reaper_stop = threading.Event()
         self._reaper_thread = threading.Thread(
@@ -294,11 +322,19 @@ def register_session(self, task_id: str, run_id: str,
                          e2b_template_namespace: str | None = None,
                          llm_base_url: str | None = None,
                          llm_api_key: str | None = None) -> str:
-        if task_id not in self.tasks:
-            raise ValueError(f"Unknown task: {task_id}")
-        token = secrets.token_hex(16)
-        with self._tokens_lock:
-            self.tokens[token] = {
+        with self._registration_lock:
+            if task_id not in self.tasks:
+                raise ValueError(f"Unknown task: {task_id}")
+            request_options = {
+                "judge_cpu_limit": judge_cpu_limit, "judge_mem_limit": judge_mem_limit,
+                "backend": backend, "k8s_image_registry": k8s_image_registry,
+                "k8s_namespace": k8s_namespace, "k8s_node_selector": k8s_node_selector,
+                "k8s_kubeconfig": k8s_kubeconfig, "max_agent_submissions": max_agent_submissions,
+                "submission_cooldown": submission_cooldown,
+                "e2b_template_map": e2b_template_map, "e2b_template_namespace": e2b_template_namespace,
+                "llm_base_url": llm_base_url, "llm_api_key": llm_api_key,
+            }
+            info = {
                 "task_id": task_id,
                 "run_id": run_id,
                 "next_agent": 1,
@@ -310,15 +346,11 @@ def register_session(self, task_id: str, run_id: str,
                 "last_agent_submit_at": 0.0,
                 "llm_base_url": llm_base_url,
                 "llm_api_key": llm_api_key,
-                "usage_log": UsageLog(
-                    self.config.log_dir / "runs" / run_id / task_id / "llm_requests.jsonl"
-                ),
+                "registration_id": uuid.uuid4().hex,
+                "feedback_relay_id": secrets.token_urlsafe(24),
+                "_request_options": copy.deepcopy(request_options),
             }
             run_key = self._run_key(run_id, task_id)
-            self.run_resource_limits[run_key] = {
-                "judge_cpu_limit": judge_cpu_limit,
-                "judge_mem_limit": judge_mem_limit,
-            }
             effective_e2b_template_map = (
                 self.config.e2b_template_map
                 if not e2b_template_map
@@ -334,11 +366,12 @@ def register_session(self, task_id: str, run_id: str,
                 or effective_e2b_template_namespace
                 != self.config.e2b_template_namespace
             )
+            prepared_backend = None
             if backend and (
                 backend != self.backend.backend_name or e2b_config_changed
             ):
                 from sforge.harness.backend.factory import create_backend
-                self.run_backends[run_key] = create_backend(
+                prepared_backend = create_backend(
                     backend,
                     k8s_namespace=k8s_namespace or self.config.k8s_namespace,
                     k8s_node_selector=k8s_node_selector or self.config.k8s_node_selector,
@@ -347,7 +380,50 @@ def register_session(self, task_id: str, run_id: str,
                     e2b_template_map=effective_e2b_template_map,
                     e2b_template_namespace=effective_e2b_template_namespace,
                 )
-        return token
+            task = copy.deepcopy(self.tasks[task_id])
+            config = copy.deepcopy(self.config)
+            if judge_cpu_limit is not None:
+                config.judge_cpu_limit = judge_cpu_limit
+            if judge_mem_limit is not None:
+                config.judge_mem_limit = judge_mem_limit
+            registered_backend = prepared_backend or self.backend
+            context = RunEvaluationContext(
+                task_id=task_id, run_id=run_id, registration_id=info["registration_id"],
+                evaluation_conditions=capture_conditions(task, registered_backend,
+                                                        self.native_source_sha256, config),
+            )
+            evaluator = RegisteredEvaluator(task, config, registered_backend, context)
+            info["evaluation_context"] = context.model_dump(mode="json")
+            with self._tokens_lock:
+                for previous_token, previous in self.tokens.items():
+                    if previous["run_id"] != run_id or previous["task_id"] != task_id:
+                        continue
+                    prior = (previous.get("evaluation_context") or {}).get("evaluation_conditions")
+                    current = info["evaluation_context"]["evaluation_conditions"]
+                    if prior is not None or current is not None:
+                        if prior == current and previous.get("_request_options") == request_options:
+                            # Keep rounds, cooldown, gateway ownership and the
+                            # existing log namespace on an exact retry.
+                            return previous_token
+                        raise RegistrationConditionsDiffer(
+                            "Run/task already registered with evaluation conditions; "
+                            "reuse its registration or use a fresh run_id"
+                        )
+                token = secrets.token_hex(16)
+                info["usage_log"] = UsageLog(
+                    self.config.log_dir / "runs" / run_id / task_id / "llm_requests.jsonl"
+                )
+                self.run_resource_limits[run_key] = {
+                    "judge_cpu_limit": judge_cpu_limit, "judge_mem_limit": judge_mem_limit,
+                }
+                if prepared_backend is not None:
+                    self.run_backends[run_key] = prepared_backend
+                self._evaluators[info["registration_id"]] = evaluator
+                if context.evaluation_conditions is not None:
+                    self._run_evaluators[run_key] = evaluator
+                self._feedback_relays[info["feedback_relay_id"]] = token
+                self.tokens[token] = info
+                return token
 
     @staticmethod
     def _run_key(run_id: str, task_id: str) -> str:
@@ -375,7 +451,25 @@ def resolve_token(self, token: str) -> dict:
             info = self.tokens.get(token)
         if info is None:
             raise KeyError("Invalid token")
-        return dict(info)
+        result = dict(info)
+        if "evaluation_context" in result:
+            result["evaluation_context"] = copy.deepcopy(result["evaluation_context"])
+        return result
+
+    def resolve_feedback_relay(self, relay_id: str) -> tuple[str, dict]:
+        """A read-only capability for one registration, without its judge token."""
+        with self._tokens_lock:
+            token = self._feedback_relays.get(relay_id)
+        if token is None:
+            raise KeyError("Invalid feedback relay")
+        return token, self.resolve_token(token)
+
+    def _evaluator_for_token(self, token: str) -> RegisteredEvaluator | None:
+        with self._tokens_lock:
+            info = self.tokens.get(token)
+            if info is None:
+                raise KeyError("Invalid token")
+            return self._evaluators.get(info.get("registration_id"))
 
     def consume_round(self, token: str, kind: str = "agent") -> tuple[str, str, str, int | None, str | None, int | None]:
         """Allocate the next round ID for a token. Returns (task_id, run_id, round_id, cpu, mem, remaining).
@@ -450,7 +544,8 @@ def get_run_history(self, run_id: str, task_id: str | None = None) -> dict:
             entries = list(self.run_history.get(history_key, []))
 
         resolved_tid = task_id or next((e["task_id"] for e in entries if e.get("task_id")), None)
-        task_spec = self.tasks.get(resolved_tid) if resolved_tid else None
+        evaluator = self._run_evaluators.get(self._run_key(run_id, resolved_tid)) if resolved_tid else None
+        task_spec = evaluator.task if evaluator else self.tasks.get(resolved_tid) if resolved_tid else None
         best = select_best(entries, task_spec.judge.direction if task_spec else "maximize") or {}
 
         sub_entries = [e for e in entries if e.get("type") == "submission"]
@@ -469,10 +564,13 @@ def load_tasks(self) -> None:
         self.tasks = {t.task_id: t for t in task_list}
 
     def submit(self, task_id: str, archive: bytes, run_id: str | None = None, round: str | None = None,
-               judge_cpu_limit: int | None = None, judge_mem_limit: str | None = None) -> str:
+               judge_cpu_limit: int | None = None, judge_mem_limit: str | None = None,
+               evaluator: RegisteredEvaluator | None = None) -> str:
         """Submit and grade asynchronously. Returns submission_id for polling."""
-        if task_id not in self.tasks:
+        if evaluator is None and task_id not in self.tasks:
             raise ValueError(f"Unknown task: {task_id}")
+        if evaluator is not None and (evaluator.context.task_id != task_id or evaluator.context.run_id != run_id):
+            raise ValueError("Evaluation registration differs from the submitted run/task")
 
         submission_id = uuid.uuid4().hex[:12]
         log_run_id = run_id or submission_id
@@ -483,13 +581,17 @@ def submit(self, task_id: str, archive: bytes, run_id: str | None = None, round:
             "round": round,
             "report": None,
             "error": None,
+            "registration_id": evaluator.context.registration_id if evaluator else None,
+            "archive_sha256": hashlib.sha256(archive).hexdigest(),
+            "archive_bytes": len(archive),
+            "evaluation_context": evaluator.context.model_dump(mode="json") if evaluator else None,
         }
         self._record_submission(log_run_id, submission_id, task_id, round, None, None, status="running")
 
         thread = threading.Thread(
             target=self._grade_worker,
             args=(submission_id, task_id, archive, run_id, round,
-                  judge_cpu_limit, judge_mem_limit),
+                  judge_cpu_limit, judge_mem_limit, evaluator),
             daemon=True,
         )
         thread.start()
@@ -497,10 +599,11 @@ def submit(self, task_id: str, archive: bytes, run_id: str | None = None, round:
 
     def _grade_worker(self, submission_id: str, task_id: str, archive: bytes,
                       run_id: str | None = None, round: int | None = None,
-                      judge_cpu_limit: int | None = None, judge_mem_limit: str | None = None) -> None:
+                      judge_cpu_limit: int | None = None, judge_mem_limit: str | None = None,
+                      evaluator: RegisteredEvaluator | None = None) -> None:
         self.submissions[submission_id]["status"] = SubmissionStatus.RUNNING
         try:
-            task_spec = self.tasks[task_id]
+            task_spec = evaluator.task if evaluator else self.tasks[task_id]
             log_run_id = run_id or submission_id
             sub_num = str(round) if round is not None else "1"
             sub_log_dir = (
@@ -512,22 +615,24 @@ def _grade_worker(self, submission_id: str, task_id: str, archive: bytes,
                 overrides["judge_cpu_limit"] = judge_cpu_limit
             if judge_mem_limit is not None:
                 overrides["judge_mem_limit"] = judge_mem_limit
-            config = replace(self.config, **overrides) if overrides else self.config
+            config = evaluator.config if evaluator else replace(self.config, **overrides) if overrides else self.config
+            conditions = evaluator.context.evaluation_conditions if evaluator else None
             report = judge_submission(
                 task_spec=task_spec,
                 archive=archive,
                 config=config,
-                backend=self._get_backend(run_id, task_id),
+                backend=evaluator.backend if evaluator else self._get_backend(run_id, task_id),
                 submission_id=submission_id,
                 log_dir=sub_log_dir,
+                **({"evaluator_image": conditions.judge_image_id} if conditions else {}),
             )
             report_dict = report.to_dict()
-            self.submissions[submission_id]["status"] = SubmissionStatus.COMPLETED
             self.submissions[submission_id]["report"] = report_dict
+            self.submissions[submission_id]["status"] = SubmissionStatus.COMPLETED
             self._record_submission(log_run_id, submission_id, task_id, round, report_dict, None)
         except Exception as e:
-            self.submissions[submission_id]["status"] = SubmissionStatus.ERROR
             self.submissions[submission_id]["error"] = str(e)
+            self.submissions[submission_id]["status"] = SubmissionStatus.ERROR
             self._record_submission(log_run_id, submission_id, task_id, round, None, str(e))
 
     def get_result(self, submission_id: str) -> dict | None:
@@ -867,6 +972,25 @@ def _bg_cleanup():
 # --- FastAPI app factory ---
 
 
+def submission_receipt(result: dict) -> SubmissionReceipt:
+    return SubmissionReceipt(
+        task_id=result["task_id"], round_id=result.get("round"),
+        archive_sha256=result.get("archive_sha256"), archive_bytes=result.get("archive_bytes"),
+        evaluation_context=result.get("evaluation_context"),
+    )
+
+
+def public_result(submission_id: str, result: dict, include_receipt: bool) -> ResultResponse:
+    # The rescale anchors remain operator-only, including through the relay.
+    report = result["report"]
+    return ResultResponse(
+        submission_id=submission_id, status=result["status"],
+        report={k: v for k, v in report.items() if k != "score_0_100"} if report is not None else None,
+        error=result["error"],
+        **({"receipt": submission_receipt(result)} if include_receipt else {}),
+    )
+
+
 def create_app(config: SForgeConfig | None = None) -> FastAPI:
     if config is None:
         config = load_config()
@@ -875,20 +999,13 @@ def create_app(config: SForgeConfig | None = None) -> FastAPI:
     state = JudgeState(config)
     state.load_tasks()
 
-    @app.get("/api/v1/result/{submission_id}")
-    def get_result(submission_id: str) -> ResultResponse:
+    @app.get("/api/v1/result/{submission_id}", response_model_exclude_unset=True)
+    def get_result(submission_id: str, include_receipt: bool = Query(False)) -> ResultResponse:
         """Get result of an async submission."""
         result = state.get_result(submission_id)
         if result is None:
             raise HTTPException(status_code=404, detail="Submission not found")
-        return ResultResponse(
-            submission_id=submission_id,
-            status=result["status"],
-            # Agents poll this; keep the rescale anchors out of their view.
-            report={k: v for k, v in (result["report"] or {}).items() if k != "score_0_100"}
-            if result["report"] is not None else None,
-            error=result["error"],
-        )
+        return public_result(submission_id, result, include_receipt)
 
     @app.post("/api/v1/register")
     def register(req: RegisterRequest) -> RegisterResponse:
@@ -912,16 +1029,21 @@ def register(req: RegisterRequest) -> RegisterResponse:
                 llm_base_url=req.llm_base_url,
                 llm_api_key=req.llm_api_key,
             )
-            return RegisterResponse(token=token)
+            info = state.resolve_token(token)
+            return RegisterResponse(token=token, feedback_relay_id=info["feedback_relay_id"],
+                                    evaluation_context=info.get("evaluation_context"))
+        except RegistrationConditionsDiffer as e:
+            raise HTTPException(status_code=409, detail=str(e))
         except ValueError as e:
             raise HTTPException(status_code=404, detail=str(e))
 
-    @app.post("/api/v1/submit")
+    @app.post("/api/v1/submit", response_model_exclude_unset=True)
     def submit(
         token: str = Form(...),
         archive: UploadFile = File(...),
         kind: str = Form("agent"),
         admin_secret: str = Form(""),
+        include_receipt: bool = Form(False),
     ) -> SubmitResponse:
         """Submit an archive using a session token.
 
@@ -933,6 +1055,7 @@ def submit(
         if kind == "auto" and admin_secret != state.admin_secret:
             raise HTTPException(status_code=403, detail="admin_secret required for auto submissions")
         try:
+            evaluator = state._evaluator_for_token(token)
             task_id, run_id, round_id, judge_cpu_limit, judge_mem_limit, remaining = state.consume_round(token, kind)
         except KeyError:
             raise HTTPException(status_code=401, detail="Invalid token")
@@ -944,12 +1067,14 @@ def submit(
         try:
             submission_id = state.submit(task_id, archive_data, run_id, round_id,
                                          judge_cpu_limit=judge_cpu_limit,
-                                         judge_mem_limit=judge_mem_limit)
+                                         judge_mem_limit=judge_mem_limit, evaluator=evaluator)
             return SubmitResponse(
                 submission_id=submission_id,
                 round_id=round_id,
                 status=SubmissionStatus.QUEUED,
                 remaining_submissions=remaining,
+                **({"receipt": submission_receipt(state.get_result(submission_id))}
+                   if include_receipt else {}),
             )
         except ValueError as e:
             raise HTTPException(status_code=404, detail=str(e))
@@ -975,7 +1100,8 @@ def history(token: str = Query(...), admin_secret: str = Query("")) -> dict:
             for e in full.get("entries", [])
             if not (e.get("round") or "").startswith("auto-")
         ]
-        task_spec = state.tasks.get(info["task_id"])
+        evaluator = state._run_evaluators.get(state._run_key(info["run_id"], info["task_id"]))
+        task_spec = evaluator.task if evaluator else state.tasks.get(info["task_id"])
         best = select_best(agent_entries, task_spec.judge.direction if task_spec else "maximize") or {}
         return {
             "run_id": info["run_id"],
@@ -999,6 +1125,31 @@ def usage(token: str = Query(...), admin_secret: str = Query("")) -> dict:
 
     add_gateway_routes(app, state, make_client(config.https_proxy, config.no_proxy))
 
+    @app.get("/api/v1/feedback/{relay_id}/context")
+    def feedback_context(relay_id: str) -> RunEvaluationContext | None:
+        """Read the original registration snapshot without submission authority."""
+        try:
+            _, info = state.resolve_feedback_relay(relay_id)
+            return info.get("evaluation_context")
+        except KeyError:
+            raise HTTPException(status_code=404, detail="Feedback relay not found")
+
+    @app.get("/api/v1/feedback/{relay_id}/result/{submission_id}", response_model_exclude_unset=True)
+    def feedback_result(relay_id: str, submission_id: str,
+                        include_receipt: bool = Query(False)) -> ResultResponse:
+        """Read only this registration's agent result, never host-only auto work."""
+        try:
+            _, info = state.resolve_feedback_relay(relay_id)
+        except KeyError:
+            raise HTTPException(status_code=404, detail="Feedback relay not found")
+        result = state.get_result(submission_id)
+        if (result is None or result.get("run_id") != info["run_id"]
+                or result.get("task_id") != info["task_id"]
+                or not str(result.get("round") or "").startswith("agent-")
+                or result.get("registration_id") not in (None, info.get("registration_id"))):
+            raise HTTPException(status_code=404, detail="Submission not found")
+        return public_result(submission_id, result, include_receipt)
+
     # --- Game session routes ---
 
     @app.post("/api/v1/game/{run_id}/{task_id}/new")
diff --git a/sforge/harness/run_agent.py b/sforge/harness/run_agent.py
index f703b35f..64a807bb 100644
--- a/sforge/harness/run_agent.py
+++ b/sforge/harness/run_agent.py
@@ -510,7 +510,8 @@ def run_agent(
                 wait = 2 * _reg_attempt
                 logger.warning("Register attempt %d/5 failed (%s), retrying in %ds...", _reg_attempt, exc, wait)
                 time.sleep(wait)
-        session_token = reg_resp.json()["token"]
+        registration = reg_resp.json()
+        session_token = registration["token"]
         logger.info("Registered session with judge server")
         # The agent reaches the model through the judge server's LLM gateway.
         api_url = f"{judge_url.rstrip('/')}{GATEWAY_ROUTE}/{session_token}"
@@ -531,6 +532,9 @@ def run_agent(
             sanitize_e2b_proxy_env(env)
         env["SFORGE_JUDGE_URL"] = judge_url
         env["SFORGE_TOKEN"] = session_token
+        relay_id = registration.get("feedback_relay_id")
+        if relay_id:
+            env["SFORGE_FEEDBACK_URL"] = f"{judge_url.rstrip('/')}/api/v1/feedback/{relay_id}"
 
         if task_spec.game_mode:
             game_api_base = (
diff --git a/sforge/harness/run_evaluation.py b/sforge/harness/run_evaluation.py
index 5f178cc6..f93fe6ff 100644
--- a/sforge/harness/run_evaluation.py
+++ b/sforge/harness/run_evaluation.py
@@ -100,6 +100,16 @@ def _filter_archive_files(
     return allowed
 
 
+def judge_container_options(task_spec: TaskSpec, config: SForgeConfig) -> dict:
+    """Resolve the execution options shared by grading and its source receipt."""
+    cpu = config.judge_cpu_limit if config.judge_cpu_limit is not None else task_spec.judge.cpu_limit
+    mem = config.judge_mem_limit if config.judge_mem_limit is not None else task_spec.judge.mem_limit
+    env = dict(config.judge_extra_env)
+    env.update(resource_env(cpu, mem))
+    return {"environment": env, "cpu_limit": cpu, "mem_limit": mem,
+            "network": task_spec.judge.internet}
+
+
 def judge_submission(
     task_spec: TaskSpec,
     archive: bytes,
@@ -109,6 +119,7 @@ def judge_submission(
     timeout: int | None = None,
     log_dir: Path | None = None,
     verbose: bool = False,
+    evaluator_image: str | None = None,
 ) -> EvalReport:
     """
     Grade an archive submission in an ephemeral judge container.
@@ -148,26 +159,19 @@ def judge_submission(
     isolation = None
     try:
         # 1. Ensure judge image exists (Docker builds locally; K8s requires pre-built images in CR)
-        if backend.backend_name == "docker":
+        if backend.backend_name == "docker" and evaluator_image is None:
             build_judge_image(task_spec, config)
 
         # 2. Create + start ephemeral container
         container_name = f"sforge.judge.{task_spec.task_id}.{submission_id}"
         # Graders run untrusted code: no model key, no proxies, nothing but
         # what the operator explicitly passes via SFORGE_JUDGE_EXTRA_ENV.
-        env = dict(config.judge_extra_env)
-        # Resolve resource limits: config > task defaults
-        cpu = config.judge_cpu_limit if config.judge_cpu_limit is not None else task_spec.judge.cpu_limit
-        mem = config.judge_mem_limit if config.judge_mem_limit is not None else task_spec.judge.mem_limit
-        env.update(resource_env(cpu, mem))
-        offline = not task_spec.judge.internet
+        options = judge_container_options(task_spec, config)
+        offline = not options["network"]
         handle = backend.create_container(
-            task_spec.judge_image_key,
+            evaluator_image or task_spec.judge_image_key,
             container_name,
-            environment=env,
-            cpu_limit=cpu,
-            mem_limit=mem,
-            network=not offline,
+            **options,
         )
         backend.start_container(handle)
         if offline and backend.backend_name != "docker":
diff --git a/sforge/harness/task_spec.py b/sforge/harness/task_spec.py
index 7e21f986..a76856b8 100644
--- a/sforge/harness/task_spec.py
+++ b/sforge/harness/task_spec.py
@@ -25,8 +25,9 @@
 from __future__ import annotations
 
 import hashlib
+import json
 import tomllib
-from dataclasses import dataclass
+from dataclasses import asdict, dataclass, field
 from functools import cache, cached_property
 from pathlib import Path
 
@@ -71,6 +72,7 @@ class Benchmark:
     name: str                   # image name prefix: <name>.{base,work,judge}.<id>
     registry: str | None = None  # where pre-built images live (sforge pull, k8s)
     blob_url: str | None = None  # where task blobs are downloaded from, as <url>/<sha256>
+    _source_sha256: str | None = field(default=None, init=False, repr=False, compare=False)
 
 
 @cache
@@ -79,8 +81,20 @@ def load_benchmark(root: Path) -> Benchmark:
     path = root / "benchmark.toml"
     if not path.exists():
         raise FileNotFoundError(f"{path} not found: pass --benchmark <dir> or set SFORGE_BENCHMARK")
-    data = tomllib.loads(path.read_text())
-    return Benchmark(root, data["name"], data.get("registry"), data.get("blob_url"))
+    raw = path.read_bytes()
+    data = tomllib.loads(raw.decode("utf-8"))
+    benchmark = Benchmark(root, data["name"], data.get("registry"), data.get("blob_url"))
+    object.__setattr__(benchmark, "_source_sha256", hashlib.sha256(raw).hexdigest())
+    return benchmark
+
+
+@dataclass(frozen=True)
+class TaskSourceRevision:
+    """Hashes of the bytes and effective specification parsed by the loader."""
+
+    task_sha256: str
+    benchmark_sha256: str
+    task_spec_sha256: str
 
 
 @dataclass(frozen=True)
@@ -99,6 +113,7 @@ class TaskSpec:
     judge: JudgeSpec
     base: str | None = None
     blobs: tuple[Blob, ...] = ()
+    _source_revision: TaskSourceRevision | None = field(default=None, init=False, repr=False, compare=False)
 
     @property
     def game_mode(self) -> bool:
@@ -152,11 +167,29 @@ def _tree_hash(root: Path, skip: set[Path] = frozenset()) -> str:
     return h.hexdigest()[:12]
 
 
+def task_spec_sha256(spec: TaskSpec) -> str:
+    """Hash logical inputs, including instructions and blobs, without host paths."""
+    value = {
+        key: getattr(spec, key)
+        for key in ("task_id", "name", "version", "platform", "cwd", "internet",
+                    "submit_paths", "submit_exclude", "base")
+    }
+    value.update(
+        benchmark=spec.benchmark.name,
+        work=asdict(spec.work),
+        judge=asdict(spec.judge),
+        blobs=[asdict(blob) for blob in spec.blobs],
+    )
+    raw = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False)
+    return hashlib.sha256(raw.encode("utf-8")).hexdigest()
+
+
 def load_task(task_dir: Path, benchmark: Benchmark | None = None) -> TaskSpec:
     """Load ``<benchmark>/tasks/<id>/``; the benchmark defaults to the one that contains it."""
     task_dir = task_dir.resolve()
     benchmark = benchmark or load_benchmark(task_dir.parent.parent)
-    data = tomllib.loads((task_dir / "task.toml").read_text())
+    raw = (task_dir / "task.toml").read_bytes()
+    data = tomllib.loads(raw.decode("utf-8"))
     env = data["environment"]
     sub = data.get("submission", {})
     judge = data["judge"]
@@ -164,7 +197,7 @@ def load_task(task_dir: Path, benchmark: Benchmark | None = None) -> TaskSpec:
     game = data.get("game", {})
     if data["id"] != task_dir.name:
         raise ValueError(f"{task_dir}: task.toml id {data['id']!r} does not match the directory name")
-    return TaskSpec(
+    spec = TaskSpec(
         task_id=data["id"],
         name=data.get("name", data["id"]),
         version=data.get("version", "0"),
@@ -193,6 +226,17 @@ def load_task(task_dir: Path, benchmark: Benchmark | None = None) -> TaskSpec:
         base=env.get("base"),
         blobs=tuple(Blob(b["path"], b["sha256"]) for b in data.get("blobs", [])),
     )
+    if benchmark._source_sha256 is not None:
+        try:
+            effective_sha256 = task_spec_sha256(spec)
+        except (TypeError, ValueError):
+            # Optional attribution cannot make a previously loadable task
+            # unavailable merely because its metadata is not canonical JSON.
+            return spec
+        object.__setattr__(spec, "_source_revision", TaskSourceRevision(
+            hashlib.sha256(raw).hexdigest(), benchmark._source_sha256, effective_sha256,
+        ))
+    return spec
 
 
 def load_all_tasks(benchmark: Benchmark) -> list[TaskSpec]:
diff --git a/tests/test_evaluation_conditions.py b/tests/test_evaluation_conditions.py
new file mode 100644
index 00000000..9c3ebec9
--- /dev/null
+++ b/tests/test_evaluation_conditions.py
@@ -0,0 +1,308 @@
+# Copyright (c) 2026 ByteDance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Protocol vectors use an explicit fake image; Docker execution is qualified separately."""
+from __future__ import annotations
+
+import copy
+import hashlib
+import io
+import json
+import threading
+import time
+from dataclasses import replace
+from pathlib import Path
+from types import SimpleNamespace
+
+from fastapi.testclient import TestClient
+import pytest
+
+from sforge.harness import judge_server as server
+from sforge.harness.config import SForgeConfig
+from sforge.harness.evaluation_conditions import capture_conditions, native_source_sha256
+from sforge.harness.grading import EvalReport
+from sforge.harness.task_spec import load_benchmark, load_task
+
+
+def benchmark_files(root: Path, *, direction="maximize", platform="linux/amd64", blob_url=None):
+    task = root / "tasks" / "toy"
+    (task / "judge").mkdir(parents=True)
+    (task / "environment").mkdir()
+    (task / "instruction.md").write_text("Produce answer.txt for the synthetic task.\n")
+    (task / "environment/Dockerfile").write_text("FROM scratch\n")
+    (task / "judge/Dockerfile").write_text("FROM scratch\n")
+    payload = b"7\n"
+    digest = hashlib.sha256(payload).hexdigest()
+    (root / "benchmark.toml").write_text('name = "conditions-test"\n' +
+        (f'blob_url = "{blob_url}"\n' if blob_url else ""))
+    (task / "task.toml").write_text(
+        'id = "toy"\nversion = "1.0.0"\n[environment]\ncwd = "/work"\n' +
+        f'platform = "{platform}"\n' +
+        '[submission]\npaths = ["answer.txt"]\n[judge]\ncmd = "python /judge/grade.py"\n'
+        'timeout_sec = 20\ncpus = 1\nmemory = "128m"\n' +
+        f'[score]\ndirection = "{direction}"\n' +
+        f'[[blobs]]\npath = "judge/data/offset.txt"\nsha256 = "{digest}"\n')
+    return task, payload
+
+
+class FakeImages:
+    def __init__(self):
+        self.image = "sha256:" + "a" * 64
+        self.calls = 0
+
+    def get(self, key):
+        self.calls += 1
+        if self.image is None:
+            raise LookupError("unprepared fixture image")
+        return SimpleNamespace(id=self.image)
+
+
+class FakeBackend:
+    backend_name = "docker"
+
+    def __init__(self):
+        self.client = SimpleNamespace(images=FakeImages())
+
+
+@pytest.fixture
+def api(tmp_path, monkeypatch):
+    task, _ = benchmark_files(tmp_path / "benchmark")
+    config = SForgeConfig(benchmark_dir=task.parent.parent, log_dir=tmp_path / "logs")
+    backend = FakeBackend()
+    monkeypatch.setattr(server, "create_backend_from_config", lambda config: backend)
+    original = server.JudgeState
+    states = []
+
+    def construct(config):
+        state = original(config)
+        states.append(state)
+        return state
+
+    monkeypatch.setattr(server, "JudgeState", construct)
+
+    def grade(**kwargs):
+        return EvalReport(task_id=kwargs["task_spec"].task_id, submission_id=kwargs["submission_id"],
+                          score=float(kwargs["archive"]), score_0_100=99, valid=True)
+
+    monkeypatch.setattr(server, "judge_submission", grade)
+    app = server.create_app(config)
+    with TestClient(app) as client:
+        yield client, states[0], backend, task
+    states[0]._reaper_stop.set()
+    states[0]._reaper_thread.join(timeout=1)
+
+
+def register(api, run="run-one", **options):
+    client, state, _, _ = api
+    return client.post("/api/v1/register", json={"task_id":"toy", "run_id":run,
+        "admin_secret":state.admin_secret, **options})
+
+
+def submit(api, registration, data=b"3", **options):
+    return api[0].post("/api/v1/submit", data={"token":registration["token"], **options},
+                       files={"archive":("submission.tar.gz", data)})
+
+
+def wait_result(client, url):
+    deadline = time.monotonic() + 5
+    while time.monotonic() < deadline:
+        response = client.get(url)
+        assert response.status_code == 200, response.text
+        if response.json()["status"] in ("completed", "error"):
+            return response.json()
+        time.sleep(.01)
+    pytest.fail("fixture result did not settle")
+
+
+def test_context_and_each_receipt_share_original_registration_without_read_resampling(api, monkeypatch):
+    client, state, backend, task = api
+    registration = register(api).json()
+    context = registration["evaluation_context"]
+    conditions = context["evaluation_conditions"]
+    assert set(context) == {"schema_version", "task_id", "run_id", "registration_id", "evaluation_conditions"}
+    assert set(conditions) == {
+        "schema_version", "task_sha256", "benchmark_sha256", "task_spec_sha256",
+        "native_source_sha256", "judge_image_id", "execution_config_sha256", "score_direction", "selection",
+    }
+    assert conditions["task_sha256"] == hashlib.sha256((task / "task.toml").read_bytes()).hexdigest()
+    assert conditions["benchmark_sha256"] == hashlib.sha256((task.parent.parent / "benchmark.toml").read_bytes()).hexdigest()
+    assert conditions["judge_image_id"] == backend.client.images.image
+    assert conditions["score_direction"] == "maximize" and conditions["selection"] == "valid_then_score"
+    assert str(task) not in json.dumps(context)
+    relay = "/api/v1/feedback/" + registration["feedback_relay_id"]
+    accepted = submit(api, registration, include_receipt="true").json()
+    assert accepted["receipt"]["evaluation_context"] == context
+    result = wait_result(client, relay + "/result/" + accepted["submission_id"] + "?include_receipt=true")
+    assert result["receipt"]["evaluation_context"] == context
+    assert result["receipt"]["archive_sha256"] == hashlib.sha256(b"3").hexdigest()
+    assert "score_0_100" not in result["report"]
+    image_calls = backend.client.images.calls
+    monkeypatch.setattr(server, "native_source_sha256", lambda: pytest.fail("read rescanned source"))
+    backend.client.images.image = "sha256:" + "b" * 64
+    (task / "task.toml").write_text('invalid changed file')
+    for _ in range(3):
+        assert client.get(relay + "/context").json() == context
+        assert client.get(relay + "/result/" + accepted["submission_id"] + "?include_receipt=true").json() == result
+    assert backend.client.images.calls == image_calls
+
+
+def test_registration_retry_preserves_budget_gateway_and_rejects_condition_or_policy_drift(api):
+    client, state, backend, task = api
+    options = dict(max_agent_submissions=1, llm_base_url="https://model.invalid", llm_api_key="fixture-key")
+    first = register(api, **options).json()
+    submit(api, first)
+    repeated = register(api, **options)
+    assert repeated.status_code == 200 and repeated.json() == first
+    assert submit(api, repeated.json()).status_code == 429
+    snapshot = copy.deepcopy(state.resolve_token(first["token"])["evaluation_context"])
+    for changed in [{"max_agent_submissions":2}, {"llm_api_key":"different-fixture-key"}]:
+        response = register(api, **(options | changed))
+        assert response.status_code == 409 and "fresh run_id" in response.text
+        assert state.resolve_token(first["token"])["evaluation_context"] == snapshot
+    backend.client.images.image = "sha256:" + "c" * 64
+    assert register(api, **options).status_code == 409
+    second = register(api, "run-two", **options).json()
+    assert second["evaluation_context"]["evaluation_conditions"]["judge_image_id"] != snapshot["evaluation_conditions"]["judge_image_id"]
+    assert len(state.tokens) == 2
+
+
+def test_two_runs_auto_work_and_unknown_ids_are_isolated_by_the_read_only_relay(api):
+    client, state, _, _ = api
+    first, second = register(api).json(), register(api, "run-two").json()
+    a = submit(api, first).json()["submission_id"]
+    b = submit(api, second).json()["submission_id"]
+    auto = submit(api, first, kind="auto", admin_secret=state.admin_secret).json()["submission_id"]
+    relay = "/api/v1/feedback/" + first["feedback_relay_id"]
+    assert client.get(relay + "/result/" + a).status_code == 200
+    for sid in [b, auto, "unknown"]:
+        assert client.get(relay + "/result/" + sid).status_code == 404
+    assert client.get("/api/v1/feedback/unknown/context").status_code == 404
+    assert client.post(relay + "/submit").status_code in (404,405)
+    capability = first["feedback_relay_id"]
+    assert client.post("/api/v1/submit", data={"token": capability}, files={"archive": ("x.tar.gz", b"3")}).status_code == 401
+    assert client.post("/api/v1/llm/" + capability + "/v1/messages", json={}).status_code == 401
+
+
+def test_legacy_result_and_default_feedback_do_not_acquire_new_conditions(api):
+    client, state, _, _ = api
+    registration = register(api).json()
+    native = {"status":server.SubmissionStatus.COMPLETED,"task_id":"toy","run_id":"run-one",
+              "round":"agent-0","report":{"score":0,"valid":True,"score_0_100":99},"error":None}
+    state.submissions["legacy"] = native
+    plain = client.get("/api/v1/result/legacy").json()
+    assert plain == {"submission_id":"legacy","status":"completed","report":{"score":0,"valid":True},"error":None}
+    enriched = client.get("/api/v1/feedback/"+registration["feedback_relay_id"]+"/result/legacy?include_receipt=true").json()
+    assert enriched["report"] == plain["report"]
+    assert enriched["receipt"]["evaluation_context"] is None
+    assert enriched["receipt"]["archive_sha256"] is None
+    assert "evaluation_context" not in native
+    assert "receipt" not in submit(api, registration).json()
+
+
+def test_queued_work_uses_its_own_task_resources_and_backend_after_reload(api, monkeypatch):
+    client, state, backend, task = api
+    registration = register(api, judge_cpu_limit=2, judge_mem_limit="160m").json()
+    context = registration["evaluation_context"]
+    release = threading.Event()
+    original = state._grade_worker
+    observed = []
+
+    def delayed(*args):
+        assert release.wait(5)
+        return original(*args)
+
+    def grade(**kwargs):
+        observed.append(kwargs)
+        return EvalReport(task_id=kwargs["task_spec"].task_id, submission_id=kwargs["submission_id"],score=5,valid=True)
+
+    monkeypatch.setattr(state, "_grade_worker", delayed)
+    monkeypatch.setattr(server, "judge_submission", grade)
+    accepted = submit(api, registration).json()
+    state.tasks["toy"] = replace(state.tasks["toy"], judge=replace(state.tasks["toy"].judge,cmd="changed",direction="minimize"))
+    state.config.judge_extra_env["changed"] = "new environment"
+    state.run_backends[state._run_key("run-one","toy")] = object()
+    release.set()
+    result = wait_result(client,"/api/v1/result/"+accepted["submission_id"]+"?include_receipt=true")
+    assert observed[0]["task_spec"].judge.cmd == "python /judge/grade.py"
+    assert observed[0]["config"].judge_cpu_limit == 2
+    assert observed[0]["config"].judge_mem_limit == "160m"
+    assert observed[0]["config"].judge_extra_env == {}
+    assert observed[0]["backend"] is backend
+    assert observed[0]["evaluator_image"] == context["evaluation_conditions"]["judge_image_id"]
+    assert result["receipt"]["evaluation_context"] == context
+    assert state.get_run_history("run-one","toy")["best_score"] == 5
+
+
+@pytest.mark.parametrize("changed", ["task-bytes","benchmark-bytes","instruction","blob","direction","cpu","memory","environment","source","image"])
+def test_conditions_distinguish_actual_supported_inputs(tmp_path, changed):
+    task_dir, _ = benchmark_files(tmp_path)
+    task = load_task(task_dir)
+    backend, config = FakeBackend(), SForgeConfig(benchmark_dir=tmp_path)
+    before = capture_conditions(task,backend,"d"*64,config)
+    source = "d"*64
+    if changed == "task-bytes":
+        (task_dir/"task.toml").write_text((task_dir/"task.toml").read_text()+"\n# different loaded bytes\n")
+    elif changed == "benchmark-bytes":
+        (tmp_path/"benchmark.toml").write_text((tmp_path/"benchmark.toml").read_text()+"\n# new benchmark bytes\n")
+        load_benchmark.cache_clear()
+    elif changed == "instruction":
+        (task_dir/"instruction.md").write_text("A changed task instruction")
+    elif changed in ("blob","direction"):
+        text=(task_dir/"task.toml").read_text()
+        text=text.replace('direction = "maximize"','direction = "minimize"') if changed=="direction" else text.replace(task.blobs[0].sha256,"e"*64)
+        (task_dir/"task.toml").write_text(text)
+    elif changed == "cpu": config.judge_cpu_limit=3
+    elif changed == "memory": config.judge_mem_limit="192m"
+    elif changed == "environment": config.judge_extra_env={"SYNTHETIC_FACTOR":"2"}
+    elif changed == "source": source="f"*64
+    elif changed == "image": backend.client.images.image="sha256:"+"b"*64
+    after=capture_conditions(load_task(task_dir),backend,source,config)
+    assert before is not None and after is not None and before != after
+    # A later disk edit must not rewrite the original parsed snapshot.
+    if changed in ("task-bytes","benchmark-bytes","instruction","blob","direction"):
+        assert capture_conditions(task,backend,"d"*64,config) == before
+
+
+@pytest.mark.parametrize("missing", ["source","image","loader","backend","game","changed-spec"])
+def test_unavailable_conditions_stay_unknown(tmp_path, missing):
+    task_dir,_=benchmark_files(tmp_path);task=load_task(task_dir)
+    backend,config,source=FakeBackend(),SForgeConfig(benchmark_dir=tmp_path),"d"*64
+    if missing=="source":source=None
+    elif missing=="image":backend.client.images.image=None
+    elif missing=="loader":task=replace(task)
+    elif missing=="backend":backend.backend_name="k8s"
+    elif missing=="game":task=replace(task,judge=replace(task.judge,game_server_cmd="synthetic"))
+    elif missing=="changed-spec":task.submit_paths.append("unobserved")
+    assert capture_conditions(task,backend,source,config) is None
+
+
+def test_caller_labels_are_not_service_facts_and_unknown_feedback_remains_usable(api):
+    client,state,backend,_=api
+    forged={"schema_version":"sforge_run_evaluation_context_v1","evaluation_conditions":{"task_sha256":"f"*64}}
+    registration=register(api,evaluation_context=forged).json()
+    assert registration["evaluation_context"] != forged
+    backend.client.images.image=None
+    unknown=register(api,"unknown-run").json()
+    assert unknown["evaluation_context"]["evaluation_conditions"] is None
+    accepted=submit(api,unknown).json()
+    result=wait_result(client,"/api/v1/result/"+accepted["submission_id"]+"?include_receipt=true")
+    assert result["status"]=="completed" and result["report"]["score"]==3
+    assert result["receipt"]["evaluation_context"]==unknown["evaluation_context"]
+
+
+def test_source_digest_observes_framework_files_not_a_caller_revision(tmp_path):
+    (tmp_path/"a.py").write_text("value=1\n")
+    before=native_source_sha256(tmp_path)
+    (tmp_path/"a.py").write_text("value=2\n")
+    assert native_source_sha256(tmp_path)!=before
diff --git a/tests/test_evaluation_conditions_docker.py b/tests/test_evaluation_conditions_docker.py
new file mode 100644
index 00000000..3a07c750
--- /dev/null
+++ b/tests/test_evaluation_conditions_docker.py
@@ -0,0 +1,407 @@
+# Copyright (c) 2026 ByteDance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Opt-in TOML/Blob -> real Docker -> TCP HTTP qualification, with no fake grader."""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+import os
+import shlex
+import socket
+import tarfile
+import threading
+import time
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import contextmanager
+from functools import partial
+from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
+
+import docker
+import pytest
+import requests
+import uvicorn
+
+from sforge.harness import judge_server as native
+from sforge.harness.agent.base import Agent
+from sforge.harness.backend.docker_backend import DockerBackend
+from sforge.harness.config import SForgeConfig
+from sforge.harness.docker_build import BuildImageError, build_judge_image, ensure_blobs
+from sforge.harness.evaluation_conditions import native_source_sha256
+from sforge.harness.run_agent import run_agent
+from sforge.harness.task_spec import load_benchmark, load_task
+
+pytestmark = pytest.mark.skipif(
+    os.environ.get("SFORGE_CONDITIONS_DOCKER_TEST") != "1",
+    reason="Requires explicit isolated Docker/HTTP evaluation-conditions qualification",
+)
+
+
+def candidate(value):
+    data = f"{value}\n".encode()
+    buf = io.BytesIO()
+    with tarfile.open(fileobj=buf, mode="w:gz") as archive:
+        info = tarfile.TarInfo("answer.txt")
+        info.size = len(data)
+        archive.addfile(info, io.BytesIO(data))
+    return buf.getvalue()
+
+
+@contextmanager
+def serving(app):
+    sock = socket.socket()
+    sock.bind(("0.0.0.0", 0))
+    server = uvicorn.Server(uvicorn.Config(app, log_level="error", lifespan="off"))
+    thread = threading.Thread(target=server.run, kwargs={"sockets": [sock]}, daemon=True)
+    thread.start()
+    try:
+        deadline = time.monotonic() + 10
+        while not server.started and thread.is_alive() and time.monotonic() < deadline:
+            time.sleep(.01)
+        assert server.started, "isolated HTTP server failed to start"
+        yield f"http://127.0.0.1:{sock.getsockname()[1]}"
+    finally:
+        server.should_exit = True
+        thread.join(timeout=10)
+        sock.close()
+        assert not thread.is_alive()
+
+
+class ObservedDockerBackend(DockerBackend):
+    """Observe actual created container configuration without replacing execution."""
+
+    def __init__(self, client):
+        super().__init__(client)
+        self.created = []
+
+    def create_container(self, *args, **kwargs):
+        handle = super().create_container(*args, **kwargs)
+        self.created.append((handle, self.client.api.inspect_container(handle.id)))
+        return handle
+
+
+@pytest.fixture
+def evaluator(tmp_path, monkeypatch):
+    client = docker.from_env()
+    base = os.environ.get("SFORGE_TEST_DOCKER_BASE", "python:3.12-slim")
+    base_image = client.images.get(base)  # Require a cached base, never pull one.
+    platform = base_image.attrs["Os"] + "/" + base_image.attrs["Architecture"]
+    root = tmp_path / "benchmark"
+    task_dir = root / "tasks/toy"
+    (task_dir / "judge").mkdir(parents=True)
+    (task_dir / "environment").mkdir()
+    (task_dir / "instruction.md").write_text("Write answer.txt with a number.\n")
+    (task_dir / "environment/Dockerfile").write_text(f"FROM {base}\n")
+    (task_dir / "judge/Dockerfile").write_text(
+        f"FROM {base}\nCOPY judge /judge\nRUN mkdir -p /work /opt/sforge\nWORKDIR /work\n"
+    )
+    (task_dir / "judge/grade.py").write_text(
+        "import json, os\nfrom pathlib import Path\n"
+        "score = int(Path('answer.txt').read_text()) + int(Path('/judge/data/offset.txt').read_text())"
+        " + int(os.environ.get('SYNTHETIC_OFFSET', '0'))\n"
+        "Path('/opt/sforge/result.json').write_text(json.dumps(dict(score=score, valid=True, summary='synthetic')) )\n"
+    )
+    payload = b"7\n"
+    sha = hashlib.sha256(payload).hexdigest()
+    blob_dir = tmp_path / "blobs"
+    blob_dir.mkdir()
+    (blob_dir / sha).write_bytes(payload)
+    handler = partial(SimpleHTTPRequestHandler, directory=str(blob_dir))
+    blob_server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
+    blob_thread = threading.Thread(target=blob_server.serve_forever, daemon=True)
+    blob_thread.start()
+    name = "conditions-" + uuid.uuid4().hex[:12]
+    (root / "benchmark.toml").write_text(
+        f'name = "{name}"\nblob_url = "http://127.0.0.1:{blob_server.server_port}"\n'
+    )
+    task_path = task_dir / "task.toml"
+    task_path.write_text(
+        f'id = "toy"\nversion = "1.0.0"\n[environment]\ncwd = "/work"\nplatform = "{platform}"\n'
+        '[submission]\npaths = ["answer.txt"]\n[judge]\ncmd = "python /judge/grade.py"\n'
+        'timeout_sec = 20\ncpus = 1\nmemory = "128m"\n[score]\ndirection = "maximize"\n'
+        f'[[blobs]]\npath = "judge/data/offset.txt"\nsha256 = "{sha}"\n'
+    )
+    config = SForgeConfig(benchmark_dir=root, log_dir=tmp_path / "logs")
+    task = load_task(task_dir)
+    owned_tags = [task.judge_image_key]
+    backend = ObservedDockerBackend(client)
+    states = []
+    real_state = native.JudgeState
+
+    def construct(config):
+        state = real_state(config)
+        states.append(state)
+        return state
+
+    monkeypatch.setattr(native, "JudgeState", construct)
+    monkeypatch.setattr(native, "create_backend_from_config", lambda config: backend)
+    try:
+        build_judge_image(task, config)
+        keep = f"{name}.fixture:keep"
+        client.images.get(task.judge_image_key).tag(keep)
+        owned_tags.append(keep)
+        assert (task_dir / "judge/data/offset.txt").read_bytes() == payload
+        yield config, task, client, backend, owned_tags, states
+    finally:
+        for state in states:
+            state._reaper_stop.set()
+            state._reaper_thread.join(timeout=2)
+        for handle, _ in backend.created:
+            try:
+                client.containers.get(handle.id).remove(force=True)
+            except docker.errors.NotFound:
+                pass
+        for tag in owned_tags:
+            try:
+                client.images.remove(tag)
+            except docker.errors.ImageNotFound:
+                pass
+        blob_server.shutdown()
+        blob_server.server_close()
+        blob_thread.join(timeout=2)
+        client.close()
+
+
+def register(url, state, run="run-a", **options):
+    response = requests.post(url + "/api/v1/register", json={
+        "task_id": "toy", "run_id": run, "admin_secret": state.admin_secret, **options,
+    }, timeout=10)
+    response.raise_for_status()
+    value = response.json()
+    return value, url + "/api/v1/feedback/" + value["feedback_relay_id"]
+
+
+def get(url):
+    response = requests.get(url, timeout=10)
+    response.raise_for_status()
+    return response.json()
+
+
+def submit(url, registration, archive, **options):
+    response = requests.post(url + "/api/v1/submit", data={
+        "token": registration["token"], "include_receipt": "true", **options,
+    }, files={"archive": ("submission.tar.gz", archive)}, timeout=15)
+    response.raise_for_status()
+    return response.json()
+
+
+def result(relay, sid):
+    deadline = time.monotonic() + 60
+    while time.monotonic() < deadline:
+        value = get(relay + "/result/" + sid + "?include_receipt=true")
+        if value["status"] in ("completed", "error"):
+            assert value["status"] == "completed", value
+            assert value["report"]["valid"] and not value["report"]["timed_out"], value
+            return value
+        time.sleep(.05)
+    pytest.fail("actual Docker result did not settle")
+
+
+def test_actual_blob_image_resources_concurrent_candidates_and_scoped_reads(evaluator, monkeypatch):
+    config, task, client, backend, _, states = evaluator
+    source_reads = []
+
+    def observed_source():
+        source_reads.append(True)
+        return native_source_sha256()
+
+    monkeypatch.setattr(native, "native_source_sha256", observed_source)
+    app = native.create_app(config)
+    with serving(app) as url:
+        state = states[0]
+        registration, relay = register(url, state, judge_mem_limit="160m")
+        context = get(relay + "/context")
+        conditions = context["evaluation_conditions"]
+        assert context == registration["evaluation_context"]
+        assert conditions["task_sha256"] == hashlib.sha256((task.task_dir / "task.toml").read_bytes()).hexdigest()
+        assert conditions["benchmark_sha256"] == hashlib.sha256((config.benchmark_dir / "benchmark.toml").read_bytes()).hexdigest()
+        assert conditions["judge_image_id"] == client.images.get(task.judge_image_key).id
+        assert conditions["native_source_sha256"] == native_source_sha256()
+        with ThreadPoolExecutor(2) as pool:
+            repeated = list(pool.map(lambda _: register(url, state, judge_mem_limit="160m")[0], range(2)))
+            assert repeated == [registration, registration]
+            archives = [candidate(3), candidate(19)]
+            accepted = list(pool.map(lambda data: submit(url, registration, data), archives))
+        reads = [result(relay, value["submission_id"]) for value in accepted]
+        assert [value["report"]["score"] for value in reads] == [10, 26]
+        assert {value["receipt"]["round_id"] for value in reads} == {"agent-1", "agent-2"}
+        for archive, value in zip(archives, reads):
+            receipt = value["receipt"]
+            assert receipt["evaluation_context"] == context
+            assert receipt["archive_sha256"] == hashlib.sha256(archive).hexdigest()
+            assert receipt["archive_bytes"] == len(archive)
+            graded = config.log_dir / "runs/run-a/toy/submissions" / receipt["round_id"] / "submission.tar.gz"
+            assert graded.read_bytes() == archive
+            plain = get(url + "/api/v1/result/" + value["submission_id"])
+            assert plain == {key: item for key, item in value.items() if key != "receipt"}
+        for _, observed in backend.created:
+            assert observed["Image"] == conditions["judge_image_id"]
+            assert observed["HostConfig"]["NanoCpus"] == 1_000_000_000
+            assert observed["HostConfig"]["Memory"] == 160 * 1024 * 1024
+            assert observed["HostConfig"]["NetworkMode"] == "none"
+        _, other = register(url, state, "run-b")
+        assert requests.get(other + "/result/" + accepted[0]["submission_id"], timeout=5).status_code == 404
+        auto = submit(url, registration, candidate(99), kind="auto", admin_secret=state.admin_secret)
+        assert requests.get(relay + "/result/" + auto["submission_id"], timeout=5).status_code == 404
+        # Wait for the owned auto container before fixture cleanup.
+        while get(url + "/api/v1/result/" + auto["submission_id"])["status"] in ("queued", "running"):
+            time.sleep(.05)
+        for _ in range(3):
+            assert get(relay + "/context") == context
+            assert get(relay + "/result/" + accepted[0]["submission_id"] + "?include_receipt=true") == reads[0]
+        assert len(source_reads) == 1
+
+
+def test_loaded_task_blob_and_image_are_frozen_while_fresh_run_observes_changes(evaluator):
+    config, task, client, backend, owned_tags, states = evaluator
+    app = native.create_app(config)
+    with serving(app) as url:
+        state = states[0]
+        original, old_relay = register(url, state)
+        before = original["evaluation_context"]
+        (task.task_dir / "judge/grade.py").write_text(
+            (task.task_dir / "judge/grade.py").read_text().replace("score=score", "score=score + 100")
+        )
+        task_path = task.task_dir / "task.toml"
+        changed_data = b"11\n"
+        changed_sha = hashlib.sha256(changed_data).hexdigest()
+        (config.benchmark_dir.parent / "blobs" / changed_sha).write_bytes(changed_data)
+        task_path.write_text(task_path.read_text()
+                             .replace('direction = "maximize"', 'direction = "minimize"')
+                             .replace(task.blobs[0].sha256, changed_sha))
+        benchmark_path = config.benchmark_dir / "benchmark.toml"
+        benchmark_path.write_text(benchmark_path.read_text() + "# another loaded revision\n")
+        load_benchmark.cache_clear()
+        changed_task = load_task(task.task_dir)
+        owned_tags.append(changed_task.judge_image_key)
+        build_judge_image(changed_task, config)
+        replacement = client.images.get(changed_task.judge_image_key)
+        replacement.tag(task.judge_image_key)
+        assert get(old_relay + "/context") == before
+        state.load_tasks()
+        state.config.judge_extra_env["SYNTHETIC_OFFSET"] = "5"
+        response = requests.post(url + "/api/v1/register", json={
+            "task_id": "toy", "run_id": "run-a", "admin_secret": state.admin_secret,
+        }, timeout=10)
+        assert response.status_code == 409 and "fresh run_id" in response.text
+        new, new_relay = register(url, state, "run-b", judge_cpu_limit=2, judge_mem_limit="192m")
+        after = new["evaluation_context"]
+        for field in ("task_sha256", "benchmark_sha256", "task_spec_sha256", "judge_image_id", "execution_config_sha256", "score_direction"):
+            assert before["evaluation_conditions"][field] != after["evaluation_conditions"][field]
+        old_result = result(old_relay, submit(url, original, candidate(3))["submission_id"])
+        new_result = result(new_relay, submit(url, new, candidate(3))["submission_id"])
+        assert old_result["report"]["score"] == 10
+        assert new_result["report"]["score"] == 119  # actual new Blob 11 + grader 100 + environment 5
+        assert old_result["receipt"]["evaluation_context"] == before
+        assert new_result["receipt"]["evaluation_context"] == after
+        result(new_relay, submit(url, new, candidate(19))["submission_id"])
+        assert get(url + "/api/v1/history?token=" + new["token"])["best_score"] == 119
+        for _, observed in backend.created[1:]:
+            assert observed["Image"] == after["evaluation_conditions"]["judge_image_id"]
+            assert observed["HostConfig"]["NanoCpus"] == 2_000_000_000
+            assert observed["HostConfig"]["Memory"] == 192 * 1024 * 1024
+
+
+def test_missing_image_is_unknown_and_blob_mismatch_cannot_build_a_false_fact(evaluator):
+    config, task, client, _, _, states = evaluator
+    image = client.images.get(task.judge_image_key)
+    client.images.remove(task.judge_image_key)
+    app = native.create_app(config)
+    with serving(app) as url:
+        state = states[0]
+        registration, relay = register(url, state)
+        assert registration["evaluation_context"]["evaluation_conditions"] is None
+        with pytest.raises(docker.errors.ImageNotFound):
+            client.images.get(task.judge_image_key)
+        image.tag(task.judge_image_key)
+        read = result(relay, submit(url, registration, candidate(3))["submission_id"])
+        assert read["report"]["score"] == 10
+        assert read["receipt"]["evaluation_context"]["evaluation_conditions"] is None
+        state.submissions[read["submission_id"]].pop("evaluation_context")
+        legacy = get(relay + "/result/" + read["submission_id"] + "?include_receipt=true")
+        assert legacy["report"] == read["report"] and legacy["receipt"]["evaluation_context"] is None
+        (task.task_dir / "judge/data/offset.txt").write_bytes(b"changed")
+        blob_store = task.task_dir.parent.parent.parent / "blobs" / task.blobs[0].sha256
+        blob_store.write_bytes(b"incorrect download")
+        with pytest.raises(BuildImageError, match="downloaded sha256"):
+            ensure_blobs(task)
+        assert (task.task_dir / "judge/data/offset.txt").read_bytes() == b"changed"
+
+
+def test_actual_run_agent_delivers_relay_and_generated_client_keeps_default_feedback(evaluator, monkeypatch):
+    config, task, client, backend, owned_tags, states = evaluator
+    tag = task.benchmark.name + ".client:qualification"
+    owned_tags.append(tag)
+    # The shipped Bash client requires curl, jq and GNU userland. Run it on
+    # Linux rather than substituting host tools or changing its generated code.
+    base = os.environ.get("SFORGE_TEST_DOCKER_BASE", "python:3.12-slim")
+    dockerfile = (
+        f"FROM {base}\nRUN apt-get update && apt-get install -y --no-install-recommends curl jq "
+        "&& rm -rf /var/lib/apt/lists/*\nRUN mkdir -p /work\nWORKDIR /work\n"
+    )
+    image, build_log = client.images.build(fileobj=io.BytesIO(dockerfile.encode()), tag=tag, rm=True)
+    list(build_log)
+    image.tag(task.work_image_key)
+    owned_tags.append(task.work_image_key)
+    probe_run = "probe-" + task.benchmark.name
+    # Exercise the owning run_agent entrypoint without installing or calling a
+    # model. Only the probe agent is synthetic; its container, registration,
+    # installed submit client, grading and feedback all use the native path.
+    class FeedbackProbe(Agent):
+        name = "feedback-probe"
+        default_version = "fixture"
+        api_key_env = "SYNTHETIC_KEY"
+
+        def install(self, backend, handle, logger):
+            pass
+
+        def format_run_cmd(self, *args, **kwargs):
+            probe = (
+                "import json,os,urllib.request; "
+                "ctx=json.load(urllib.request.urlopen(os.environ['SFORGE_FEEDBACK_URL']+'/context')); "
+                f"assert ctx['run_id']=={probe_run!r}; print('CONTEXT '+json.dumps(ctx))"
+            )
+            return "python -c " + shlex.quote(probe) + " && printf '3\\n' > answer.txt && sforge-submit --details"
+
+    from sforge.harness import network_isolation
+    # Do not inspect or clean machine-wide iptables rules in an isolated test.
+    monkeypatch.setattr(network_isolation, "cleanup_stale_chains", lambda logger: None)
+    app = native.create_app(config)
+    with serving(app) as url:
+        state = states[0]
+        port = url.rsplit(":", 1)[1]
+        host = os.environ.get("SFORGE_TEST_JUDGE_HOST", "host.docker.internal")
+        executed = run_agent(
+            task, FeedbackProbe(config), config, backend, probe_run, timeout=45,
+            judge_url=f"http://{host}:{port}", host_judge_url=url,
+            disable_stop_hook=True, disable_auto_eval=True, disable_auto_resume=True,
+        )
+        assert not executed.timed_out, executed.agent_output
+        assert executed.best_score == 10 and executed.agent_submissions == 1
+        assert "Score:       10" in executed.agent_output and "synthetic" in executed.agent_output
+        token, info = next(iter(state.tokens.items()))
+        observed_context = json.loads(next(line[8:] for line in executed.agent_output.splitlines() if line.startswith("CONTEXT ")))
+        assert observed_context == info["evaluation_context"]
+        assert "token" not in observed_context and "admin_secret" not in observed_context
+        relay = url + "/api/v1/feedback/" + info["feedback_relay_id"]
+        history = get(url + "/api/v1/history?token=" + token)
+        sid = history["entries"][0]["submission_id"]
+        enriched = result(relay, sid)
+        assert enriched["receipt"]["evaluation_context"] == observed_context
+        graded = config.log_dir / "runs" / probe_run / "toy/submissions/agent-1/submission.tar.gz"
+        assert hashlib.sha256(graded.read_bytes()).hexdigest() == enriched["receipt"]["archive_sha256"]
+        assert "receipt" not in get(url + "/api/v1/result/" + sid)