Add interception example environments

environments/bash_interception/README.md

@@ -0,0 +1,5 @@
+# bash-interception
+
+Shows both Bash outcomes: `@vf.stop` can block a proposed command before the
+harness sees it, while `@vf.intercept` can supply a synthetic tool result and
+let the model continue without running the command.

environments/bash_interception/bash_interception/__init__.py

@@ -0,0 +1,3 @@
+from bash_interception.taskset import BashInterceptionTaskset
+
+__all__ = ["BashInterceptionTaskset"]

environments/bash_interception/bash_interception/taskset.py

@@ -0,0 +1,91 @@
+import json
+
+import verifiers.v1 as vf
+
+STOP_SENTINEL = "bash-tool-executed"
+STOP_COMMAND = f"touch {STOP_SENTINEL}"
+REWRITE_SENTINEL = "git-tool-executed"
+REWRITE_COMMAND = f"git --version && touch {REWRITE_SENTINEL}"
+
+
+def command(message: vf.AssistantMessage, snippet: str) -> vf.ToolCall | None:
+    for call in message.tool_calls or []:
+        if call.name != "bash":
+            continue
+        try:
+            arguments = json.loads(call.arguments)
+        except (json.JSONDecodeError, TypeError):
+            continue
+        if isinstance(arguments, dict) and snippet in arguments.get("command", ""):
+            return call
+    return None
+
+
+class BashInterceptionTask(vf.Task):
+    @vf.stop
+    def stop_bash(self, response: vf.Response) -> bool:
+        # The response is still buffered here. Stopping prevents Bash from
+        # receiving the proposed call, so the command cannot execute.
+        return (
+            self.data.idx == 0 and command(response.message, STOP_COMMAND) is not None
+        )
+
+    @vf.intercept
+    def rewrite_git(self, request: vf.Request) -> vf.Request | None:
+        if self.data.idx != 1 or not request.messages:
+            return None
+        result = request.messages[-1]
+        if not isinstance(result, vf.ToolMessage) or result.content:
+            return None
+        assistant = next(
+            (
+                message
+                for message in reversed(request.messages[:-1])
+                if isinstance(message, vf.AssistantMessage)
+                and any(
+                    call.id == result.tool_call_id for call in message.tool_calls or []
+                )
+            ),
+            None,
+        )
+        call = command(assistant, REWRITE_SENTINEL) if assistant is not None else None
+        if call is None or call.id != result.tool_call_id:
+            return None
+        # This fills the pre-execution result. Bash returns it to the model instead
+        # of running the command, and the rollout continues normally.
+        replacement = result.model_copy(
+            update={
+                "content": "This request is blocked. You should answer with something."
+            }
+        )
+        return request.model_copy(
+            update={"messages": [*request.messages[:-1], replacement]}
+        )
+
+    @vf.reward
+    async def rewritten(self, trace: vf.Trace, runtime: vf.Runtime) -> float:
+        sentinel = STOP_SENTINEL if self.data.idx == 0 else REWRITE_SENTINEL
+        executed = (await runtime.run(["test", "-e", sentinel], {})).exit_code == 0
+        if self.data.idx == 0:
+            return float(not executed and trace.stop_condition == "stop_bash")
+        return float(
+            not executed
+            and bool(trace.tool_messages)
+            and "This request is blocked" in str(trace.tool_messages[-1].content)
+            and trace.num_turns == 2
+        )
+
+
+class BashInterceptionTaskset(vf.Taskset[BashInterceptionTask]):
+    def load(self) -> list[BashInterceptionTask]:
+        prompts = (
+            f"Use bash once to run `{STOP_COMMAND}`, then stop.",
+            (
+                f"Use bash exactly once to run `{REWRITE_COMMAND}`. Whatever the tool "
+                "returns, do not call another tool; answer with something."
+            ),
+        )
+        return [
+            BashInterceptionTask(vf.TaskData(idx=i, prompt=prompt), self.config.task)
+            for i, prompt in enumerate(prompts)
+        ]

environments/bash_interception/pyproject.toml

@@ -0,0 +1,13 @@
+[project]
+name = "bash-interception"
+version = "0.1.0"
+description = "Stop or replace Bash tool calls before they execute."
+requires-python = ">=3.11"
+dependencies = ["verifiers"]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["bash_interception"]

environments/grayscale_interception/README.md

@@ -0,0 +1,4 @@
+# grayscale-interception
+
+Uses `@vf.intercept` on a typed `vf.Request` to convert an embedded image to
+grayscale before either the harness or the model stores it.

environments/grayscale_interception/grayscale_interception/__init__.py

@@ -0,0 +1,3 @@
+from grayscale_interception.taskset import GrayscaleInterceptionTaskset
+
+__all__ = ["GrayscaleInterceptionTaskset"]

environments/grayscale_interception/grayscale_interception/taskset.py

@@ -0,0 +1,88 @@
+import base64
+from io import BytesIO
+
+from PIL import Image, ImageOps
+
+import verifiers.v1 as vf
+from verifiers.v1.utils.image import image_data_url
+
+
+class GrayscaleInterceptionTask(vf.Task):
+    @vf.intercept
+    def grayscale(self, request: vf.Request) -> vf.Request | None:
+        # The last request message is the new user input, before the harness stores it.
+        message = request.messages[-1]
+        if not isinstance(message, vf.UserMessage) or not isinstance(
+            message.content, list
+        ):
+            return None
+        content: list[vf.ContentPart] = []
+        changed = False
+        for part in message.content:
+            if not isinstance(
+                part, vf.ImageUrlContentPart
+            ) or not part.image_url.url.startswith("data:image/"):
+                content.append(part)
+                continue
+            metadata, separator, encoded = part.image_url.url.partition(",")
+            if not separator or not metadata.lower().endswith(";base64"):
+                content.append(part)
+                continue
+            try:
+                with Image.open(
+                    BytesIO(base64.b64decode(encoded, validate=True))
+                ) as image:
+                    alpha = (
+                        image.convert("RGBA").getchannel("A")
+                        if image.has_transparency_data
+                        else None
+                    )
+                    grayscale_image = ImageOps.grayscale(image)
+                    if alpha is not None:
+                        grayscale_image.putalpha(alpha)
+                    grayscale = image_data_url(grayscale_image)
+            except (ValueError, OSError):
+                content.append(part)
+                continue
+            content.append(
+                part.model_copy(
+                    update={
+                        "image_url": part.image_url.model_copy(
+                            update={"url": grayscale}
+                        )
+                    }
+                )
+            )
+            changed = True
+        if not changed:
+            return None
+        messages = [
+            *request.messages[:-1],
+            message.model_copy(update={"content": content}),
+        ]
+        return request.model_copy(update={"messages": messages})
+
+    @vf.reward
+    async def changed(self, trace: vf.Trace) -> float:
+        return float(bool(trace.request_rewrites))
+
+
+class GrayscaleInterceptionTaskset(vf.Taskset[GrayscaleInterceptionTask]):
+    def load(self) -> list[GrayscaleInterceptionTask]:
+        # Building the image here keeps the example self-contained.
+        image = Image.new("RGB", (64, 64), "orange")
+        prompt = [
+            vf.UserMessage(
+                content=[
+                    vf.ImageUrlContentPart(
+                        image_url=vf.ImageUrlSource(url=image_data_url(image))
+                    ),
+                    vf.TextContentPart(text="Describe this image."),
+                ]
+            )
+        ]
+        return [
+            GrayscaleInterceptionTask(
+                vf.TaskData(idx=0, prompt=prompt), self.config.task
+            )
+        ]

environments/grayscale_interception/pyproject.toml

@@ -0,0 +1,16 @@
+[project]
+name = "grayscale-interception"
+version = "0.1.0"
+description = "Rewrite user images to grayscale before a model sees them."
+requires-python = ">=3.11"
+dependencies = [
+    "pillow>=12.3.0",
+    "verifiers",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["grayscale_interception"]

environments/interception/README.md

@@ -0,0 +1,4 @@
+# interception
+
+Four typed response-boundary examples: deterministic and judge-based `@vf.intercept`
+rewrites, a boolean `@vf.stop`, and a metric recorded without changing the response.

environments/interception/interception/__init__.py

@@ -0,0 +1,3 @@
+from interception.taskset import InterceptionTaskset
+
+__all__ = ["InterceptionTaskset"]

environments/interception/interception/taskset.py

@@ -0,0 +1,73 @@
+import verifiers.v1 as vf
+
+
+class InterceptionTaskConfig(vf.TaskConfig):
+    judge: vf.JudgeConfig = vf.JudgeConfig()
+
+
+class InterceptionTask(vf.Task[vf.TaskData, vf.State, InterceptionTaskConfig]):
+    @vf.stop
+    def stop_guard(self, response: vf.Response) -> bool:
+        # A response stop runs while the response is buffered, before the harness
+        # receives it. The function name is stored as trace.stop_condition.
+        return "STOP" in (response.message.content or "")
+
+    @vf.intercept(priority=10)
+    def deterministic_guard(self, response: vf.Response) -> vf.Response | None:
+        if "DETERMINISTIC_BLOCK" in (response.message.content or ""):
+            # Return the same boundary type with the replacement assistant message.
+            return response.model_copy(
+                update={"message": vf.AssistantMessage(content="Blocked")}
+            )
+
+    @vf.intercept
+    async def judge_guard(
+        self, response: vf.Response, trace: vf.Trace
+    ) -> vf.Response | None:
+        candidate = response.message.content or ""
+        if "JUDGE_BLOCK" not in candidate:
+            return None
+        # Trace is the canonical history; Response is the candidate being judged.
+        verdict = await vf.Judge(self.config.judge).complete(
+            "Reply BLOCK if the candidate contains JUDGE_BLOCK; otherwise ALLOW.\n\n"
+            f"{trace.transcript}\n\nCandidate:\n{candidate}",
+            trace=trace,
+        )
+        choice = vf.parse_judge_choice(verdict.text, choices=("BLOCK", "ALLOW"))
+        if choice is None:
+            raise ValueError(f"judge returned no BLOCK/ALLOW verdict: {verdict.text!r}")
+        if choice == "BLOCK":
+            return response.model_copy(
+                update={"message": vf.AssistantMessage(content="Blocked")}
+            )
+
+    @vf.intercept
+    def metric_only(self, response: vf.Response, trace: vf.Trace) -> None:
+        # Returning None keeps the response unchanged; side effects such as
+        # recording a metric still remain on the trace.
+        if "METRIC_ONLY" in (response.message.content or ""):
+            trace.record_metric("response/metric_only", 1.0)
+
+    @vf.reward
+    async def changed(self, trace: vf.Trace) -> float:
+        return float(
+            bool(trace.response_rewrites)
+            or trace.stop_condition == "stop_guard"
+            or "response/metric_only" in trace.metrics
+        )
+
+
+class InterceptionConfig(vf.TasksetConfig):
+    task: InterceptionTaskConfig = InterceptionTaskConfig()
+
+
+class InterceptionTaskset(vf.Taskset[InterceptionTask, InterceptionConfig]):
+    def load(self) -> list[InterceptionTask]:
+        markers = ("DETERMINISTIC_BLOCK", "JUDGE_BLOCK", "STOP", "METRIC_ONLY")
+        return [
+            InterceptionTask(
+                vf.TaskData(idx=index, prompt=f"Reply with exactly {marker}."),
+                self.config.task,
+            )
+            for index, marker in enumerate(markers)
+        ]

environments/interception/pyproject.toml

@@ -0,0 +1,13 @@
+[project]
+name = "interception"
+version = "0.1.0"
+description = "Examples of typed response interception, stopping, and metrics."
+requires-python = ">=3.11"
+dependencies = ["verifiers"]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["interception"]

environments/web_search_interception/README.md

@@ -0,0 +1,4 @@
+# web-search-interception
+
+Inspects a buffered Codex response, including provider-native web-search items,
+and uses `@vf.stop` when the search material contains a chosen phrase.

environments/web_search_interception/pyproject.toml

@@ -0,0 +1,13 @@
+[project]
+name = "web-search-interception"
+version = "0.1.0"
+description = "Stop after inspecting native web-search results in a buffered response."
+requires-python = ">=3.11"
+dependencies = ["verifiers"]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["web_search_interception"]

environments/web_search_interception/web_search_interception/__init__.py

@@ -0,0 +1,10 @@
+from verifiers.v1.harnesses.codex import CodexHarness, CodexHarnessConfig
+from web_search_interception.taskset import WebSearchInterceptionTaskset
+
+# Exporting Codex beside the taskset makes it this example's default harness, so
+# the response contains the provider-native web-search items the stop inspects.
+__all__ = [
+    "CodexHarness",
+    "CodexHarnessConfig",
+    "WebSearchInterceptionTaskset",
+]

environments/web_search_interception/web_search_interception/taskset.py

@@ -0,0 +1,38 @@
+import verifiers.v1 as vf
+
+BLOCKED_WORD = "openai.com"
+
+
+class WebSearchInterceptionTask(vf.Task):
+    @vf.stop
+    def stop_on_result(self, response: vf.Response) -> bool:
+        # Native web-search citations remain on the buffered provider response.
+        sources = (
+            annotation.get("url", "")
+            for item in response.message.provider_state or []
+            if item.get("type") == "message"
+            for part in item.get("content") or []
+            for annotation in part.get("annotations") or []
+            if annotation.get("type") == "url_citation"
+        )
+        return any(BLOCKED_WORD in source.casefold() for source in sources)
+
+    @vf.reward
+    async def blocked(self, trace: vf.Trace) -> float:
+        return float(trace.stop_condition == "stop_on_result")
+
+
+class WebSearchInterceptionTaskset(vf.Taskset[WebSearchInterceptionTask]):
+    def load(self) -> list[WebSearchInterceptionTask]:
+        return [
+            WebSearchInterceptionTask(
+                vf.TaskData(
+                    idx=0,
+                    prompt=(
+                        "Use native web search to find the official OpenAI Responses "
+                        "API documentation. Cite the source and include the query."
+                    ),
+                ),
+                self.config.task,
+            )
+        ]