fix: improve sandbox example runtime validation

examples/sandbox/extensions/modal_runner.py

@@ -191,6 +191,17 @@ async def _verify_stop_resume(
                 SNAPSHOT_CHECK_PATH,
                 io.BytesIO(SNAPSHOT_CHECK_CONTENT.encode("utf-8")),
             )
+            if mount_check_path is not None:
+                await sandbox.write(
+                    mount_check_path,
+                    io.BytesIO(MOUNT_CHECK_CONTENT.encode("utf-8")),
+                )
+                mount_text = await _read_text(sandbox, mount_check_path)
+                if mount_text != MOUNT_CHECK_CONTENT:
+                    raise RuntimeError(
+                        "Native cloud bucket write verification failed before resume: "
+                        f"expected {MOUNT_CHECK_CONTENT!r}, got {mount_text!r}"
+                    )
             await sandbox.stop()
         finally:
             await sandbox.shutdown()
@@ -204,10 +215,31 @@ async def _verify_stop_resume(
                     f"Snapshot resume verification failed for {workspace_persistence!r}: "
                     f"expected {SNAPSHOT_CHECK_CONTENT!r}, got {restored_text!r}"
                 )
+            if mount_check_path is not None:
+                mount_text = await _read_text(resumed_sandbox, mount_check_path)
+                if mount_text != MOUNT_CHECK_CONTENT:
+                    raise RuntimeError(
+                        "Native cloud bucket resume verification failed: "
+                        f"expected {MOUNT_CHECK_CONTENT!r}, got {mount_text!r}"
+                    )
+                await resumed_sandbox.write(
+                    mount_check_path,
+                    io.BytesIO(MOUNT_CHECK_UPDATED_CONTENT.encode("utf-8")),
+                )
+                updated_mount_text = await _read_text(resumed_sandbox, mount_check_path)
+                if updated_mount_text != MOUNT_CHECK_UPDATED_CONTENT:
+                    raise RuntimeError(
+                        "Native cloud bucket update verification failed after resume: "
+                        f"expected {MOUNT_CHECK_UPDATED_CONTENT!r}, got {updated_mount_text!r}"
+                    )
+                await resumed_sandbox.rm(mount_check_path)
         finally:
             await resumed_sandbox.aclose()
 
-        print(f"native cloud bucket read/write ok ({mount_check_path})")
+        if mount_check_path is None:
+            print("native cloud bucket check skipped (no bucket configured)")
+        else:
+            print(f"native cloud bucket read/write and resume ok ({mount_check_path})")
     print(f"snapshot round-trip ok ({workspace_persistence})")
 
 

examples/sandbox/extensions/runloop/capabilities.py

@@ -482,16 +482,17 @@ async def _query_runloop_network_policy(
         limit=10,
     )
     for policy in policies:
-        if getattr(policy, "name", None) != name:
+        policy_id = cast(str | None, getattr(policy, "id", None))
+        if policy_id is None:
+            continue
+        info = cast(Any, await client.platform.network_policies.get(policy_id).get_info())
+        if getattr(info, "name", None) != name:
             continue
-        info = cast(
-            Any, await client.platform.network_policies.get(cast(str, policy.id)).get_info()
-        )
         return RunloopResourceQueryResult(
             resource_type="network_policy",
             name=name,
             found=True,
-            id=cast(str | None, getattr(policy, "id", None)),
+            id=cast(str | None, getattr(info, "id", None)) or policy_id,
             description=cast(str | None, getattr(info, "description", None)),
         )
 
@@ -634,9 +635,23 @@ async def _bootstrap_persistent_resources(
             policy_result.action = "reused"
             policy_result.found_before_bootstrap = True
             refreshed_policy = await _query_runloop_network_policy(client, name=network_policy_name)
+            if not refreshed_policy.found or refreshed_policy.id is None:
+                raise RuntimeError(
+                    "Runloop reported a network policy conflict, but the existing policy "
+                    f"{network_policy_name!r} could not be resolved."
+                ) from exc
             policy_result.id = refreshed_policy.id
         else:
             policy_result.id = cast(str | None, getattr(created_policy, "id", None))
+            if policy_result.id is None:
+                refreshed_policy = await _query_runloop_network_policy(
+                    client, name=network_policy_name
+                )
+                policy_result.id = refreshed_policy.id
+    if policy_result.id is None:
+        raise RuntimeError(
+            f"Runloop network policy {network_policy_name!r} did not resolve to an ID."
+        )
     print(
         "persistent network policy bootstrap:",
         policy_result.model_dump(mode="json"),

examples/sandbox/extensions/runloop/runner.py

@@ -16,6 +16,7 @@
 from agents import ModelSettings, Runner
 from agents.run import RunConfig
 from agents.sandbox import Manifest, SandboxAgent, SandboxRunConfig
+from agents.sandbox.snapshot import NoopSnapshotSpec
 
 if __package__ is None or __package__ == "":
     sys.path.insert(0, str(Path(__file__).resolve().parents[4]))
@@ -107,6 +108,7 @@ async def main(
                 pause_on_exit=pause_on_exit,
                 user_parameters=(RunloopUserParameters(username="root", uid=0) if root else None),
             ),
+            snapshot=NoopSnapshotSpec(),
         ),
         workflow_name="Runloop sandbox example",
     )

examples/sandbox/memory.py

@@ -74,7 +74,9 @@ def _build_agent(*, model: str, manifest: Manifest) -> SandboxAgent:
             "Answer questions about the sandbox workspace. Inspect files before answering, make "
             "minimal edits, and keep the response concise. "
             "Use the shell tool to inspect and validate the workspace. Use apply_patch for text "
-            "edits when it is the clearest option. Use a non-login POSIX shell for commands. "
+            "edits when it is the clearest option. Use a non-login POSIX shell for commands. Keep "
+            "searches inside the workspace, and use `grep` or `find .` instead of `rg` or "
+            "parent-directory searches. "
             "Make one focused pytest attempt; if the local sandbox blocks Python or toolchain "
             "access, report that validation was blocked and finish instead of retrying repeatedly. "
             "Do not invent files you did not read."

examples/sandbox/memory_multi_agent_multiturn.py

@@ -123,7 +123,9 @@ def _build_engineering_agent(*, model: str, manifest: Manifest) -> SandboxAgent:
         model=model,
         instructions=(
             "You are an engineer. Inspect files before editing, make minimal changes, and verify "
-            "with tests. Use a non-login POSIX shell for commands. Make one focused pytest attempt; "
+            "with tests. Use a non-login POSIX shell for commands. Keep searches inside the "
+            "workspace, and use `grep` or `find .` instead of `rg` or parent-directory searches. "
+            "Make one focused pytest attempt; "
             "if the local sandbox blocks Python or toolchain access, report that validation was "
             "blocked and finish instead of retrying repeatedly."
         ),

examples/tools/codex.py

@@ -64,9 +64,11 @@ async def on_codex_stream(payload: CodexToolStreamEvent) -> None:
     if isinstance(item, CommandExecutionItem):
         command = item.command
         output = item.aggregated_output
-        output_preview = output[-200:] if isinstance(output, str) else ""
+        output_tail = output[-200:] if isinstance(output, str) else ""
         status = item.status
-        log(f"codex command {event.type}: {command} | status={status} | output={output_preview}")
+        log(
+            f"codex command {event.type}: {command} | status={status} | output_tail={output_tail!r}"
+        )
         return
     if isinstance(item, McpToolCallItem):
         server = item.server

src/agents/sandbox/sandboxes/unix_local.py

@@ -624,7 +624,12 @@ def _workspace_relative_command_parts(
         return rewritten
 
     @staticmethod
-    def _darwin_allowable_read_roots(path: Path, *, host_home: Path) -> list[Path]:
+    def _darwin_allowable_read_roots(
+        path: Path,
+        *,
+        host_home: Path,
+        allow_virtual_environment_root: bool = False,
+    ) -> list[Path]:
         candidates: set[Path] = set()
         normalized = path.expanduser()
         try:
@@ -642,6 +647,14 @@ def _darwin_allowable_read_roots(path: Path, *, host_home: Path) -> list[Path]:
         else:
             candidates.add(resolved.parent)
 
+        if allow_virtual_environment_root:
+            for candidate in (normalized, resolved):
+                if candidate.name != "bin":
+                    continue
+                virtual_env_root = candidate.parent
+                if (virtual_env_root / "pyvenv.cfg").is_file():
+                    candidates.add(virtual_env_root)
+
         resolved_text = resolved.as_posix()
         if resolved_text == "/opt/homebrew" or resolved_text.startswith("/opt/homebrew/"):
             candidates.add(Path("/opt/homebrew"))
@@ -680,13 +693,21 @@ def _darwin_additional_read_paths(
         allowed: list[Path] = []
         seen: set[str] = set()
 
-        def _append(path: str | Path | None) -> None:
+        def _append(
+            path: str | Path | None,
+            *,
+            allow_virtual_environment_root: bool = False,
+        ) -> None:
             if path is None:
                 return
             candidate = Path(path).expanduser()
             if not candidate.is_absolute():
                 return
-            for root in self._darwin_allowable_read_roots(candidate, host_home=host_home):
+            for root in self._darwin_allowable_read_roots(
+                candidate,
+                host_home=host_home,
+                allow_virtual_environment_root=allow_virtual_environment_root,
+            ):
                 key = root.as_posix()
                 if key in seen:
                     continue
@@ -699,6 +720,12 @@ def _append(path: str | Path | None) -> None:
 
         executable = shutil.which(command_parts[0], path=env.get("PATH"))
         _append(executable)
+
+        # Only host-controlled PATH entries may widen a bin grant to its virtual environment
+        # root. Manifest environment overrides must not authorize broader host filesystem reads.
+        for path_entry in os.environ.get("PATH", "").split(os.pathsep):
+            if path_entry:
+                _append(path_entry, allow_virtual_environment_root=True)
         return allowed
 
     def _darwin_extra_path_grant_roots(self) -> list[tuple[Path, bool]]:

tests/extensions/sandbox/test_runloop_capabilities_example.py

@@ -89,6 +89,11 @@ async def get_info(self) -> object:
         )
 
 
+class _FakePolicyListItem:
+    def __init__(self, policy_id: str) -> None:
+        self.id = policy_id
+
+
 class _FakeNetworkPoliciesClient:
     def __init__(self) -> None:
         self.policies: dict[str, _FakePolicy] = {}
@@ -106,12 +111,12 @@ def add(self, name: str, description: str | None = None) -> _FakePolicy:
         self.policies[policy.id] = policy
         return policy
 
-    async def list(self, **params: object) -> list[_FakePolicy]:
+    async def list(self, **params: object) -> list[_FakePolicyListItem]:
         name = params.get("name")
         policies = list(self.policies.values())
         if isinstance(name, str):
-            return [policy for policy in policies if policy.name == name]
-        return policies
+            policies = [policy for policy in policies if policy.name == name]
+        return [_FakePolicyListItem(policy.id) for policy in policies]
 
     async def create(self, **params: object) -> _FakePolicy:
         self.create_calls.append(dict(params))

tests/sandbox/test_runtime.py

@@ -4033,6 +4033,115 @@ def _fake_which(name: str, path: str | None = None) -> str | None:
     assert '(allow file-write* (subpath "/opt/homebrew"))' not in profile
 
 
+def test_unix_local_confined_exec_command_allows_python_virtual_environment_root(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    workspace_root = tmp_path / "workspace"
+    workspace_root.mkdir()
+    host_project_root = tmp_path / "host-project"
+    virtual_env_root = host_project_root / ".venv"
+    virtual_env_bin = virtual_env_root / "bin"
+    virtual_env_bin.mkdir(parents=True)
+    (virtual_env_root / "pyvenv.cfg").write_text("home = /usr/bin\n", encoding="utf-8")
+    python_executable = virtual_env_bin / "python"
+    python_executable.write_text("", encoding="utf-8")
+    session = UnixLocalSandboxSession.from_state(
+        UnixLocalSandboxSessionState(
+            session_id=uuid.uuid4(),
+            manifest=_unix_local_manifest(root=str(workspace_root)),
+            snapshot=NoopSnapshot(id="darwin-virtual-environment"),
+            workspace_root_owned=False,
+        )
+    )
+    unix_local = cast(Any, unix_local_module)
+    path_env = str(virtual_env_bin)
+
+    def _fake_which(name: str, path: str | None = None) -> str | None:
+        if name == "sandbox-exec":
+            return "/usr/bin/sandbox-exec"
+        if name == "python":
+            assert path == path_env
+            return str(python_executable)
+        return None
+
+    monkeypatch.setattr(unix_local.sys, "platform", "darwin")
+    monkeypatch.setattr(unix_local.shutil, "which", _fake_which)
+    monkeypatch.setenv("PATH", path_env)
+
+    command = session._confined_exec_command(
+        command_parts=["python", "-V"],
+        workspace_root=workspace_root,
+        env={"PATH": path_env},
+    )
+    profile_lines = set(command[2].splitlines())
+
+    assert (
+        f'(allow file-read-data file-read-metadata (subpath "{virtual_env_root}"))' in profile_lines
+    )
+    assert (
+        f'(allow file-read-data file-read-metadata (subpath "{host_project_root}"))'
+        not in profile_lines
+    )
+    assert f'(allow file-write* (subpath "{virtual_env_root}"))' not in profile_lines
+
+
+def test_unix_local_confined_exec_command_does_not_expand_manifest_virtual_environment(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    workspace_root = tmp_path / "workspace"
+    workspace_root.mkdir()
+    host_project_root = tmp_path / "host-project"
+    virtual_env_root = host_project_root / ".venv"
+    virtual_env_bin = virtual_env_root / "bin"
+    virtual_env_bin.mkdir(parents=True)
+    (virtual_env_root / "pyvenv.cfg").write_text("home = /usr/bin\n", encoding="utf-8")
+    python_executable = virtual_env_bin / "python"
+    python_executable.write_text("", encoding="utf-8")
+    session = UnixLocalSandboxSession.from_state(
+        UnixLocalSandboxSessionState(
+            session_id=uuid.uuid4(),
+            manifest=_unix_local_manifest(root=str(workspace_root)),
+            snapshot=NoopSnapshot(id="darwin-manifest-virtual-environment"),
+            workspace_root_owned=False,
+        )
+    )
+    unix_local = cast(Any, unix_local_module)
+    manifest_path = str(virtual_env_bin)
+
+    def _fake_which(name: str, path: str | None = None) -> str | None:
+        if name == "sandbox-exec":
+            return "/usr/bin/sandbox-exec"
+        if name == "python":
+            assert path == manifest_path
+            return str(python_executable)
+        return None
+
+    monkeypatch.setattr(unix_local.sys, "platform", "darwin")
+    monkeypatch.setattr(unix_local.shutil, "which", _fake_which)
+    monkeypatch.setenv("PATH", "/usr/bin:/bin")
+
+    command = session._confined_exec_command(
+        command_parts=["python", "-V"],
+        workspace_root=workspace_root,
+        env={"PATH": manifest_path},
+    )
+    profile_lines = set(command[2].splitlines())
+
+    assert (
+        f'(allow file-read-data file-read-metadata (subpath "{virtual_env_bin}"))' in profile_lines
+    )
+    assert (
+        f'(allow file-read-data file-read-metadata (subpath "{virtual_env_root}"))'
+        not in profile_lines
+    )
+    assert (
+        f'(allow file-read-data file-read-metadata (subpath "{host_project_root}"))'
+        not in profile_lines
+    )
+
+
 def test_unix_local_darwin_exec_profile_allows_extra_path_grants(tmp_path: Path) -> None:
     workspace_root = tmp_path / "workspace"
     read_write_root = tmp_path / "read-write"