Record per-case Docker image observations in EdgeBench

benchmark/edgebench/README.md

@@ -114,6 +114,25 @@ its PR manifest current. Changing the default repository/branch requires updatin
 the single source file and this protocol; adding a second contributor-specific
 default would split the source of truth again.
 
+### Inspect cached images by case
+
+The worker's existing Docker preflight records the local image ID and logical
+size for the work and judge images in `runtime-receipt.json`. Review observations
+across selected run directories without reading trajectories or contacting Docker:
+
+```sh
+python -m benchmark.edgebench.image_inventory \
+  --log-dir /data/private-runs --log-dir /data/older-runs --task TASK_ID
+```
+
+Omit `--task` to include every case in those directories. The report groups
+observations by task specification digest, image role and image ID. It counts
+older receipts without observations and keeps distinct image revisions separate.
+Docker's logical image sizes include shared layers. Use them for cache inventory,
+and measure incremental disk use and runtime memory separately. Observations
+describe the worker's local Docker daemon at preflight. A remote judge requires
+its own image provenance, and a later retag can change what gets launched.
+
 ## Worker profiles
 
 The five worker profiles are `official`, `single`, `native-goal`,

benchmark/edgebench/image_inventory.py

@@ -0,0 +1,106 @@
+"""Read-only inventory of local Docker images observed by EdgeBench preflight."""
+
+from __future__ import annotations
+
+import argparse
+from collections import Counter, defaultdict
+import json
+from pathlib import Path
+import re
+
+
+def _digest(value, pattern):
+    return value if isinstance(value, str) and re.fullmatch(pattern, value) else None
+
+
+def _size(value):
+    return value if type(value) is int and value >= 0 else None
+
+
+def inventory(log_dirs, *, task=None):
+    """Reduce selected run roots without reading submissions or contacting Docker."""
+    receipts = set()
+    unresolved_receipts = set()
+    for log_dir in log_dirs:
+        root = log_dir.resolve(strict=True)
+        if not root.is_dir():
+            raise ValueError(f"Expected an EdgeBench log directory: {root}")
+        for path in root.glob("runs/*/*/runtime-receipt.json"):
+            case_id = path.parent.name
+            if task is not None and case_id != task:
+                continue
+            try:
+                receipt = path.resolve()
+            except (OSError, RuntimeError):
+                unresolved_receipts.add(path)
+                continue
+            if not receipt.is_relative_to(root):
+                continue
+            parts = receipt.relative_to(root).parts
+            if len(parts) == 4 and parts[0] == "runs" and parts[2:] == (case_id, path.name):
+                receipts.add(receipt)
+
+    cases = {}
+    unreadable = len(unresolved_receipts)
+    for path in receipts:
+        try:
+            receipt = json.loads(path.read_bytes())
+        except (OSError, ValueError):
+            unreadable += 1
+            continue
+        if not isinstance(receipt, dict) or receipt.get("task") != path.parent.name:
+            unreadable += 1
+            continue
+        case_id = receipt["task"]
+        task_sha = _digest(receipt.get("task_sha256"), r"[0-9a-f]{64}")
+        images, counts = cases.setdefault((case_id, task_sha), (defaultdict(Counter), Counter()))
+        observations = receipt.get("local_docker_images")
+        if not isinstance(observations, dict):
+            counts["without_preflight"] += 1
+            continue
+        counts["with_preflight"] += 1
+        for role in ("work", "judge"):
+            entry = observations.get(role)
+            entry = entry if isinstance(entry, dict) else {}
+            image_id = _digest(entry.get("image_id"), r"sha256:[0-9a-f]{64}")
+            images[role, image_id][_size(entry.get("logical_size_bytes"))] += 1
+
+    rows = []
+    for case_id, task_sha in sorted(cases, key=lambda key: (key[0], key[1] or "")):
+        images, counts = cases[case_id, task_sha]
+        rows.append({
+            "task": case_id,
+            "task_sha256": task_sha,
+            "receipts_with_preflight": counts["with_preflight"],
+            "receipts_without_preflight": counts["without_preflight"],
+            "images": [
+                {
+                    "role": role,
+                    "image_id": image_id,
+                    "logical_sizes_bytes": sorted(size for size in sizes if size is not None),
+                    "observations": sum(sizes.values()),
+                    "observations_without_size": sizes[None],
+                }
+                for (role, image_id), sizes in sorted(
+                    images.items(), key=lambda entry: (entry[0][0], entry[0][1] or "")
+                )
+            ],
+        })
+    return {"cases": rows, "unreadable_receipts": unreadable}
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--log-dir", type=Path, action="append", required=True,
+                        help="Local run root. Repeat for additional cohorts.")
+    parser.add_argument("--task", help="Only read receipts for this task ID.")
+    args = parser.parse_args(argv)
+    try:
+        result = inventory(args.log_dir, task=args.task)
+    except (OSError, ValueError) as exc:
+        parser.error(str(exc))
+    print(json.dumps(result, indent=2))
+
+
+if __name__ == "__main__":
+    main()

benchmark/edgebench/run.py

@@ -481,9 +481,9 @@ def main(argv=None):
                                      docker_host_bridge=args.docker_host_bridge,
                                      research_sidecar=research_sidecar,
                                      **({"checkpoint": checkpoint} if checkpoint is not None else {}))
-    for image in (task.work_image_key, task.judge_image_key):
-        if not backend.image_exists(image):
-            raise RuntimeError(f"Missing native image: {image}")
+    local_docker_images = backend.observe_local_images(
+        work_key=task.work_image_key, judge_key=task.judge_image_key,
+    )
     if warm_start is not None:
         activated = restore(args.checkpoint_controller_file, root=capsule,
             expected_sha256=snapshot["native_capsule_sha256"], run_id=args.run_id,
@@ -509,6 +509,7 @@ def main(argv=None):
         **({"docker_host_bridge": True} if args.docker_host_bridge else {}),
         "loopx_commit": pins[0], "runner_commit": pins[1],
         "sforge_source": active_source_receipt(),
+        "local_docker_images": local_docker_images,
         **turn_envelope_receipt(explicit=args.turn_envelope),
         **({"context_delivery": "on"} if args.context_delivery != "off" else {}),
         **({"execution_observation": args.execution_observation}

benchmark/runtime/sforge_backend.py

@@ -9,6 +9,7 @@
 import tarfile
 from pathlib import Path, PurePosixPath
 
+from docker.errors import ImageNotFound
 from sforge.harness.backend.docker_backend import DockerBackend
 from sforge.harness.network_isolation import AllowedEndpoint, resolve_hostname
 
@@ -43,6 +44,21 @@ def __init__(self, *, log_dir: Path, logger, oauth_proxy: bool = False,
         if not oauth_proxy and not self.auth_ips:
             raise RuntimeError("Cannot resolve Codex OAuth endpoint")
 
+    def observe_local_images(self, *, work_key, judge_key):
+        """Check both required local images and retain their Docker metadata."""
+        observations = {}
+        for key in dict.fromkeys((work_key, judge_key)):
+            try:
+                image = self.client.images.get(key)
+            except ImageNotFound as exc:
+                raise RuntimeError(f"Missing native image: {key}") from exc
+            size = image.attrs.get("Size")
+            observations[key] = {
+                "image_id": image.id,
+                "logical_size_bytes": size if type(size) is int and size >= 0 else None,
+            }
+        return {"work": observations[work_key], "judge": observations[judge_key]}
+
     def registration_options(self, run_agent):
         # Older frozen providers expose registration through container env;
         # newer providers keep that credential exclusively on the host.

benchmark/tests/test_checkpoint_storage.py

@@ -136,7 +136,7 @@ def backend(**kwargs):
             assert kwargs["checkpoint"].destination == expected
         else:
             assert not calls and "checkpoint" not in kwargs
-        return SimpleNamespace(image_exists=lambda image: True, registration_options=lambda f: {})
+        return SimpleNamespace(observe_local_images=lambda **keys: {}, registration_options=lambda f: {})
 
     monkeypatch.setattr(run, "RecordingDockerBackend", backend)
     def before_solver(**kwargs):

benchmark/tests/test_codex_readiness_budget.py

@@ -132,7 +132,7 @@ def test_edgebench_runner_passes_selected_setup_budget_before_model(tmp_path, mo
         cwd="/task", work_image_key="work", judge_image_key="judge", internet=False,
         game_mode=False, submit_paths=["solver.py"], work=SimpleNamespace(agent_query="Original acceptance."),
         judge=SimpleNamespace(selection="score_first", score_direction="maximize")))
-    backend = SimpleNamespace(image_exists=lambda image: True,
+    backend = SimpleNamespace(observe_local_images=lambda **keys: {},
                               exec_run=lambda *a, **k: SimpleNamespace(exit_code=0))
     monkeypatch.setattr(run, "RecordingDockerBackend", lambda **k: backend)
 

benchmark/tests/test_edgebench_feedback.py

@@ -361,7 +361,7 @@ def qualify(self):
     constructed = {}
     def backend(**kwargs):
         constructed.update(kwargs)
-        return SimpleNamespace(image_exists=lambda image: True,
+        return SimpleNamespace(observe_local_images=lambda **keys: {},
             registration_options=lambda runner: {"host_registration_callback": kwargs["feedback"].bind})
     monkeypatch.setattr(run, "RecordingDockerBackend", backend)
     def handoff(**kwargs):

benchmark/tests/test_edgebench_image_inventory.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from benchmark.edgebench.image_inventory import main
+
+
+def test_inventory_keeps_image_revisions_and_legacy_receipts_separate(tmp_path, capsys):
+    old_image, new_image = "sha256:" + "a" * 64, "sha256:" + "b" * 64
+
+    def record(root, run, *, task="case", task_sha="c" * 64, image_id=None, size=None):
+        receipt = root / "runs" / run / task / "runtime-receipt.json"
+        receipt.parent.mkdir(parents=True)
+        body = {"task": task, "task_sha256": task_sha, "run_id": run,
+                "private_field": "never export this"}
+        if image_id is not None:
+            image = {"image_id": image_id, "logical_size_bytes": size}
+            body["local_docker_images"] = {"work": image, "judge": image}
+        receipt.write_text(json.dumps(body))
+        return receipt
+
+    first, second = tmp_path / "first", tmp_path / "second"
+    record(first, "legacy")
+    record(first, "observed", image_id=old_image, size=100)
+    record(second, "another", image_id=old_image, size=None)
+    record(second, "retagged", image_id=new_image, size=250)
+    record(second, "new-task", task_sha="d" * 64, image_id=old_image, size=100)
+    record(first, "ignored", task="other", image_id=new_image, size=200)
+    record(first, "unreadable").write_text("{")
+    cyclic = record(first, "cyclic")
+    cyclic.unlink()
+    cyclic.symlink_to(cyclic.name)
+    escape = record(first, "escape")
+    escape.unlink()
+    escape.symlink_to(record(tmp_path / "outside", "outside", task_sha="e" * 64))
+
+    main(["--log-dir", str(first), "--log-dir", str(second), "--log-dir", str(first), "--task", "case"])
+    output = capsys.readouterr().out
+    assert "never export this" not in output
+    result = json.loads(output)
+    assert result["unreadable_receipts"] == 2
+    original, revised = result["cases"]
+    assert original["task_sha256"] == "c" * 64
+    assert original["receipts_with_preflight"] == 3
+    assert original["receipts_without_preflight"] == 1
+    assert revised["task_sha256"] == "d" * 64
+    assert revised["receipts_with_preflight"] == 1
+    assert {row["role"] for row in original["images"]} == {"judge", "work"}
+    assert [row for row in original["images"] if row["role"] == "work"] == [
+        {"role": "work", "image_id": old_image, "logical_sizes_bytes": [100],
+         "observations": 2, "observations_without_size": 1},
+        {"role": "work", "image_id": new_image, "logical_sizes_bytes": [250],
+         "observations": 1, "observations_without_size": 0},
+    ]
+
+
+def test_inventory_keeps_resolved_receipts_in_their_selected_case(tmp_path, capsys):
+    def write(relative, task):
+        path = tmp_path / relative
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps({"task": task}))
+        return path
+
+    case_a = write("runs/first/case-a/runtime-receipt.json", "case-a")
+    case_b = write("runs/second/case-b/runtime-receipt.json", "case-b")
+    sidecar = write("runs/first/case-a/unrelated.json", "case-a")
+    misplaced = write("notes/case-a/runtime-receipt.json", "case-a")
+    for run, target in (("alias", case_a), ("foreign", case_b),
+                        ("sidecar", sidecar), ("misplaced", misplaced)):
+        path = tmp_path / "runs" / run / "case-a" / "runtime-receipt.json"
+        path.parent.mkdir(parents=True)
+        path.symlink_to(target)
+
+    args = ["--log-dir", str(tmp_path)]
+    for filters, expected in ((["--task", "case-a"], ["case-a"]), ([], ["case-a", "case-b"])):
+        main([*args, *filters])
+        result = json.loads(capsys.readouterr().out)
+        assert [row["task"] for row in result["cases"]] == expected
+        assert [row["receipts_without_preflight"] for row in result["cases"]] == [1] * len(expected)
+        assert result["unreadable_receipts"] == 0
+
+
+def test_preflight_looks_up_each_required_image_once_and_keeps_missing_image_gate(tmp_path):
+    pytest.importorskip("sforge")
+    pytest.importorskip("harbor")
+    from docker.errors import ImageNotFound
+    from benchmark.runtime.sforge_backend import RecordingDockerBackend
+
+    queries = []
+
+    def get(key):
+        queries.append(key)
+        if key == "missing":
+            raise ImageNotFound("missing")
+        return SimpleNamespace(id="sha256:" + "a" * 64, attrs={"Size": 42} if key == "shared" else {})
+
+    backend = RecordingDockerBackend(log_dir=tmp_path, logger=None, oauth_proxy=True,
+                                    client=SimpleNamespace(images=SimpleNamespace(get=get)))
+    observed = backend.observe_local_images(work_key="shared", judge_key="shared")
+    assert queries == ["shared"]
+    assert observed["work"] == observed["judge"] == {
+        "image_id": "sha256:" + "a" * 64, "logical_size_bytes": 42,
+    }
+    observed = backend.observe_local_images(work_key="shared", judge_key="old-engine")
+    assert observed["judge"]["logical_size_bytes"] is None
+    with pytest.raises(RuntimeError, match="Missing native image: missing"):
+        backend.observe_local_images(work_key="shared", judge_key="missing")

benchmark/tests/test_native_goal_outcomes.py

@@ -184,7 +184,7 @@ def test_edgebench_cli_persists_native_outcome_even_when_transport_exits_zero(tm
         internet=False, game_mode=False, submit_paths=["solver.py"],
         work=SimpleNamespace(agent_query="Synthetic task"),
         judge=SimpleNamespace(selection="score_first", score_direction="maximize")))
-    backend = SimpleNamespace(image_exists=lambda image: True,
+    backend = SimpleNamespace(observe_local_images=lambda **keys: {},
                               execution_receipt={"exit_code": 0, "execution_started": True},
                               native_goal_receipt={"native_goal_outcome": outcome} if outcome else None,
                               provider_failure_receipt=({

benchmark/tests/test_research_artifacts.py

@@ -88,7 +88,7 @@ def test_native_entry_installs_identical_research_scope_and_declares_checkpoint_
         judge=SimpleNamespace(selection="score_first", score_direction="maximize")))
     def backend(**kwargs):
         assert kwargs["research_sidecar"] is True
-        return SimpleNamespace(image_exists=lambda image: True)
+        return SimpleNamespace(observe_local_images=lambda **keys: {})
     monkeypatch.setattr(run, "RecordingDockerBackend", backend)
     capture = profile.startswith("heartbeat-")
     request = {"destination": "/logs/agent/checkpoints", "components": [

benchmark/tests/test_sforge_runtime.py

@@ -934,7 +934,7 @@ def test_edgebench_native_goal_preserves_task_with_resolved_entry_and_treatment(
         work=SimpleNamespace(agent_query="fixture task"),
         judge=SimpleNamespace(selection="score_first", score_direction="maximize")))
     monkeypatch.setenv("CODEX_AUTH_JSON_PATH", "/synthetic-credential")
-    monkeypatch.setattr(run, "RecordingDockerBackend", lambda **k: SimpleNamespace(image_exists=lambda image: True))
+    monkeypatch.setattr(run, "RecordingDockerBackend", lambda **k: SimpleNamespace(observe_local_images=lambda **keys: {}))
     def stop_before_solver(**kwargs):
         # Exercise the real adapter at the CLI handoff, including both context
         # modes. SForge must not resume an exited LoopX scheduler.
@@ -1391,10 +1391,14 @@ def test_edgebench_cli_merges_sforge_env_with_explicit_runner_config(
         cwd='/task', work_image_key='work', judge_image_key='judge', internet=False,
         game_mode=False, submit_paths=['solver.py'], work=SimpleNamespace(agent_query='fixture task'),
         judge=SimpleNamespace(selection='score_first', score_direction='maximize')))
+    images = {
+        'work': {'image_id': 'sha256:' + 'a' * 64, 'logical_size_bytes': 100},
+        'judge': {'image_id': 'sha256:' + 'b' * 64, 'logical_size_bytes': 200},
+    }
     def backend(**kwargs):
         assert kwargs['oauth_proxy'] is custom
         assert kwargs.get('docker_host_bridge', False) is bridge
-        return SimpleNamespace(image_exists=lambda image: True)
+        return SimpleNamespace(observe_local_images=lambda **keys: images)
     monkeypatch.setattr(run, 'RecordingDockerBackend', backend)
     def stop_before_solver(**kwargs):
         config = kwargs['config']
@@ -1417,6 +1421,7 @@ def stop_before_solver(**kwargs):
         run.main(args)
     receipt = (tmp_path / 'runs/route/fixture/runtime-receipt.json').read_text()
     assert 'synthetic-provider-secret' not in receipt
+    assert json.loads(receipt)['local_docker_images'] == images
     assert json.loads(receipt).get('provider_request_admission') == ('on' if provider_request_admission else None)
     assert json.loads(receipt).get('docker_host_bridge', False) is bridge
 
@@ -1440,7 +1445,7 @@ def test_native_checkpoint_storage_guidance_reaches_both_prompt_surfaces(tmp_pat
         cwd="/task", work_image_key="work", judge_image_key="judge", internet=False,
         game_mode=False, submit_paths=["solver.py"], work=SimpleNamespace(agent_query="Original acceptance."),
         judge=SimpleNamespace(selection="score_first", score_direction="maximize")))
-    monkeypatch.setattr(run, "RecordingDockerBackend", lambda **k: SimpleNamespace(image_exists=lambda image: True))
+    monkeypatch.setattr(run, "RecordingDockerBackend", lambda **k: SimpleNamespace(observe_local_images=lambda **keys: {}))
     expected_delivery = "on" if context_delivery is None else context_delivery
     def before_solver(**kwargs):
         worker = kwargs["agent"]
@@ -1495,7 +1500,7 @@ def test_edgebench_cli_reuses_local_probe_and_reports_selection_before_model(tmp
         cwd="/task", work_image_key="work", judge_image_key="judge", internet=False,
         game_mode=False, submit_paths=["solver.py"], work=SimpleNamespace(agent_query="Original acceptance."),
         judge=SimpleNamespace(selection="score_first", score_direction="maximize")))
-    backend = SimpleNamespace(image_exists=lambda image: True,
+    backend = SimpleNamespace(observe_local_images=lambda **keys: {},
                               exec_run=lambda *a, **k: SimpleNamespace(exit_code=0))
     monkeypatch.setattr(run, "RecordingDockerBackend", lambda **k: backend)