fix(v1): transport unbounded harness input safely

environments/compact/compact/harness.py

@@ -51,4 +51,7 @@ async def launch(
                 {"mcpServers": {name: {"url": url} for name, url in mcp_urls.items()}}
             )
         program = await runtime.prepare_uv_script(PROGRAM_SOURCE, self.config.env)
-        return await runtime.run_program([*program, trace.task.data.prompt], env)
+        prompt = trace.task.data.prompt
+        if not isinstance(prompt, str):
+            raise ValueError("compact harness requires a string task prompt")
+        return await runtime.run_program(program, env, stdin=prompt.encode("utf-8"))

environments/compact/compact/program.py

@@ -99,7 +99,7 @@ async def call_mcp(dispatch: dict, name: str, arguments: dict) -> str:
 
 
 async def main() -> None:
-    task = sys.argv[1]
+    task = sys.stdin.read()
     config = json.loads(os.environ.get("MCP_CONFIG", "{}"))
     notes: str | None = None  # the durable memory carried across turns
     tool_output: str | None = None  # the last tool result, kept for exactly one turn

verifiers/v1/harnesses/bash/harness.py

@@ -68,19 +68,13 @@ async def launch(
             f"--base-url={endpoint}",
             f"--api-key={secret}",
             f"--model={ctx.model}",
-            f"--system-prompt={system_prompt}",
+            "--payload-stdin",
         ]
         if self.config.edit:
             args.append("--edit")
         if self.config.search:
-            # Resolve the key and keep it OUT of the program env: it's handed to the program over
-            # argv (--serper-key), so popping it here stops the agent's `bash` subprocesses from
-            # inheriting it via $SERPER_API_KEY / /proc/self/environ. Prefer a key set in the harness
-            # env (--harness.env / forward_env); fall back to the host env only when the key is
-            # *absent* (None), not present-but-empty — a rollout setting SERPER_API_KEY="" is
-            # deliberately masking the host secret, so honor that (the check below then fails loudly
-            # rather than leaking the host key). The pop is scoped to search=true, so an unrelated
-            # key forwarded for the agent's own bash-side use is left untouched.
+            # Keep the search key out of the program environment so agent-spawned Bash commands
+            # cannot inherit it. It is bounded secret/config data rather than task payload.
             serper_key = env.pop("SERPER_API_KEY", None)
             if serper_key is None:
                 serper_key = os.environ.get("SERPER_API_KEY")
@@ -90,30 +84,23 @@ async def launch(
                     "(the host env or --harness.env)"
                 )
             args += ["--search", f"--serper-key={serper_key}"]
-        if mcp_urls:
-            # The program connects to the tool servers over HTTP; hand it a standard
-            # `mcpServers` URL config (the `mcp` client itself comes from the uv deps).
-            args.append(
-                "--mcp-config="
-                + json.dumps(
-                    {
-                        "mcpServers": {
-                            name: {"url": url} for name, url in mcp_urls.items()
-                        }
-                    }
-                )
-            )
-        if isinstance(prompt, str):
-            args.append(f"--prompt={prompt}")
-        elif prompt is not None:
-            # Base64 images can exceed exec limits, so hand Messages off through a file.
-            path = f".vf-initial-messages-{trace.id}.json"
-            await runtime.write(
-                path,
-                json.dumps([message_to_wire(m) for m in prompt]).encode(),
-            )
-            args.append(f"--initial-messages-file={path}")
+        payload = {
+            "system_prompt": system_prompt,
+            "prompt": prompt if isinstance(prompt, str) else None,
+            "initial_messages": (
+                [message_to_wire(message) for message in prompt]
+                if prompt is not None and not isinstance(prompt, str)
+                else []
+            ),
+            "mcp_config": {
+                "mcpServers": {name: {"url": url} for name, url in mcp_urls.items()}
+            },
+        }
         program = await runtime.prepare_uv_script(
             PROGRAM_SOURCE, self.config.resolved_env
         )
-        return await runtime.run_program([*program, *args], env)
+        return await runtime.run_program(
+            [*program, *args],
+            env,
+            stdin=json.dumps(payload).encode("utf-8"),
+        )

verifiers/v1/harnesses/bash/program.py

@@ -8,6 +8,7 @@
 import asyncio
 import json
 import subprocess
+import sys
 from contextlib import AsyncExitStack, asynccontextmanager, suppress
 from pathlib import Path
 
@@ -136,7 +137,11 @@ def run_search(query: str, api_key: str, num_results: int = 5) -> str:
 def run_bash(command: str) -> str:
     try:
         result = subprocess.run(
-            ["bash", "-c", command], capture_output=True, text=True, timeout=3600
+            ["bash"],
+            input=command,
+            capture_output=True,
+            text=True,
+            timeout=3600,
         )
         return result.stdout + result.stderr
     except Exception as e:
@@ -300,7 +305,9 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--api-key", required=True)
     parser.add_argument("--model", required=True)
     parser.add_argument("--system-prompt", default="")
+    parser.add_argument("--payload-stdin", action="store_true")
     parser.add_argument("--prompt", default="")
+    parser.add_argument("--prompt-file", default="")
     parser.add_argument("--initial-messages-file", default="")
     parser.add_argument("--mcp-config", default="")
     parser.add_argument("--edit", action="store_true")
@@ -312,13 +319,26 @@ def parse_args() -> argparse.Namespace:
 async def main() -> None:
     args = parse_args()
     initial = []
+    prompt = args.prompt
+    system_prompt = args.system_prompt
+    mcp_config = args.mcp_config
+    if args.payload_stdin:
+        payload = json.load(sys.stdin)
+        system_prompt = payload.get("system_prompt") or ""
+        prompt = payload.get("prompt") or ""
+        initial = payload.get("initial_messages") or []
+        mcp_config = json.dumps(payload.get("mcp_config") or {})
+    if args.prompt_file:
+        path = Path(args.prompt_file)
+        prompt = path.read_text(encoding="utf-8")
+        path.unlink()
     if args.initial_messages_file:
         path = Path(args.initial_messages_file)
         payload = path.read_bytes()
         path.unlink()
         initial = json.loads(payload)
     client = AsyncOpenAI(base_url=args.base_url, api_key=args.api_key)
-    config = json.loads(args.mcp_config or "{}")
+    config = json.loads(mcp_config or "{}")
     tools = [BASH_TOOL]
     reserved = {"bash"}
     if args.edit:
@@ -333,15 +353,11 @@ async def main() -> None:
         else ([], {}, {})
     )
     tools += mcp_tools
-    messages = (
-        [{"role": "system", "content": args.system_prompt}]
-        if args.system_prompt
-        else []
-    )
+    messages = [{"role": "system", "content": system_prompt}] if system_prompt else []
     if initial:
         messages.extend(initial)
-    elif args.prompt:
-        messages.append({"role": "user", "content": args.prompt})
+    elif prompt:
+        messages.append({"role": "user", "content": prompt})
     while True:
         message = await chat(client, args.model, messages, tools)
         messages.append(message.model_dump(exclude_none=True))

verifiers/v1/harnesses/claude_code/harness.py

@@ -79,7 +79,9 @@ async def launch(
             ctx.model,
         ]
         if system_prompt:
-            argv += ["--append-system-prompt", system_prompt]
+            system_prompt_path = f".vf-claude-system-{trace.id}.txt"
+            await runtime.write(system_prompt_path, system_prompt.encode("utf-8"))
+            argv += ["--append-system-prompt-file", system_prompt_path]
         argv += [
             arg
             for tool in self.config.disabled_tools or []
@@ -97,6 +99,7 @@ async def launch(
             mcp_path,
             "--strict-mcp-config",
             "--",
-            instruction or "",
         ]
-        return await runtime.run_program(argv, env)
+        return await runtime.run_program(
+            argv, env, stdin=(instruction or "").encode("utf-8")
+        )

verifiers/v1/harnesses/codex/harness.py

@@ -119,6 +119,8 @@ async def launch(
                     image_args += ["-i", path]
                     image_index += 1
             prompt = "\n\n".join(texts)
+        if prompt is None:
+            raise ValueError("Codex requires a task prompt (it has no user simulator)")
         # codex authenticates to the interception server with the session secret (its provider
         # api key) and posts Responses calls to `{endpoint}/responses`.
         env = {**self.config.resolved_env, KEY_VAR: secret}
@@ -163,10 +165,10 @@ async def launch(
             *tool_config,
             *image_args,
             "--",
-            prompt,
+            "-",
         ]
         try:
-            return await runtime.run_program(argv, env)
+            return await runtime.run_program(argv, env, stdin=prompt.encode("utf-8"))
         finally:
             if image_args:
                 try:

verifiers/v1/harnesses/null/harness.py

@@ -38,33 +38,23 @@ async def launch(
             f"--base-url={endpoint}",
             f"--api-key={secret}",
             f"--model={ctx.model}",
+            "--payload-stdin",
         ]
-        if system_prompt:
-            args.append(f"--system-prompt={system_prompt}")
-        if mcp_urls:
-            # The program connects to the tool servers over HTTP; hand it a standard
-            # `mcpServers` URL config (the `mcp` client itself comes from the uv deps).
-            args.append(
-                "--mcp-config="
-                + json.dumps(
-                    {
-                        "mcpServers": {
-                            name: {"url": url} for name, url in mcp_urls.items()
-                        }
-                    }
-                )
-            )
-        if isinstance(prompt, str):
-            args.append(f"--prompt={prompt}")
-        elif prompt is not None:
-            # Base64 images can exceed exec limits, so hand Messages off through a file.
-            path = f".vf-initial-messages-{trace.id}.json"
-            await runtime.write(
-                path,
-                json.dumps([message_to_wire(m) for m in prompt]).encode(),
-            )
-            args.append(f"--initial-messages-file={path}")
+        payload = {
+            "system_prompt": system_prompt,
+            "prompt": prompt if isinstance(prompt, str) else None,
+            "initial_messages": (
+                [message_to_wire(message) for message in prompt]
+                if prompt is not None and not isinstance(prompt, str)
+                else []
+            ),
+            "mcp_config": {
+                "mcpServers": {name: {"url": url} for name, url in mcp_urls.items()}
+            },
+        }
         program = await runtime.prepare_uv_script(
             PROGRAM_SOURCE, self.config.resolved_env
         )
-        return await runtime.run_program([*program, *args], env)
+        return await runtime.run_program(
+            [*program, *args], env, stdin=json.dumps(payload).encode("utf-8")
+        )

verifiers/v1/harnesses/null/program.py

@@ -7,6 +7,7 @@
 import argparse
 import asyncio
 import json
+import sys
 from contextlib import AsyncExitStack, asynccontextmanager, suppress
 from pathlib import Path
 
@@ -144,6 +145,7 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--api-key", required=True)
     parser.add_argument("--model", required=True)
     parser.add_argument("--system-prompt", default="")
+    parser.add_argument("--payload-stdin", action="store_true")
     parser.add_argument("--prompt", default="")
     parser.add_argument("--initial-messages-file", default="")
     parser.add_argument("--mcp-config", default="")
@@ -153,28 +155,33 @@ def parse_args() -> argparse.Namespace:
 async def main() -> None:
     args = parse_args()
     initial = []
+    prompt = args.prompt
+    system_prompt = args.system_prompt
+    mcp_config = args.mcp_config
+    if args.payload_stdin:
+        payload = json.load(sys.stdin)
+        system_prompt = payload.get("system_prompt") or ""
+        prompt = payload.get("prompt") or ""
+        initial = payload.get("initial_messages") or []
+        mcp_config = json.dumps(payload.get("mcp_config") or {})
     if args.initial_messages_file:
         path = Path(args.initial_messages_file)
         payload = path.read_bytes()
         path.unlink()
         initial = json.loads(payload)
     client = AsyncOpenAI(base_url=args.base_url, api_key=args.api_key)
-    config = json.loads(args.mcp_config or "{}")
+    config = json.loads(mcp_config or "{}")
     if config.get("mcpServers"):
         # Bound only tool enumeration; each session is opened and closed within this task.
         async with asyncio.timeout(60):
             tools, dispatch, servers = await connect_mcp(config)
     else:
         tools, dispatch, servers = [], {}, {}
-    messages = (
-        [{"role": "system", "content": args.system_prompt}]
-        if args.system_prompt
-        else []
-    )
+    messages = [{"role": "system", "content": system_prompt}] if system_prompt else []
     if initial:
         messages.extend(initial)
-    elif args.prompt:
-        messages.append({"role": "user", "content": args.prompt})
+    elif prompt:
+        messages.append({"role": "user", "content": prompt})
     while True:
         message = await chat(client, args.model, messages, tools)
         messages.append(message.model_dump(exclude_none=True))

verifiers/v1/harnesses/pi/harness.py

@@ -234,7 +234,12 @@ async def launch(
             if self.config.disabled_tools
             else []
         )
-        system_args = ["--append-system-prompt", system_prompt] if system_prompt else []
+        system_args: list[str] = []
+        if system_prompt:
+            system_prompt_path = f"{agent_dir}/system-prompt.txt"
+            await runtime.write(system_prompt_path, system_prompt.encode("utf-8"))
+            # Pi interprets an existing path as the system-prompt file contents.
+            system_args = ["--append-system-prompt", system_prompt_path]
         argv = [
             "sh",
             "-c",

verifiers/v1/harnesses/terminus_2/harness.py

@@ -1,3 +1,4 @@
+import json
 import logging
 from pathlib import Path
 
@@ -58,13 +59,15 @@ async def launch(
             f"--base-url={endpoint}",
             f"--api-key={secret}",
             f"--model={ctx.model}",
-            f"--system-prompt={system_prompt or ''}",
-            f"--task={prompt}",
+            "--payload-stdin",
         ]
+        payload = json.dumps(
+            {"system_prompt": system_prompt or "", "task": prompt}
+        ).encode("utf-8")
         try:
             source = PROGRAM_SOURCE.replace("{version}", self.config.version)
             program = await runtime.prepare_uv_script(source, self.config.resolved_env)
-            return await runtime.run_program([*program, *args], env)
+            return await runtime.run_program([*program, *args], env, stdin=payload)
         finally:
             # Harbor normally destroys its whole sandbox; this adapter borrows the
             # Verifiers runtime, so clean up Terminus's detached tmux server ourselves.

verifiers/v1/harnesses/terminus_2/program.py

@@ -5,8 +5,10 @@
 
 import argparse
 import asyncio
+import json
 import os
 import subprocess
+import sys
 from pathlib import Path, PurePosixPath
 
 from harbor.agents.terminus_2 import Terminus2
@@ -29,8 +31,8 @@ async def exec(
     ) -> ExecResult:
         _ = user
         result = subprocess.run(
-            command,
-            shell=True,
+            ["sh"],
+            input=command,
             cwd=cwd,
             env={**os.environ, **(env or {})},
             capture_output=True,
@@ -50,13 +52,18 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--api-key", required=True)
     parser.add_argument("--model", required=True)
     parser.add_argument("--system-prompt", default="")
-    parser.add_argument("--task", required=True)
+    parser.add_argument("--task", default="")
+    parser.add_argument("--payload-stdin", action="store_true")
     return parser.parse_args()
 
 
 async def main() -> None:
     args = parse_args()
     model, system_prompt, task = args.model, args.system_prompt, args.task
+    if args.payload_stdin:
+        payload = json.load(sys.stdin)
+        system_prompt = payload.get("system_prompt") or ""
+        task = payload.get("task") or ""
     logs_dir = Path(os.environ["TMUX_TMPDIR"])
     logs_dir.mkdir(mode=0o700, exist_ok=True)
     EnvironmentPaths.agent_dir = PurePosixPath(logs_dir)