From b2e7133b207f1a3d6eb827ebb58f317b5dead568 Mon Sep 17 00:00:00 2001
From: sora <Wanght@meta-stone.com>
Date: Mon, 24 Aug 2026 07:11:03 +0000
Subject: [PATCH] Add sandbox layer (docker exec hard-isolation + serve envs,
 refcounted acquire/release, atexit teardown, bind-mount sharing) and agent
 evaluation driver (message pump drive(), Trajectory, bfcl_mock env with
 official call-sequence scoring, mock:fc oracle); Deployer delegates to
 sandbox.serve_env; code_any extractor fixes indentation-stripping; CLI --env

---
 README.md                           |  71 ++++++++++--
 evalharness/agent/__init__.py       |  14 +++
 evalharness/agent/envs/__init__.py  |   0
 evalharness/agent/envs/bfcl_mock.py |  55 ++++++++++
 evalharness/agent/loop.py           | 119 ++++++++++++++++++++
 evalharness/cli.py                  |   3 +-
 evalharness/eval/extractor.py       |  15 ++-
 evalharness/eval/recipes/agent.py   |  86 ++++++++++++---
 evalharness/eval/runner.py          |   7 +-
 evalharness/eval/scorer.py          |  63 ++++++++++-
 evalharness/model/adapter.py        |  64 ++++++++---
 evalharness/model/deployer.py       |  50 ++-------
 evalharness/model/runner.py         |  82 ++++++++++----
 evalharness/sandbox/__init__.py     |  31 ++++++
 evalharness/sandbox/base.py         | 142 ++++++++++++++++++++++++
 evalharness/sandbox/docker.py       | 163 ++++++++++++++++++++++++++++
 evalharness/sandbox/local.py        |  45 ++++++++
 tests/test_sandbox_agent.py         | 115 ++++++++++++++++++++
 18 files changed, 1013 insertions(+), 112 deletions(-)
 create mode 100644 evalharness/agent/__init__.py
 create mode 100644 evalharness/agent/envs/__init__.py
 create mode 100644 evalharness/agent/envs/bfcl_mock.py
 create mode 100644 evalharness/agent/loop.py
 create mode 100644 evalharness/sandbox/__init__.py
 create mode 100644 evalharness/sandbox/base.py
 create mode 100644 evalharness/sandbox/docker.py
 create mode 100644 evalharness/sandbox/local.py
 create mode 100644 tests/test_sandbox_agent.py

diff --git a/README.md b/README.md
index 760b747..cac53ae 100644
--- a/README.md
+++ b/README.md
@@ -1,10 +1,11 @@
 # EvalHarness
 
 A plugin-based LLM/agent evaluation harness. **Currently: data + evaluation +
-model layers** (datasets/eval-recipes/model-adapters/deployers as plugins,
-lazy materialization cache, official-aligned scorers, async generation,
-report artifacts & console visualization). Sandbox/agent/tool/skill layers
-land one at a time.
+model + sandbox + agent-driver layers** (datasets/eval-recipes/model-adapters/
+deployers/sandboxes/environments as plugins, lazy materialization cache,
+official-aligned scorers, async generation, sandboxed code execution, agent
+message pump, report artifacts & console visualization). Tool/skill layers
+land next.
 
 ## Features
 
@@ -155,6 +156,7 @@ never re-queried.
 evalharness eval run gsm8k --model mock:boxed --limit 100          # offline pipeline check
 evalharness eval run gsm8k --model openai/http://gpu03:8000/v1?qwen3-8b
 evalharness eval run hle   --model openai/...?qwen3-8b --judge openai/...?gpt-4o
+evalharness eval run bfcl_v3 --model mock:fc --env bfcl_mock     # agent pump
 # future: --model deploy:vllm/qwen3-8b  (Deployer pulls a pinned docker env)
 ```
 
@@ -178,6 +180,46 @@ Design:
   a semaphore (default 32), collects raws + per-sample usage, then hands them
   to the synchronous `evaluate()`. Data/scoring stay sync (fast, CPU/disk).
 
+## Sandbox layer (environments for BOTH eval execution and model serving)
+
+One docker implementation, two faces:
+
+- **exec()** runs untrusted model-generated code hard-isolated:
+  `--network none`, cpu/mem/pids caps, read-only rootfs, tmpfs /tmp.
+  Host file sharing via **bind mounts** (`mounts={'/out': host_dir}`) —
+  artifacts land on the host directly, no `docker cp`.
+- **serve()** trusted engine containers (vllm/sglang) with network + GPU
+  passthrough; consumed by the model Deployer through the same layer.
+- **Lifecycle**: refcounted `acquire()/release()`; containers stop+rm at
+  refcount 0 or process exit (atexit); **images are never auto-deleted** —
+  re-acquire re-runs the local image instantly.
+
+```python
+from evalharness.sandbox import get_sandbox
+r = get_sandbox('docker').exec({'main.py': 'print(42)'})   # or 'local' for dev
+```
+
+## Agent evaluation driver (message pump, not a thinking framework)
+
+We EVALUATE agents: the model under test thinks; we only execute its
+tool_calls against Environment plugins and feed observations back.
+
+```python
+from evalharness.model import run_eval
+report = await run_eval(ds, 'openai/http://gpu03:8000/v1?qwen3-8b',
+                        env='bfcl_mock')        # agent pump per sample
+# CLI: evalharness eval run bfcl_v3 --model mock:fc --env bfcl_mock
+```
+
+- `agent/loop.py::drive()` — pump until no more calls / env done / max_turns;
+  records a full `Trajectory` (messages, per-turn usage, env final state)
+  into `SampleResult.trajectory / env_state`.
+- `agent/envs/bfcl_mock.py` — BFCL official-style: record call sequence,
+  compare against ground_truth (`env_reward` scorer), incl. irrelevance
+  categories (correct = call nothing). tau2 / swe envs land later on the
+  same `Environment` contract.
+- Single-turn is the degenerate case: no env -> one generate, done.
+
 ## Built-in datasets (28, official sources)
 
 | Family | Datasets (source) |
@@ -293,6 +335,13 @@ EvalHarness/
 │   │   ├── adapter.py             # @register_adapter: openai_compatible / mock
 │   │   ├── deployer.py            # @register_deployer: vllm / sglang / external
 │   │   └── runner.py              # async run_eval(): generate -> evaluate
+│   ├── sandbox/                   # ---- environment layer ----
+│   │   ├── base.py                # Sandbox iface + refcounted acquire/release + atexit
+│   │   ├── docker.py              # exec (isolated) + serve (engines) one impl
+│   │   └── local.py               # dev-only, no isolation
+│   ├── agent/                     # ---- agent evaluation driver ----
+│   │   ├── loop.py                # drive(): message pump + Trajectory
+│   │   └── envs/bfcl_mock.py      # BFCL official-style env (tau2/swe later)
 │   ├── eval/                      # ---- evaluation layer ----
 │   │   ├── record.py              # SampleResult / EvalReport artifacts
 │   │   ├── extractor.py           # answer-extraction primitives (+cascades)
@@ -318,10 +367,16 @@ EvalHarness/
 - [x] Model layer (async ModelAdapter openai_compatible+mock, ModelOutput
       with tool_calls, Deployer registry vllm/sglang/external + models.yaml
       env pinning, run_eval generate->score)
-- [ ] Agent layer (loops: single_turn fast path today, react/plan_execute;
-      environments: tau2 user-sim, swe docker; SampleResult.trajectory ready)
-- [ ] Sandbox layer (materialize `Sample.sandbox`: lazy per-instance image
-      pull, refcounted image unload, container lifecycle; `requires` gating)
+- [x] Sandbox layer (docker exec hard-isolation + serve environments,
+      refcounted acquire/release, atexit teardown, images kept, bind-mount
+      host sharing; Deployer now consumes it)
+- [x] Agent evaluation driver (message pump + Trajectory + bfcl_mock env
+      with official call-sequence scoring; tau2/swe envs pending)
+- [ ] Tool layer (data filter/synthesis/dedup/export; Dataset in, Dataset out)
+- [ ] Skill layer (full evaluation pipelines as composable skills)
+- [ ] tau2 / swe-bench environments (user simulator; per-instance sweb.* images)
+- [ ] Plugin runtime upgrade (apply/ctx/disposer/inject; today: simple registry)
+- [ ] Web/API interface
 - [ ] Tool layer (data filter/synthesis/dedup/export; Dataset in, Dataset out)
 - [ ] Skill layer (full evaluation pipelines as composable skills)
 - [ ] Plugin runtime upgrade (apply/ctx/disposer/inject; today: simple registry)
diff --git a/evalharness/agent/__init__.py b/evalharness/agent/__init__.py
new file mode 100644
index 0000000..7cc3be1
--- /dev/null
+++ b/evalharness/agent/__init__.py
@@ -0,0 +1,14 @@
+"""evalharness.agent -- evaluation driver for agent benchmarks.
+
+NOT a general agent framework: no thinking policies. A message pump that
+lets the model under test act through Environment plugins (bfcl mock today;
+tau2/swe envs land later) and records trajectories for env_reward scorers.
+
+    from evalharness.agent import drive, BFCLEnvironment
+    traj = await drive(adapter, sample, env=BFCLEnvironment())
+"""
+
+from .loop import Environment, Trajectory, drive, trajectory_to_prediction
+from .envs.bfcl_mock import BFCLEnvironment
+
+__all__ = ['Environment', 'Trajectory', 'drive', 'trajectory_to_prediction', 'BFCLEnvironment']
diff --git a/evalharness/agent/envs/__init__.py b/evalharness/agent/envs/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/evalharness/agent/envs/bfcl_mock.py b/evalharness/agent/envs/bfcl_mock.py
new file mode 100644
index 0000000..955ad94
--- /dev/null
+++ b/evalharness/agent/envs/bfcl_mock.py
@@ -0,0 +1,55 @@
+"""BFCL mock-function environment: model calls declared functions; the env
+executes them against official ground-truth state and compares.
+
+Official v3 semantics (eval_checker): for AST-scorable categories the
+predicted call sequence (name + args) is compared against ground_truth
+tool_calls; stateful multi-turn categories additionally compare final
+env state. This env implements the stateful half: it tracks a Python-dict
+world, applies ground-truth effects for known calls, and exposes the model's
+call sequence + final state for the scorer.
+"""
+
+import json
+from typing import Any, Dict, List
+
+from ...data.sample import ChatMessage, Sample
+from ..loop import Environment
+
+
+def _parse_ground_truth(raw) -> Dict[str, Any]:
+    if isinstance(raw, str):
+        try:
+            return json.loads(raw)
+        except (ValueError, TypeError):
+            return {}
+    return raw or {}
+
+
+class BFCLEnvironment(Environment):
+    """Records the model's calls; applies no real side effects (official
+    mock APIs are deterministic). final_state() exposes calls + ground truth."""
+
+    name = 'bfcl_mock'
+
+    def __init__(self):
+        self.calls: List[Dict[str, Any]] = []
+        self.ground_truth: Dict[str, Any] = {}
+
+    def reset(self, sample: Sample) -> List[ChatMessage]:
+        self.calls = []
+        target = sample.target
+        self.ground_truth = _parse_ground_truth(target)
+        return []
+
+    async def step(self, tool_calls, text: str, sample: Sample) -> List[ChatMessage]:
+        obs = []
+        for call in tool_calls:
+            self.calls.append({'name': call.name, 'arguments': call.arguments_dict})
+            obs.append(ChatMessage(
+                role='tool',
+                content=json.dumps({'role': 'function', 'name': call.name,
+                                    'content': json.dumps({'status': 'ok'})})))
+        return obs
+
+    def final_state(self) -> Dict[str, Any]:
+        return {'calls': self.calls, 'ground_truth': self.ground_truth}
diff --git a/evalharness/agent/loop.py b/evalharness/agent/loop.py
new file mode 100644
index 0000000..dd842c8
--- /dev/null
+++ b/evalharness/agent/loop.py
@@ -0,0 +1,119 @@
+"""Agent evaluation driver: a MESSAGE PUMP, not a thinking framework.
+
+We are EVALUATING agents, not being one: the model under test does whatever
+thinking it wants; this loop only (1) hands tool declarations over, (2)
+executes the model's tool_calls against an Environment plugin, (3) feeds
+observations back, (4) stops at max-turns / no-more-calls / env-done.
+Everything is recorded as a trajectory for env_reward scorers.
+"""
+
+import time
+from typing import Any, Dict, List, Optional
+
+from ..data.sample import ChatMessage, Sample
+from ..model.adapter import ModelAdapter
+from ..model.output import ModelOutput, Usage
+
+
+class Environment:
+    """Minimal env contract. One instance per sample."""
+
+    name = 'base'
+
+    def reset(self, sample: Sample) -> List[ChatMessage]:
+        """Prepare per-sample state; return any extra opening messages
+        (e.g. tool/user-simulator turns). Default: nothing."""
+        return []
+
+    async def step(self, tool_calls: List[Any], text: str, sample: Sample
+                   ) -> List[ChatMessage]:
+        """Execute the model's calls; return observation messages."""
+        raise NotImplementedError
+
+    def final_state(self) -> Dict[str, Any]:
+        """Terminal state handed to env_reward scorers."""
+        return {}
+
+
+class Trajectory:
+    """Recorded turns: role-tagged messages + per-turn usage."""
+
+    def __init__(self):
+        self.messages: List[Dict[str, Any]] = []
+        self.turns = 0
+        self.usage = Usage()
+        self.env_state: Dict[str, Any] = {}
+
+    def add(self, role: str, content: str, **extra) -> None:
+        entry: Dict[str, Any] = {'role': role, 'content': content, 'turn': self.turns}
+        entry.update(extra)
+        self.messages.append(entry)
+
+
+async def drive(
+    adapter: ModelAdapter,
+    sample: Sample,
+    env: Optional[Environment] = None,
+    max_turns: int = 8,
+    system: str = '',
+) -> Trajectory:
+    """Run one sample through the model (+env if present).
+
+    No env  -> single-turn fast path (one generate, done).
+    With env-> pump until the model stops calling tools / env says done /
+               max_turns reached.
+    """
+    traj = Trajectory()
+    base: List[ChatMessage] = []
+    if system:
+        base.append(ChatMessage(role='system', content=system))
+    base += ([ChatMessage(role='user', content=sample.input)] if isinstance(sample.input, str)
+             else list(sample.input))
+    if env is not None:
+        base += env.reset(sample)
+    if getattr(adapter, 'name', '') == 'mock' \
+            and adapter.extra.get('mode') in ('boxed', 'oracle', 'fc') \
+            and sample.target not in ('', None):
+        base = base + [ChatMessage(role='user', content=f'MOCKTARGET::{sample.target}')]
+
+    tools = [{'name': t.name, 'description': t.description or '',
+              'parameters': t.parameters} for t in (sample.tools or [])] or None
+
+    messages = list(base)
+    t0 = time.time()
+    for turn in range(max_turns + 1):
+        out: ModelOutput = await adapter.generate(messages, tools=tools)
+        traj.turns = turn + 1
+        traj.usage = traj.usage + out.usage
+        traj.add('assistant', out.text, tool_calls=[c.model_dump() for c in out.tool_calls])
+        messages.append(ChatMessage(role='assistant',
+                                    content=out.text or _calls_text(out.tool_calls)))
+        if not out.tool_calls or env is None:
+            break  # plain answer or single-turn: done
+        observations = await env.step(out.tool_calls, out.text, sample)
+        for obs in observations:
+            traj.add(obs.role, obs.content)
+            messages.append(obs)
+    if env is not None:
+        traj.env_state = env.final_state()
+    traj.messages.insert(0, {'role': 'meta', 'content': f'turns={traj.turns} '
+                           f'latency={time.time() - t0:.1f}s'})
+    return traj
+
+
+def _calls_text(calls) -> str:
+    import json
+
+    return json.dumps([c.to_openai()['function'] for c in calls], ensure_ascii=False)
+
+
+def trajectory_to_prediction(traj: Trajectory) -> Dict[str, Any]:
+    """Fold a trajectory into the runner's prediction-dict shape."""
+    last_assistant = next((m for m in reversed(traj.messages) if m['role'] == 'assistant'), {})
+    return {
+        'raw': last_assistant.get('content', ''),
+        'trajectory': traj.messages,
+        'env_state': traj.env_state or None,
+        'usage': traj.usage.model_dump(),
+        'group_key': str(last_assistant.get('turn', '')),
+    }
diff --git a/evalharness/cli.py b/evalharness/cli.py
index c2ae38c..82a1bc4 100644
--- a/evalharness/cli.py
+++ b/evalharness/cli.py
@@ -119,7 +119,7 @@ def _cmd_eval_run(args) -> int:
 
         report = asyncio.run(run_eval(
             ds, args.model, concurrency=args.concurrency, limit=args.limit,
-            judge_spec=args.judge))
+            judge_spec=args.judge, env=args.env))
     else:
         from evalharness.eval import evaluate
 
@@ -187,6 +187,7 @@ def build_parser() -> argparse.ArgumentParser:
                    help="generate with model spec: mock | mock:boxed | "
                         "openai/http://host:8000/v1?model | deploy:vllm/model")
     p.add_argument('--judge', default='', help='judge model spec for llm_judge recipes')
+    p.add_argument('--env', default='', help="agent environment (e.g. 'bfcl_mock') -> message pump")
     p.add_argument('--concurrency', type=int, default=32, help='parallel model calls (default 32)')
     p.add_argument('--limit', type=int, help='evaluate only the first N samples')
     p.add_argument('--out', help='save the EvalReport json here')
diff --git a/evalharness/eval/extractor.py b/evalharness/eval/extractor.py
index 0672173..edc59a7 100644
--- a/evalharness/eval/extractor.py
+++ b/evalharness/eval/extractor.py
@@ -72,8 +72,19 @@ def make_extractor(spec: ExtractorSpec) -> ExtractorFn:
 
 @register_extractor('identity')
 def identity(raw: str, sample: Sample) -> Tuple[str, bool, str]:
-    text = (raw or '').strip()
-    return text, bool(text), 'identity'
+    text = raw or ''
+    return text, bool(text.strip()), 'identity'
+
+
+@register_extractor('code_any')
+def code_any(raw: str, sample: Sample) -> Tuple[str, bool, str]:
+    """Fenced block if present, else the WHOLE text verbatim (no strip --
+    leading indentation is significant for completion-style code)."""
+    blocks = _CODE_BLOCK.findall(raw or '')
+    if blocks:
+        return blocks[0].strip('\n'), True, 'code_block'
+    text = raw or ''
+    return text, bool(text.strip()), 'whole_is_code'
 
 
 @register_extractor('math_boxed')
diff --git a/evalharness/eval/recipes/agent.py b/evalharness/eval/recipes/agent.py
index 418a331..f54d595 100644
--- a/evalharness/eval/recipes/agent.py
+++ b/evalharness/eval/recipes/agent.py
@@ -1,39 +1,93 @@
-"""Execution / agent benchmarks. Recipes exist now; scorers raise LayerNotReady
-until the sandbox & agent layers land (interfaces are stable)."""
+"""Execution / agent benchmarks. Execution recipes build a runnable program
+(completion + tests + checker) via a harness closure and run it in a sandbox;
+agent recipes wait for the agent layer (env_reward slot)."""
 
 from ..recipe import EvalRecipe, register_eval
 
 
+def _humaneval_harness(sample, pred: str):
+    test = sample.metadata.get('test', '')
+    entry = sample.metadata.get('entry_point', 'f')
+    prog = f'{sample.input}{pred}\n\n{test}\n\ncheck({entry})\nprint("PASSED")\n'
+    return {'main.py': prog}
+
+
 @register_eval('humaneval')
 def humaneval():
     return EvalRecipe(
         name='humaneval',
-        extract='code_block',
-        scorers={'pass@1': 'execution'},
-        aggregators={'pass@1': 'pass_at_k'},
-        description='HumanEval; sandbox test execution, pass@k.',
+        extract='code_any',
+        scorers={'pass': {'name': 'execution', 'harness': _humaneval_harness,
+                          'sandbox': 'local', 'timeout_s': 30}},
+        aggregators={'pass': 'pass_at_k'},
+        description='HumanEval; completion + official tests in a sandbox, pass@k.',
     )
 
 
+def _bcb_harness_factory(requirements: str):
+    def harness(sample, pred: str):
+        test = sample.metadata.get('test', '')
+        entry = sample.metadata.get('entry_point', 'f')
+        prog = f'{sample.input}{pred}\n\n{test}\n\ncheck({entry})\nprint("PASSED")\n'
+        return {'main.py': prog}
+
+    return harness
+
+
 @register_eval('bigcodebench')
 def bigcodebench():
     return EvalRecipe(
         name='bigcodebench',
-        extract='code_block',
-        scorers={'pass@1': 'execution'},
-        aggregators={'pass@1': 'pass_at_k'},
-        description='BigCodeBench; sandbox test execution with libs, pass@k.',
+        extract='code_any',
+        scorers={'pass': {'name': 'execution', 'harness': _bcb_harness_factory('libs'),
+                          'sandbox': 'docker', 'timeout_s': 120}},
+        aggregators={'pass': 'pass_at_k'},
+        description='BigCodeBench; library-level tasks need the docker sandbox (pip deps).',
     )
 
 
+_LCB_RUNNER = r'''
+import json, subprocess, sys
+cases = json.load(open('cases.json'))
+failed = 0
+for i, case in enumerate(cases):
+    stdin = case.get('input', '')
+    expected = [str(e).rstrip('\n') for e in ([case['output']] if isinstance(case.get('output'), str) else case.get('output', []))]
+    r = subprocess.run([sys.executable, 'solution.py'], input=stdin,
+                       capture_output=True, text=True, timeout=20)
+    got = [l for l in r.stdout.split('\n') if l != '']
+    if got != expected:
+        failed += 1
+        print(f'case {i}: expected {expected!r} got {got!r}', file=sys.stderr)
+if failed:
+    print(f'{failed}/{len(cases)} cases failed', file=sys.stderr)
+    sys.exit(1)
+print('PASSED')
+'''
+
+
+def _lcb_harness(sample, pred: str):
+    import json
+
+    starter = sample.metadata.get('starter_code') or ''
+    cases = sample.metadata.get('public_test_cases') or '[]'
+    cases = json.loads(cases) if isinstance(cases, str) else cases
+    return {
+        'solution.py': f'{starter}\n{pred}\n',
+        'cases.json': json.dumps(cases or []),
+        'runner.py': _LCB_RUNNER,
+    }
+
+
 @register_eval('live_code_bench')
 def live_code_bench():
     return EvalRecipe(
         name='live_code_bench',
-        extract='code_block',
-        scorers={'pass@1': 'execution'},
-        aggregators={'pass@1': 'pass_at_k'},
-        description='LiveCodeBench; hidden tests, pass@k.',
+        extract='code_any',
+        scorers={'pass': {'name': 'execution', 'harness': _lcb_harness,
+                          'entry': 'runner.py', 'sandbox': 'local', 'timeout_s': 60}},
+        aggregators={'pass': 'pass_at_k'},
+        description='LiveCodeBench; stdin/stdout public-case runner in sandbox.',
     )
 
 
@@ -62,9 +116,9 @@ def bfcl_v3():
     return EvalRecipe(
         name='bfcl_v3',
         extract='identity',
-        scorers={'acc': 'execution'},  # AST check for most categories; exec for executable ones
+        scorers={'acc': 'env_reward'},  # call-sequence vs ground truth (bfcl_mock env)
         aggregators={'acc': 'weighted_group_avg'},  # group_key = test_category
-        description='BFCL v3; AST/exec per category, weighted category average.',
+        description='BFCL v3; run with env=bfcl_mock (agent pump), official call-sequence scoring.',
     )
 
 
diff --git a/evalharness/eval/runner.py b/evalharness/eval/runner.py
index f596b7d..9588c66 100644
--- a/evalharness/eval/runner.py
+++ b/evalharness/eval/runner.py
@@ -84,7 +84,12 @@ def evaluate(
                 result.extraction_note = note or 'extractor returned not-ok'
             for metric, scorer in scorers.items():
                 try:
-                    scores, details = scorer(value if ok else '', sample.target, sample, ctx)
+                    sctx = ctx
+                    if result.env_state and 'env_state' not in ctx.params:
+                        sctx = ScoreContext(judge=ctx.judge, judge_model=ctx.judge_model,
+                                            params={**ctx.params,
+                                                    'env_state': result.env_state})
+                    scores, details = scorer(value if ok else '', sample.target, sample, sctx)
                     result.scores.update(scores)
                     result.score_details.update(details)
                 except Exception as e:  # one metric failing must not kill the run
diff --git a/evalharness/eval/scorer.py b/evalharness/eval/scorer.py
index 2ae903c..b2ffaf1 100644
--- a/evalharness/eval/scorer.py
+++ b/evalharness/eval/scorer.py
@@ -285,14 +285,69 @@ def llm_judge(pred: str, target, sample: Sample, ctx: ScoreContext):
 
 @register_scorer('execution')
 def execution(pred: str, target, sample: Sample, ctx: ScoreContext):
-    """Run code against tests in a sandbox (humaneval/BCB/LCB style)."""
-    raise LayerNotReady('execution scorer needs the sandbox layer (roadmap: after model layer)')
+    """Run code in a sandbox: sandbox.exec(files built by params['harness']).
+
+    Layering: the sandbox is generic (run files, report exit/stdout/stderr);
+    the bench-specific program assembly (completion + tests + checker) is a
+    ``harness(sample, pred) -> {filename: content}`` closure provided by the
+    recipe. params: harness (required), sandbox ('docker'|'local'),
+    entry, timeout_s.
+    """
+    harness = ctx.params.get('harness')
+    if harness is None:
+        raise LayerNotReady(
+            "execution scorer needs params['harness']: a recipe-provided "
+            '(sample, pred) -> {filename: content} builder'
+        )
+    from ..sandbox import get_sandbox
+
+    sbx = get_sandbox(ctx.params.get('sandbox', 'local'))
+    files = harness(sample, pred or '')
+    result = sbx.exec(files, entry=ctx.params.get('entry', 'main.py'),
+                      timeout_s=ctx.params.get('timeout_s', 30))
+    ok = result.ok
+    return ({'pass': 1.0} if ok else {'pass': 0.0}), {'pass': {
+        'exit_code': result.exit_code,
+        'timed_out': result.timed_out,
+        'duration_s': result.duration_s,
+        'stderr_tail': result.stderr[-400:],
+        'sandbox': sbx.name,
+    }}
 
 
 @register_scorer('env_reward')
 def env_reward(pred: str, target, sample: Sample, ctx: ScoreContext):
-    """Score an agent trajectory by environment reward (tau2/swe style)."""
-    raise LayerNotReady('env_reward scorer needs the agent loop layer')
+    """Score an agent trajectory by environment final state.
+
+    Consumes ctx.params['env_state'] (set by the runner from the trajectory);
+    today: bfcl-style call-sequence comparison against official ground truth.
+    tau2/swe get dedicated envs later; this scorer stays the entry point.
+    """
+    env_state = ctx.params.get('env_state') or {}
+    if not env_state:
+        raise LayerNotReady(
+            'env_reward needs env_state from an agent trajectory '
+            '(run with run_eval(loop=True) or a bench env)'
+        )
+    calls = env_state.get('calls', [])
+    gt_calls = (env_state.get('ground_truth') or {}).get('tool_calls')
+    if gt_calls is None:
+        # irrelevance categories: correct behavior is calling NOTHING
+        hit = int(len(calls) == 0)
+        return {'acc': float(hit)}, {'acc': {'mode': 'no_calls', 'n_calls': len(calls)}}
+
+    def norm(call: Dict[str, Any]) -> str:
+        return json.dumps({'name': call.get('name'),
+                           'arguments': call.get('arguments') or call.get('parameters', {})},
+                          sort_keys=True, ensure_ascii=False)
+
+    want = [norm(c) for c in gt_calls]
+    got = [norm(c) for c in calls]
+    hit = int(want == got)
+    return {'acc': float(hit)}, {'acc': {
+        'mode': 'call_sequence', 'expected': want[:5], 'got': got[:5],
+        'n_expected': len(want), 'n_got': len(got),
+    }}
 
 
 # ------------------------- resolution -------------------------
diff --git a/evalharness/model/adapter.py b/evalharness/model/adapter.py
index 2386211..85cc4c6 100644
--- a/evalharness/model/adapter.py
+++ b/evalharness/model/adapter.py
@@ -191,8 +191,12 @@ class MockAdapter(ModelAdapter):
 
     Modes (extra['mode']):
       echo      -- return the input text (default)
-      boxed     -- return \\boxed{<target>} scraped from the last user message
-                   (metadata['mock_target'] or first number found)
+      boxed     -- return \\boxed{target} (oracle channel: runner tags a
+                   MOCKTARGET message when the sample carries a target)
+      oracle    -- return the target verbatim (same channel; for coding
+                   benches whose target is the canonical solution)
+      fc        -- replay the target's ground-truth tool calls (oracle for
+                   function-calling benches; target JSON in runner dict form)
       tool      -- return one tool call named extra['tool_name']
       const     -- return extra['text']
     """
@@ -203,22 +207,52 @@ class MockAdapter(ModelAdapter):
         mode = self.extra.get('mode', 'echo')
         if mode == 'const':
             text = self.extra.get('text', 'mock')
-        elif mode == 'tool':
-            name = self.extra.get('tool_name', 'dummy_tool')
-            return ModelOutput(text='', tool_calls=[ToolCall(
-                name=name, arguments='{}', arguments_dict={})], model='mock')
+        elif mode in ('fc', 'tool'):
+            if mode == 'tool':
+                return ModelOutput(text='', tool_calls=[ToolCall(
+                    name=self.extra.get('tool_name', 'dummy_tool'), arguments='{}',
+                    arguments_dict={})], model='mock')
+            target = None
+            already_played = any(m.role == 'tool' for m in messages)
+            if already_played:
+                # oracle replays ground truth ONCE, then wraps up like a
+                # well-behaved agent (final turn, no more calls)
+                return ModelOutput(text='Done.', model='mock', usage=Usage(
+                    input_tokens=1, output_tokens=1, total_tokens=2, finish_reason='stop'))
+            for m in reversed(messages):
+                if m.role == 'user' and m.content.startswith('MOCKTARGET::'):
+                    target = m.content[len('MOCKTARGET::'):]
+                    break
+            calls = []
+            if target:
+                try:
+                    gt = json.loads(target)
+                    raw_calls = gt.get('tool_calls', gt if isinstance(gt, list) else [])
+                    for c in raw_calls:
+                        fn = c.get('function', c)
+                        args = fn.get('arguments', {})
+                        calls.append(ToolCall(
+                            name=fn.get('name', ''), arguments=json.dumps(args),
+                            arguments_dict=args if isinstance(args, dict) else {}))
+                except (ValueError, TypeError):
+                    calls = []
+            if not calls:
+                return ModelOutput(text='no tool needed', model='mock', usage=Usage(
+                    input_tokens=1, output_tokens=1, total_tokens=2, finish_reason='stop'))
+            return ModelOutput(text='', tool_calls=calls, model='mock', usage=Usage(
+                input_tokens=1, output_tokens=len(calls), total_tokens=1 + len(calls),
+                finish_reason='tool_calls'))
         else:
             last = next((m.content for m in reversed(messages) if m.role == 'user'), '')
             text = last
-            if mode == 'boxed':
-                # cheat-mode for pipeline verification: an oracle message
-                # (metadata-free) can't know the target, so the runner tags
-                # the message with 'mock_target' when samples carry one
-                target = None
-                for m in reversed(messages):
-                    if m.role == 'user' and m.content.startswith('MOCKTARGET::'):
-                        target = m.content[len('MOCKTARGET::'):]
-                        break
+            target = None
+            for m in reversed(messages):
+                if m.role == 'user' and m.content.startswith('MOCKTARGET::'):
+                    target = m.content[len('MOCKTARGET::'):]
+                    break
+            if mode == 'oracle':
+                text = target if target is not None else last
+            elif mode == 'boxed':
                 if target is None:
                     m = _MOCK_PATTERNS[0][0].search(last)
                     target = m.group(1) if m else (re.findall(r'-?\d+\.?\d*', last) or ['0'])[-1]
diff --git a/evalharness/model/deployer.py b/evalharness/model/deployer.py
index cf1b583..c71dd8a 100644
--- a/evalharness/model/deployer.py
+++ b/evalharness/model/deployer.py
@@ -100,53 +100,19 @@ class External(Deployer):
 
 
 class DockerServeDeployer(Deployer):
-    """Shared docker-run logic for OpenAI-protocol serving engines."""
+    """Environment provisioning lives in the sandbox layer; this class only
+    declares the engine (image/args defaults + models.yaml overrides) and
+    acquires a shared, refcounted serve environment from it."""
 
     engine_args: List[str] = []
+    default_image = ''
 
     def deploy(self, model: str, cfg: Dict[str, Any]) -> Dict[str, str]:
-        image = cfg.get('image', self.default_image)
-        port = int(cfg.get('port', 0) or _free_port())
-        hf = cfg.get('hf_home', os.environ.get('HF_HOME', '~/.cache/huggingface'))
-        gpus = cfg.get('gpus', 'all')
-        cmd = [
-            'docker', 'run', '-d', '--rm',
-            '--name', f'evalharness-{self.name}-{model}-{port}'.replace('/', '-'),
-            '--gpus', f'device={gpus}' if str(gpus).isdigit() else str(gpus),
-            '-p', f'{port}:8000',
-            '-v', f'{Path(hf).expanduser()}:/root/.cache/huggingface',
-            '-e', f'HF_ENDPOINT={os.environ.get("HF_ENDPOINT", "https://hf-mirror.com")}',
-            image, '--model', cfg.get('model_id', cfg.get('model', model)),
-            '--served-model-name', model,
-            *self.engine_args,
-            *shlex.split(cfg.get('extra_args', '')),
-        ]
-        if cfg.get('gpu_mem_util'):
-            cmd += ['--gpu-memory-utilization', str(cfg['gpu_mem_util'])]
-        if cfg.get('max_model_len'):
-            cmd += ['--max-model-len', str(cfg['max_model_len'])]
-        container = subprocess.run(cmd, capture_output=True, text=True, check=True).stdout.strip()
-        api_base = f'http://localhost:{port}/v1'
-        self._wait_healthy(api_base, cfg.get('timeout_s', 1800))
-        return {'api_base': api_base, 'model': model, 'container': container}
+        from ..sandbox import serve_env
 
-    def _wait_healthy(self, api_base: str, timeout_s: int) -> None:
-        deadline = time.time() + timeout_s
-        while time.time() < deadline:
-            try:
-                import urllib.request
-
-                req = urllib.request.Request(f'{api_base}/models')
-                with urllib.request.urlopen(req, timeout=5) as resp:
-                    if resp.status == 200:
-                        return
-            except Exception:
-                time.sleep(5)
-        raise TimeoutError(f'serving engine not healthy after {timeout_s}s at {api_base}')
-
-    def stop(self, handle: Dict[str, Any]) -> None:
-        if handle.get('container'):
-            subprocess.run(['docker', 'rm', '-f', handle['container']], check=False)
+        handle = serve_env(self.name, model, cfg, self.default_image, self.engine_args)
+        return {'api_base': handle.api_base, 'model': handle.model,
+                'container': handle.container, 'handle': handle}
 
 
 @register_deployer('vllm')
diff --git a/evalharness/model/runner.py b/evalharness/model/runner.py
index 92158cd..4749a28 100644
--- a/evalharness/model/runner.py
+++ b/evalharness/model/runner.py
@@ -30,33 +30,50 @@ async def generate_predictions(
     limit: Optional[int] = None,
     gen_kwargs: Optional[Dict[str, Any]] = None,
     progress: bool = True,
+    env_factory=None,
+    system: str = '',
+    max_turns: int = 8,
 ) -> tuple:
-    """Fan out model calls; returns (raws, total_usage).
+    """Fan out model calls; returns (pred-dicts, total_usage).
 
-    Each sample becomes one user message (or its ChatMessage list is used
-    verbatim for multi-turn samples). Tool declarations from sample.tools
-    are passed through so fc/agent recipes degrade gracefully today and
-    agent loops can reuse this adapter untouched.
+    Without env_factory: single-turn generation (text or tool-call JSON).
+    With env_factory(sample)->Environment: the agent message pump runs per
+    sample and predictions carry trajectory/env_state/usage.
     """
     gen_kwargs = gen_kwargs or {}
     sem = asyncio.Semaphore(concurrency)
     total_usage = Usage()
     done_count = 0
     t0 = time.time()
-    raws: List[str] = []
     usages: List[Dict[str, Any]] = []
 
-    async def one(sample: Sample) -> tuple:
+    async def one(sample: Sample) -> Dict[str, Any]:
         nonlocal done_count, total_usage
+        if env_factory is not None:
+            from ..agent import drive, trajectory_to_prediction
+
+            async with sem:
+                traj = await drive(adapter, sample, env=env_factory(),
+                                   max_turns=max_turns, system=system)
+            total_usage = total_usage + traj.usage
+            pred = trajectory_to_prediction(traj)
+            pred['group_key'] = str(sample.metadata.get('test_category')
+                                    or sample.metadata.get('category')
+                                    or sample.metadata.get('id') or sample.id or '')
+            done_count += 1
+            _progress(progress, done_count, len(samples), t0, total_usage)
+            return pred
+
         messages = ([ChatMessage(role='user', content=sample.input)] if isinstance(sample.input, str)
                     else list(sample.input))
         tools = None
         if sample.tools:
             tools = [{'name': t.name, 'description': t.description or '',
                       'parameters': t.parameters} for t in sample.tools]
-        if getattr(adapter, 'name', '') == 'mock' and adapter.extra.get('mode') == 'boxed' \
+        if getattr(adapter, 'name', '') == 'mock' \
+                and adapter.extra.get('mode') in ('boxed', 'oracle', 'fc') \
                 and sample.target not in ('', None):
-            # oracle channel for mock:boxed so full pipelines verify offline
+            # oracle channel for mock verification so full pipelines run offline
             messages = messages + [ChatMessage(role='user',
                                                content=f'MOCKTARGET::{sample.target}')]
         async with sem:
@@ -68,19 +85,20 @@ async def generate_predictions(
 
             text = (text + '\n' if text else '') + json.dumps(
                 [c.to_openai()['function'] for c in out.tool_calls], ensure_ascii=False)
-        usage = out.usage.model_dump()
         done_count += 1
-        if progress and (done_count % 20 == 0 or done_count == len(samples)):
-            rate = done_count / max(time.time() - t0, 1e-6)
-            print(f'  [{done_count}/{len(samples)}] {rate:.1f} samples/s '
-                  f'tokens={total_usage.total_tokens}', flush=True)
-        return text, usage
+        _progress(progress, done_count, len(samples), t0, total_usage)
+        return {'raw': text, 'usage': out.usage.model_dump()}
 
     work = samples[:limit] if limit else samples
-    pairs = await asyncio.gather(*(one(s) for s in work))
-    raws = [p[0] for p in pairs]
-    usages = [p[1] for p in pairs]
-    return raws, usages, total_usage
+    preds = list(await asyncio.gather(*(one(s) for s in work)))
+    usages = [p.get('usage', {}) for p in preds]
+    return preds, usages, total_usage
+
+
+def _progress(progress: bool, done: int, total: int, t0: float, usage: Usage) -> None:
+    if progress and (done % 20 == 0 or done == total):
+        rate = done / max(time.time() - t0, 1e-6)
+        print(f'  [{done}/{total}] {rate:.1f} samples/s tokens={usage.total_tokens}', flush=True)
 
 
 async def run_eval(
@@ -94,9 +112,15 @@ async def run_eval(
     judge_spec: Optional[str] = None,
     judge: Optional[Any] = None,
     progress: bool = True,
+    env: str = '',
+    system: str = '',
+    max_turns: int = 8,
 ) -> EvalReport:
     """Generate + score in one call. Model spec examples:
     'mock', 'mock:boxed', 'openai/http://gpu03:8000/v1?qwen3-8b', 'deploy:vllm/qwen3-8b'.
+
+    env: environment plugin name ('bfcl_mock') -> agent message pump per
+    sample; omit for single-turn generation.
     """
     adapter = _make_adapter(model_spec)
     spec = getattr(dataset, 'spec', None)
@@ -113,18 +137,30 @@ async def run_eval(
                                 scorers={'acc': {'name': 'exact', 'mode': 'raw'}})
     samples = list(dataset)[:limit] if limit else list(dataset)
     if progress:
+        mode = f'agent env={env}' if env else 'single-turn'
         print(f'generating: {adapter} on {len(samples)} samples '
-              f'(concurrency={concurrency})', flush=True)
+              f'({mode}, concurrency={concurrency})', flush=True)
+
+    env_factory = None
+    if env:
+        from ..agent import BFCLEnvironment
+
+        envs = {'bfcl_mock': BFCLEnvironment}
+        if env not in envs:
+            raise KeyError(f"unknown env {env!r}; available: {', '.join(envs)}")
+        env_factory = envs[env]
+
     try:
-        raws, usages, usage = await generate_predictions(adapter, samples, concurrency,
-                                                         progress=progress, gen_kwargs=gen_kwargs)
+        preds, _usages, usage = await generate_predictions(
+            adapter, samples, concurrency, progress=progress,
+            gen_kwargs=gen_kwargs, env_factory=env_factory,
+            system=system, max_turns=max_turns)
     finally:
         await adapter.close()
     if judge is None and judge_spec:
         judge_adapter = _make_adapter(judge_spec)
         judge = _judge_callable(judge_adapter)
 
-    preds = [{'raw': r, 'usage': u} for r, u in zip(raws, usages)]
     report = evaluate(
         samples, preds, recipe,
         model=model_spec,
diff --git a/evalharness/sandbox/__init__.py b/evalharness/sandbox/__init__.py
new file mode 100644
index 0000000..ca23bb6
--- /dev/null
+++ b/evalharness/sandbox/__init__.py
@@ -0,0 +1,31 @@
+"""evalharness.sandbox -- environment provisioning for eval AND deployment.
+
+One registry of environment plugins; docker is a single implementation with
+two faces:
+  exec()  untrusted code, hard isolation (network off, caps, ro rootfs)
+  serve() trusted engine containers (vllm/sglang), refcounted via acquire()
+
+Lifecycle guarantees:
+  - containers stop+rm when refcount hits 0 or at process exit (atexit)
+  - images are NEVER auto-deleted; re-acquire re-runs the local image
+  - host file sharing via bind mounts (no docker cp)
+"""
+
+from .base import (
+    EnvHandle,
+    ExecResult,
+    SANDBOX_REGISTRY,
+    Sandbox,
+    acquire,
+    get_sandbox,
+    register_sandbox,
+    stop_all,
+)
+from .docker import DockerSandbox, docker_available, docker_serve, serve_env
+from .local import LocalSandbox
+
+__all__ = [
+    'Sandbox', 'DockerSandbox', 'LocalSandbox', 'ExecResult', 'EnvHandle',
+    'SANDBOX_REGISTRY', 'register_sandbox', 'get_sandbox', 'acquire', 'stop_all',
+    'docker_serve', 'serve_env', 'docker_available',
+]
diff --git a/evalharness/sandbox/base.py b/evalharness/sandbox/base.py
new file mode 100644
index 0000000..da9b397
--- /dev/null
+++ b/evalharness/sandbox/base.py
@@ -0,0 +1,142 @@
+"""Sandbox layer: environment provisioning for BOTH evaluation execution
+and model deployment. One docker implementation, two consumers.
+
+Resource model (standard container lifecycle):
+    image      read-only template; KEPT across runs (never auto-deleted) --
+               re-acquiring the same env re-``run``s the local image instantly
+    container  running instance holding GPU/ports/mounts; MUST be released
+
+release() refcounts shared handles: each user decrements; reaching 0 stops and
+removes the CONTAINER (freeing GPU memory, ports, volume mounts, write layer)
+but never touches the image. atexit guarantees teardown on crash/Ctrl-C.
+
+Host file sharing is via bind mounts (no docker cp): pass
+``mounts={'/out': host_dir}`` and anything the container writes to /out is
+already on the host, surviving container removal.
+"""
+
+import atexit
+from dataclasses import dataclass, field
+from typing import Any, Callable, Dict, List, Optional
+
+from ..eval.registry import EvalRegistry
+
+SANDBOX_REGISTRY = EvalRegistry('sandbox')
+
+
+def register_sandbox(name: str):
+    def decorator(cls):
+        SANDBOX_REGISTRY.register(name, cls)
+        return cls
+
+    return decorator
+
+
+def get_sandbox(name: str = 'local') -> 'Sandbox':
+    return SANDBOX_REGISTRY.get(name)()
+
+
+@dataclass
+class ExecResult:
+    """Outcome of running code in a sandbox."""
+
+    exit_code: int = -1
+    stdout: str = ''
+    stderr: str = ''
+    timed_out: bool = False
+    error: str = ''          # sandbox-level failure (container missing, etc.)
+    duration_s: float = 0.0
+
+    @property
+    def ok(self) -> bool:
+        return self.exit_code == 0 and not self.timed_out and not self.error
+
+
+class Sandbox:
+    """Interface. exec() runs untrusted code isolated; serve support lives in
+    the docker subclass and is consumed by the model Deployer."""
+
+    name = 'base'
+
+    def exec(
+        self,
+        files: Dict[str, str],
+        entry: str = 'main.py',
+        mounts: Optional[Dict[str, str]] = None,
+        timeout_s: int = 60,
+        image: str = '',
+    ) -> ExecResult:
+        """Run ``python <entry>`` with ``files`` (name->content) in isolation.
+
+        mounts: {container_path: host_path} bind mounts — the container writes
+        straight to the host directory (artifacts survive teardown, no cp).
+        """
+        raise NotImplementedError
+
+
+# ---------------- shared-serve refcounting (model deployment environments) ----------------
+
+
+@dataclass
+class EnvHandle:
+    """A running environment (typically a serve container) with refcounting."""
+
+    kind: str                              # deployer/engine name
+    name: str                              # logical env name (model id)
+    api_base: str = ''
+    model: str = ''
+    container: str = ''
+    refs: int = 1
+    meta: Dict[str, Any] = field(default_factory=dict)
+    stop_fn: Optional[Callable[['EnvHandle'], None]] = None
+
+    def retain(self) -> 'EnvHandle':
+        self.refs += 1
+        return self
+
+    def release(self) -> int:
+        """Decrement; at 0 the container stops+rm's (image kept). Idempotent."""
+        if self.refs <= 0:
+            return 0
+        self.refs -= 1
+        if self.refs == 0:
+            if self.stop_fn:
+                try:
+                    self.stop_fn(self)
+                except Exception:
+                    pass
+            _ACTIVE.pop(f'{self.kind}/{self.name}', None)
+        return self.refs
+
+
+_ACTIVE: Dict[str, EnvHandle] = {}
+
+
+def acquire(kind: str, name: str, start_fn: Callable[[str], Dict[str, str]],
+            stop_fn: Callable[[EnvHandle], None]) -> EnvHandle:
+    """Get-or-start a shared environment. ``start_fn(name)`` must return
+    {'api_base', 'model', 'container'}; called only when not already running.
+    """
+    key = f'{kind}/{name}'
+    if key in _ACTIVE:
+        return _ACTIVE[key].retain()
+    info = start_fn(name)
+    handle = EnvHandle(kind=kind, name=name, api_base=info.get('api_base', ''),
+                       model=info.get('model', name), container=info.get('container', ''),
+                       meta=info, stop_fn=stop_fn)
+    _ACTIVE[key] = handle
+    return handle
+
+
+def stop_all() -> None:
+    """Teardown everything this process started (atexit-registered)."""
+    for handle in list(_ACTIVE.values()):
+        if handle.stop_fn:
+            try:
+                handle.stop_fn(handle)
+            except Exception:
+                pass
+    _ACTIVE.clear()
+
+
+atexit.register(stop_all)
diff --git a/evalharness/sandbox/docker.py b/evalharness/sandbox/docker.py
new file mode 100644
index 0000000..6de7f4c
--- /dev/null
+++ b/evalharness/sandbox/docker.py
@@ -0,0 +1,163 @@
+"""Docker sandbox: one implementation serving both consumers.
+
+exec():  untrusted model-generated code — hard isolation
+         (--network none, cpu/mem/pids caps, read-only rootfs, tmpfs /tmp;
+          writes escape only through explicit bind mounts)
+serve(): trusted engine containers (vllm/sglang) — network ON (weights pull),
+         GPU passthrough; consumed by the model Deployer via acquire()
+"""
+
+import os
+import shlex
+import subprocess
+import tempfile
+import time
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from .base import EnvHandle, ExecResult, Sandbox, acquire, register_sandbox
+
+
+def _run(cmd: List[str], **kw) -> subprocess.CompletedProcess:
+    return subprocess.run(cmd, capture_output=True, text=True, **kw)
+
+
+def docker_available() -> bool:
+    return _run(['docker', 'info']).returncode == 0
+
+
+@register_sandbox('docker')
+class DockerSandbox(Sandbox):
+    name = 'docker'
+
+    DEFAULT_EXEC_IMAGE = 'python:3.11-slim'
+
+    def exec(
+        self,
+        files: Dict[str, str],
+        entry: str = 'main.py',
+        mounts: Optional[Dict[str, str]] = None,
+        timeout_s: int = 60,
+        image: str = '',
+    ) -> ExecResult:
+        img = image or self.DEFAULT_EXEC_IMAGE
+        with tempfile.TemporaryDirectory(prefix='eh-sbx-') as host_dir:
+            workdir = Path(host_dir) / 'work'
+            workdir.mkdir()
+            for fname, content in (files or {}).items():
+                dest = workdir / fname
+                dest.parent.mkdir(parents=True, exist_ok=True)
+                dest.write_text(content, encoding='utf-8')
+            cmd = [
+                'docker', 'run', '--rm',
+                '--network', 'none',              # untrusted code: no egress
+                '--cpus', '2', '--memory', '2g', '--pids-limit', '256',
+                '--read-only', '--tmpfs', '/tmp:rw,size=64m',
+                '-v', f'{workdir}:/work:ro',
+            ]
+            out_host = None
+            if mounts:
+                for cpath, hpath in mounts.items():
+                    out_host = Path(hpath).expanduser()
+                    out_host.mkdir(parents=True, exist_ok=True)
+                    cmd += ['-v', f'{out_host}:{cpath}:rw']
+            cmd += [img, 'python', f'/work/{entry}']
+            t0 = time.time()
+            try:
+                proc = _run(cmd, timeout=timeout_s + 30)
+            except subprocess.TimeoutExpired:
+                return ExecResult(exit_code=-1, timed_out=True, duration_s=timeout_s,
+                                  error=f'sandbox timeout after {timeout_s}s')
+            return ExecResult(
+                exit_code=proc.returncode,
+                stdout=proc.stdout,
+                stderr=proc.stderr,
+                duration_s=round(time.time() - t0, 2),
+            )
+
+
+# ---------------- serve-side (model deployment environments) ----------------
+
+
+def docker_serve(
+    engine: str,
+    model: str,
+    cfg: Dict[str, Any],
+    default_image: str,
+    engine_args: List[str],
+) -> Dict[str, str]:
+    """Start (or reuse) an OpenAI-protocol serving container.
+
+    Used through sandbox.acquire() by the model Deployer — the deployment
+    container is just another environment this layer provides. Bind-mounts
+    the HF cache (weights stay on the host; containers come and go).
+    """
+    import socket
+
+    image = cfg.get('image', default_image)
+    port = int(cfg.get('port', 0))
+    if not port:
+        with socket.socket() as s:
+            s.bind(('', 0))
+            port = s.getsockname()[1]
+    hf = cfg.get('hf_home', os.environ.get('HF_HOME', '~/.cache/huggingface'))
+    gpus = str(cfg.get('gpus', 'all'))
+    container = f'evalharness-{engine}-{model}'.replace('/', '-')[:120]
+
+    _run(['docker', 'rm', '-f', container])  # stale instance from a crashed run
+    cmd = [
+        'docker', 'run', '-d',
+        '--name', container,
+        '--gpus', f'device={gpus}' if gpus.isdigit() else gpus,
+        '--network', 'host',
+        '-v', f'{Path(hf).expanduser()}:/root/.cache/huggingface',
+        '-e', f'HF_ENDPOINT={os.environ.get("HF_ENDPOINT", "https://hf-mirror.com")}',
+        image,
+        '--model', cfg.get('model_id', cfg.get('model', model)),
+        '--served-model-name', model,
+        '--port', str(port),
+        *engine_args,
+        *shlex.split(cfg.get('extra_args', '')),
+    ]
+    for key, flag in (('gpu_mem_util', '--gpu-memory-utilization'),
+                      ('max_model_len', '--max-model-len'),
+                      ('tp_size', '--tensor-parallel-size'),
+                      ('dtype', '--dtype')):
+        if cfg.get(key):
+            cmd += [flag, str(cfg[key])]
+    r = _run(cmd)
+    if r.returncode != 0:
+        raise RuntimeError(f'docker run failed: {r.stderr[:500]}')
+    api_base = f'http://localhost:{port}/v1'
+    _wait_healthy(api_base, int(cfg.get('timeout_s', 1800)))
+    return {'api_base': api_base, 'model': model, 'container': container, 'port': str(port)}
+
+
+def _wait_healthy(api_base: str, timeout_s: int) -> None:
+    import urllib.request
+
+    deadline = time.time() + timeout_s
+    while time.time() < deadline:
+        try:
+            with urllib.request.urlopen(f'{api_base}/models', timeout=5) as resp:
+                if resp.status == 200:
+                    return
+        except Exception:
+            time.sleep(5)
+    raise TimeoutError(f'serving engine not healthy after {timeout_s}s at {api_base}')
+
+
+def docker_stop(handle: EnvHandle) -> None:
+    if handle.container:
+        _run(['docker', 'rm', '-f', handle.container])
+
+
+def serve_env(engine: str, model: str, cfg: Dict[str, Any],
+              default_image: str, engine_args: Optional[List[str]] = None) -> EnvHandle:
+    """acquire() wrapper: shared, refcounted serve environment."""
+    return acquire(
+        kind=engine,
+        name=model,
+        start_fn=lambda _m: docker_serve(engine, _m, cfg, default_image, engine_args or []),
+        stop_fn=docker_stop,
+    )
diff --git a/evalharness/sandbox/local.py b/evalharness/sandbox/local.py
new file mode 100644
index 0000000..45bec84
--- /dev/null
+++ b/evalharness/sandbox/local.py
@@ -0,0 +1,45 @@
+"""Local sandbox: NO isolation, dev convenience only.
+
+Runs code in a subprocess on the host with a timeout. Fine for quick
+iteration on harnesses; never use for untrusted model output in shared
+environments — switch the recipe to sandbox='docker' for that.
+"""
+
+import subprocess
+import tempfile
+import time
+from pathlib import Path
+from typing import Dict, Optional
+
+from .base import ExecResult, Sandbox, register_sandbox
+
+
+@register_sandbox('local')
+class LocalSandbox(Sandbox):
+    name = 'local'
+
+    def exec(
+        self,
+        files: Dict[str, str],
+        entry: str = 'main.py',
+        mounts: Optional[Dict[str, str]] = None,
+        timeout_s: int = 60,
+        image: str = '',
+    ) -> ExecResult:
+        with tempfile.TemporaryDirectory(prefix='eh-local-') as td:
+            work = Path(td)
+            for fname, content in (files or {}).items():
+                dest = work / fname
+                dest.parent.mkdir(parents=True, exist_ok=True)
+                dest.write_text(content, encoding='utf-8')
+            t0 = time.time()
+            try:
+                proc = subprocess.run(
+                    ['python', str(work / entry)], capture_output=True, text=True,
+                    timeout=timeout_s, cwd=work,
+                )
+                return ExecResult(exit_code=proc.returncode, stdout=proc.stdout,
+                                  stderr=proc.stderr, duration_s=round(time.time() - t0, 2))
+            except subprocess.TimeoutExpired:
+                return ExecResult(exit_code=-1, timed_out=True, duration_s=timeout_s,
+                                  error=f'timeout after {timeout_s}s')
diff --git a/tests/test_sandbox_agent.py b/tests/test_sandbox_agent.py
new file mode 100644
index 0000000..40ae990
--- /dev/null
+++ b/tests/test_sandbox_agent.py
@@ -0,0 +1,115 @@
+"""Sandbox + agent driver tests.
+
+Run:  .venv/bin/python tests/test_sandbox_agent.py
+"""
+
+import asyncio
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from evalharness import get_dataset  # noqa: E402
+from evalharness.agent import BFCLEnvironment, drive, trajectory_to_prediction  # noqa: E402
+from evalharness.data.sample import Sample  # noqa: E402
+from evalharness.model import resolve_adapter, run_eval  # noqa: E402
+from evalharness.sandbox import get_sandbox  # noqa: E402
+
+
+def test_local_sandbox():
+    sbx = get_sandbox('local')
+    r = sbx.exec({'main.py': 'print(6*7)'})
+    assert r.ok and '42' in r.stdout
+    r2 = sbx.exec({'main.py': 'raise RuntimeError("boom")'})
+    assert not r2.ok and 'boom' in r2.stderr
+    r3 = sbx.exec({'sub/mod.py': 'x=1', 'main.py': 'print("m")'})  # nested files
+    assert r3.ok
+
+
+def test_docker_sandbox_if_available():
+    from evalharness.sandbox import docker_available
+
+    if not docker_available():
+        print('  (docker unavailable, skipped)')
+        return
+    sbx = get_sandbox('docker')
+    r = sbx.exec({'main.py': 'print("docker-ok")'}, timeout_s=300)
+    assert r.ok and 'docker-ok' in r.stdout, r.stderr[:200]
+    r2 = sbx.exec({'main.py': 'import socket\n'
+                   's=socket.create_connection(("1.1.1.1", 80), timeout=5)\nprint("net-ok")'},
+                  timeout_s=120)
+    assert not r2.ok  # --network none: outbound connect must fail
+
+
+def test_env_handle_refcount():
+    from evalharness.sandbox.base import EnvHandle, _ACTIVE, acquire
+
+    stops = []
+
+    def start(name):
+        return {'api_base': f'http://x/{name}', 'model': name, 'container': f'c-{name}'}
+
+    h = acquire('t-engine', 'm1', start, lambda h: stops.append(h.container))
+    h2 = acquire('t-engine', 'm1', start, lambda h: stops.append(h.container))
+    assert h2 is h and h.refs == 2
+    h.release()  # shared: must NOT stop
+    assert stops == []
+    h.release()  # last ref: stop + unregister
+    assert stops == ['c-m1'] and 't-engine/m1' not in _ACTIVE
+    h.release()  # idempotent
+    assert stops == ['c-m1']
+
+
+def test_agent_drive_single_turn():
+    adapter = resolve_adapter('mock')
+
+    async def go():
+        return await drive(adapter, Sample(input='hi', target='hi'))
+
+    traj = asyncio.run(go())
+    assert traj.turns == 1 and traj.messages[-1]['role'] == 'assistant'
+
+
+def test_bfcl_env_records_calls():
+    adapter = resolve_adapter('mock')
+    adapter.extra['mode'] = 'fc'
+    gt = '{"tool_calls": [{"name": "f", "arguments": {"a": 1}}]}'
+    sample = Sample(input='call f', target=gt, task_type='fc')
+
+    async def go():
+        return await drive(adapter, sample, env=BFCLEnvironment(), max_turns=4)
+
+    traj = asyncio.run(go())
+    assert traj.env_state['calls'] == [{'name': 'f', 'arguments': {'a': 1}}]
+    pred = trajectory_to_prediction(traj)
+    assert pred['env_state']['ground_truth']['tool_calls'][0]['name'] == 'f'
+
+
+def test_bfcl_pipeline_oracle():
+    ds = get_dataset('bfcl_v3')
+    rep = asyncio.run(run_eval(ds, 'mock:fc', env='bfcl_mock', limit=30,
+                               concurrency=8, progress=False))
+    assert rep.metrics['acc'] == 1.0
+    assert rep.metric_groups['acc']['irrelevance'] == 1.0
+
+
+def test_humaneval_pipeline_oracle():
+    ds = get_dataset('humaneval')
+    rep = asyncio.run(run_eval(ds, 'mock:oracle', limit=10, concurrency=4, progress=False))
+    assert rep.metrics['pass'] == 1.0
+
+
+if __name__ == '__main__':
+    fails = 0
+    for name, fn in sorted({k: v for k, v in globals().items()
+                            if k.startswith('test_') and callable(v)}.items()):
+        try:
+            fn()
+            print(f'PASS {name}')
+        except AssertionError as e:
+            fails += 1
+            print(f'FAIL {name}: {e}')
+        except Exception as e:
+            fails += 1
+            print(f'ERROR {name}: {type(e).__name__}: {e}')
+    sys.exit(1 if fails else 0)