feat(sandbox): driver-independent protected mock service for CLI fixtures

docs/TASK_DEFINITION_GUIDE.md

@@ -541,6 +541,60 @@ Notes:
 - **Collisions are rejected.** If a `mock_path_dirs` entry already provides an executable of the same name, setup raises rather than letting directory order decide which one runs.
 - **It stubs a tool; it does not proxy one, and it does not serve per-invocation responses.** Recording a *real* executable on the way through, or returning different output per invocation, stays a hand-written mock under `mock_path_dirs` — both depend on state the harness cannot guarantee (the tool being installed, PATH order, live credentials, a fixture set).
 
+### Protected Fixture-Backed CLIs
+
+Use `protected_mocks` when `uip` or another mock needs different fixture-backed responses per invocation and the fixture itself must not be readable by the evaluated agent. The fixture is loaded host-side by a small per-run server; the agent's workspace only ever contains a thin client shim, so no encoding or sealing of grading material is involved:
+
+```yaml
+sandbox:
+  driver: tempdir
+  protected_mocks:
+    - tool: uip
+      fixture: ./fixtures/uip-troubleshoot.json
+      max_requests: 100
+      passthrough_argv_prefixes:
+        - [docsai, ask]
+```
+
+Notes:
+
+- **Drivers.** Supported under `driver: tempdir` (the default). Under `driver: docker` the task fails validation: protected mocks under the docker driver require the UID/GID isolation layer, which is not yet available — that combination fails closed rather than running without the isolation it assumes.
+- **Endpoint.** The server binds a per-run endpoint at start: an AF_UNIX socket in a run-scoped scratch directory when the platform supports it (probed with a real bind), else TCP on `127.0.0.1` with an ephemeral port. Every request must carry the run's random token, which the generated shim bakes in. The token keeps other local processes from casually querying the service; it is same-user hygiene, not a security boundary — the protection comes from the server only ever answering with configured command responses, never fixture contents.
+- **Fixture paths.** `fixture` resolves against the task YAML's directory (like `uipath_eval.eval_set`). The file is read host-side only and is never copied into the sandbox.
+- **Shims.** Each entry generates a `protected_mocks/<tool>` shim (plus a `.cmd` twin for Windows PATHEXT lookup) that is PATH-prepended for the agent exactly like `mock_path_dirs` entries. The shim carries its endpoint, token, and call-log path itself; it does not rely on the agent's environment.
+- **Call log.** Every invocation is appended, in the `cli_called` JSON Lines schema, to `protected_mock_calls.jsonl` next to `task.json` in the run directory — outside the sandbox. This is a diagnostic surface: the [`cli_called`](#cli_called) criterion resolves its `log` field sandbox-relative and cannot read this host-side file today.
+- **Budget and audit.** `max_requests` caps calls per tool per run (exceeded calls get exit 75). The run's `environment_info` records the endpoint kind and a SHA-256 digest of the fixture contents.
+- `protected_mocks` and `record_cli` cannot claim the same tool name.
+
+Fixture files map argument lists to responses:
+
+```json
+{
+  "version": 1,
+  "responses": [
+    {
+      "argv": ["rpa", "get-errors", "--output", "json"],
+      "exit_code": 0,
+      "stdout": "{\"errors\":[]}\n",
+      "stderr": ""
+    }
+  ],
+  "default": {
+    "exit_code": 2,
+    "stderr": "command not configured for this scenario\n"
+  }
+}
+```
+
+Matching defaults to exact argv equality. Two further modes exist, selected per response via `match_mode`:
+
+- `"normalized"` still selects from a finite command map but ignores `--output <format>`, treats `--flag=value` like `--flag value`, and permits token reordering. Duplicate keys are rejected at load for both finite modes.
+- `"subset"` matches when every rule token appears in the invocation's normalized token set, regardless of order or extra arguments. Subset rules are evaluated in fixture-file order and the first match wins; exact and normalized matches always take precedence over subset scanning. Duplicate subset rules are allowed (an earlier rule shadows a later one); an empty subset `argv` is rejected at load.
+
+Malformed responses, oversized output, request-budget exhaustion, and service startup failures all fail loudly.
+
+`passthrough_argv_prefixes` is for deliberately public live operations such as `uip docsai ask`. The server invokes the real tool only when argv begins with one of these typed prefixes, caches the response in memory for the run, and never reveals the executable path to the agent. Do not use a broad prefix such as `[or]` or `[auth]`.
+
 ## Template Sources
 
 Tasks can start with preset files instead of an empty sandbox. Multiple sources are applied sequentially (last wins for conflicts).

src/coder_eval/models/__init__.py

@@ -155,11 +155,13 @@
 
 # Sandbox
 from coder_eval.models.sandbox import (
+    PROTECTED_MOCK_DIR,
     RECORD_CLI_DIR,
     RECORD_CLI_LOG,
     DockerBuildConfig,
     DockerDriverConfig,
     NodeEnvConfig,
+    ProtectedMockConfig,
     PythonEnvConfig,
     RecordedCli,
     ResourceLimits,
@@ -274,6 +276,8 @@
     "NodeEnvConfig",
     "PythonEnvConfig",
     "SandboxConfig",
+    "ProtectedMockConfig",
+    "PROTECTED_MOCK_DIR",
     "RecordedCli",
     "RECORD_CLI_DIR",
     "RECORD_CLI_LOG",

src/coder_eval/models/sandbox.py

@@ -398,6 +398,69 @@ def validate_tool_name(cls, v: str) -> str:
         return v
 
 
+# Sandbox-relative directory the protected-mock client shims are generated into.
+# Separate from RECORD_CLI_DIR so a wipe-and-regenerate of either feature can
+# never delete the other's shims. Not dot-prefixed for the same artifact-upload
+# reason as RECORD_CLI_DIR.
+PROTECTED_MOCK_DIR = "protected_mocks"
+
+
+class ProtectedMockConfig(BaseModel):
+    """Fixture-backed CLI served by the host-side protected mock service.
+
+    ``fixture`` is read host-side by the service process, never by the
+    evaluated agent, and is never copied into the agent workspace. The fixture
+    schema maps argv lists to bounded stdout/stderr/exit-code responses; it
+    exposes no general file or search operation. Relative ``fixture`` paths
+    resolve against the task YAML's directory.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    tool: str = Field(description="Bare executable name presented to the agent (for example, 'uip')")
+    fixture: str = Field(
+        description=(
+            "Path to the protected command-response fixture, resolved against the task YAML's "
+            "directory when relative. Read host-side only; never copied into the sandbox."
+        )
+    )
+    max_requests: int = Field(default=100, ge=1, le=10_000, description="Per-run request budget for this tool")
+    passthrough_argv_prefixes: list[list[str]] = Field(
+        default_factory=list,
+        max_length=16,
+        description=(
+            "Public argv prefixes the service may proxy to the real tool, for example [['docsai', 'ask']]. "
+            "All other invocations remain fixture-backed or receive the fixed default response."
+        ),
+    )
+
+    @field_validator("tool")
+    @classmethod
+    def validate_tool_name(cls, value: str) -> str:
+        if not value or value != value.strip() or value in {".", ".."} or "/" in value or "\\" in value:
+            raise ValueError("protected mock tool must be a non-empty bare executable name")
+        if value.lower().endswith((".cmd", ".bat", ".exe")):
+            raise ValueError("protected mock tool must not include a platform executable suffix")
+        return value
+
+    @field_validator("passthrough_argv_prefixes")
+    @classmethod
+    def validate_passthrough_prefixes(cls, prefixes: list[list[str]]) -> list[list[str]]:
+        normalized: list[list[str]] = []
+        seen: set[tuple[str, ...]] = set()
+        for prefix in prefixes:
+            if not prefix or len(prefix) > 8:
+                raise ValueError("protected mock passthrough prefixes must contain 1 to 8 argv tokens")
+            if any(not isinstance(token, str) or not token or len(token) > 256 for token in prefix):
+                raise ValueError("protected mock passthrough prefix tokens must be non-empty strings up to 256 chars")
+            key = tuple(prefix)
+            if key in seen:
+                raise ValueError("protected mock passthrough prefixes must be unique")
+            seen.add(key)
+            normalized.append(list(prefix))
+        return normalized
+
+
 class SandboxConfig(BaseModel):
     """Configuration for the sandboxed execution environment.
 
@@ -451,6 +514,17 @@ class SandboxConfig(BaseModel):
         ),
     )
 
+    protected_mocks: list[ProtectedMockConfig] | None = MergeField(
+        strategy="replace",
+        default=None,
+        description=(
+            "Fixture-backed mock CLIs served by a host-side per-run service. The agent receives a thin "
+            "client shim; fixture bytes stay host-side and are never copied into its workspace. "
+            "Supported under driver: tempdir. Under driver: docker this fails validation until the "
+            "UID/GID isolation layer lands. Replaced (not merged) across config layers."
+        ),
+    )
+
     record_cli: list[RecordedCli] | None = MergeField(
         strategy="replace",
         default=None,
@@ -489,4 +563,17 @@ def validate_template_sources(self) -> SandboxConfig:
         """Validate template sources configuration."""
         if self.template_sources:
             validate_template_sources_list(self.template_sources)
+        if self.protected_mocks:
+            if self.driver == "docker":
+                raise ValueError(
+                    "sandbox.protected_mocks under the docker driver requires the UID/GID isolation "
+                    + "layer; not yet available. Use driver: tempdir."
+                )
+            tools = [mock.tool for mock in self.protected_mocks]
+            if len(tools) != len(set(tools)):
+                raise ValueError("sandbox.protected_mocks tool names must be unique")
+            recorded = {spec.tool for spec in self.record_cli or []}
+            overlap = sorted(recorded & set(tools))
+            if overlap:
+                raise ValueError(f"protected_mocks and record_cli cannot both provide: {overlap}")
         return self

src/coder_eval/orchestrator.py

@@ -59,6 +59,8 @@
 from .orchestration.early_stop import EarlyStopWatcher, early_stop_active, validate_early_stop
 from .orchestration.evaluation import load_reference
 from .path_utils import format_task_log_id, task_log_path
+from .protected_mock.runtime import CALL_LOG_NAME as PROTECTED_MOCK_CALL_LOG_NAME
+from .protected_mock.runtime import ProtectedMockRuntime
 from .sandbox import Sandbox
 from .simulation import DialogStopReason, SimulatorResult, UserSimulator, evaluate_stop
 from .streaming.callbacks import CompositeStreamCallback, StreamCallback, TaskScopedCallback, safe_emit
@@ -395,6 +397,10 @@ def __init__(
         # so the default path is entirely unaffected).
         self._early_stop_watcher: EarlyStopWatcher | None = None
 
+        # Protected mock service (started in _setup only when the task declares
+        # sandbox.protected_mocks; stopped unconditionally in _cleanup).
+        self._protected_mock_runtime: ProtectedMockRuntime | None = None
+
         # One-shot flag: emit the "cost budget configured but no cost data" warning
         # exactly once per task even if _check_run_limits fires every turn.
         self._cost_budget_skipped_logged: bool = False
@@ -1077,6 +1083,12 @@ async def _setup_sandbox() -> Any:
         logger.info("API routing: %s", _format_routing(self.route, self.task.agent.model if self.task.agent else None))
         self.success_checker = SuccessChecker(self.sandbox, route=self.eval_route)
 
+        # Start the protected mock service (host-side fixture server) and write
+        # its client shims into the sandbox BEFORE the agent's PATH is assembled
+        # below. Teardown is unconditional in _cleanup, which runs on every exit
+        # path (success, crash, timeout, early stop) via run()'s finally block.
+        await self._start_protected_mocks()
+
         # Create and start the agent. For a no-op (type: none) task this dispatches
         # to NoOpAgent, whose start/communicate/stop are no-ops — the orchestrator
         # runs the normal lifecycle without any agentless branching, and the
@@ -1116,6 +1128,51 @@ async def _start_agent() -> None:
         if self.sandbox and self.sandbox.installed_tool_versions:
             self.result.environment_info["installed_tools"] = self.sandbox.installed_tool_versions
 
+        # After the environment_info re-capture above, which would otherwise
+        # discard keys written earlier in _setup.
+        self._record_protected_mock_environment_info()
+
+    async def _start_protected_mocks(self) -> None:
+        """Start the host-side fixture service and generate its sandbox shims.
+
+        No-op unless the task declares ``sandbox.protected_mocks``. Fixture
+        paths resolve against the task YAML's directory host-side; fixture
+        bytes never enter the sandbox. The per-task invocation log lands next
+        to task.json in the run dir, outside the sandbox.
+        """
+        mocks = self.task.sandbox.protected_mocks
+        if not mocks:
+            return
+        assert self.sandbox is not None
+        task_dir = self.task_file.parent.resolve() if self.task_file else None
+        runtime = ProtectedMockRuntime(mocks, task_dir=task_dir)
+        await asyncio.to_thread(runtime.start)
+        # Stored before shim generation so _cleanup stops the server even if
+        # generation fails.
+        self._protected_mock_runtime = runtime
+        call_log = self.run_dir / PROTECTED_MOCK_CALL_LOG_NAME
+        # Seed the log so the diagnostic surface always exists, even for a run
+        # that never called the tool.
+        await asyncio.to_thread(call_log.write_text, "", encoding="utf-8")
+        await asyncio.to_thread(
+            self.sandbox.generate_protected_mock_shims,
+            endpoint=runtime.endpoint,
+            token=runtime.token,
+            call_log=call_log,
+        )
+        logger.info(
+            "Protected mock service up (%s endpoint) for tool(s): %s",
+            runtime.endpoint_kind,
+            ", ".join(mock.tool for mock in mocks),
+        )
+
+    def _record_protected_mock_environment_info(self) -> None:
+        """Persist the protected mock audit record into ``result.environment_info``."""
+        if self._protected_mock_runtime is None or self.result is None:
+            return
+        self.result.environment_info["protected_mock_endpoint_kind"] = self._protected_mock_runtime.endpoint_kind
+        self.result.environment_info["protected_mock_fixture_digest"] = self._protected_mock_runtime.fixture_digest
+
     def _sync_sandbox_command_path_with_agent(self) -> None:
         """Align criteria command PATH with the PATH used for the last agent query.
 
@@ -2269,6 +2326,15 @@ async def _cleanup(self) -> None:
             except Exception as e:
                 logger.warning(f"Failed to stop agent: {e}")
 
+        # Stop the protected mock service. After the agent (no in-flight calls),
+        # before sandbox teardown; the server must never outlive the task.
+        if self._protected_mock_runtime is not None:
+            try:
+                await asyncio.to_thread(self._protected_mock_runtime.stop)
+            except Exception as e:
+                logger.warning(f"Failed to stop protected mock service: {e}")
+            self._protected_mock_runtime = None
+
         # Cleanup sandbox. Preservation and cleanup() are SIBLING try blocks:
         # a preservation failure (e.g. disk full during preserve_to) must never
         # skip cleanup(), or the tempdir leaks.

src/coder_eval/protected_mock/__init__.py

@@ -0,0 +1 @@
+"""Protected fixture-backed CLI service: host-side server, thin sandbox client shim."""