Sandbox

Run commands and agent-written programs inside an operating-system boundary. See the guide.

class zett_agent.sandbox.base.Sandbox(policy=None, *, memory_strategy=MemoryStrategy.AUTO)

Run a command inside a boundary described by a SandboxPolicy.

A backend implements two things: check(), which says whether it can run on this machine, and _wrap(), which rewrites a command so the operating system enforces the policy on it. Everything else, including the environment, the time limit, the output cap, stopping the whole process tree and cancellation, is shared, so every backend behaves the same except in what it is able to confine.

The interface is deliberately split in two. wrap() returns the rewritten command without running it, for hosts that manage processes themselves; run() runs it and collects the result.

参数:

• policy (SandboxPolicy | None) -- The boundary to enforce, including its resource limits. Defaults to SandboxPolicy.read_only().

• memory_strategy (MemoryStrategy | str) -- How to enforce memory_bytes; see MemoryStrategy. The default takes the strongest mechanism the machine has.

示例

Run a program that may write one directory:

sandbox = create_sandbox(SandboxPolicy.workspace_write("/work/project"))
result = await sandbox.run(["python3", "build.py"], cwd="/work/project")
if not result.succeeded:
    print(result.stderr)

Run a shell pipeline:

result = await sandbox.run("grep -r TODO . | wc -l", cwd="/work/project")

备注

The sandbox confines a process and everything it starts. It does not decide whether a command should run at all; asking a person first is a separate control, for example ShellApprovalExtension.

警告

Several runs share one policy but no state: each starts a fresh process in a fresh boundary. Files a run writes inside a writable path stay there.

name: ClassVar[str]

Short stable name of the backend, reported in results.

isolated: ClassVar[bool]

Whether this backend enforces a boundary. False only for the backend that runs commands as they are.

abstractmethod check()

Raise SandboxUnavailableError unless this backend can run here.

The error says what is missing and how to fix it. Calling it is cheap after the first time.

返回类型:

None

capabilities()

Return how each limit of this sandbox's policy is enforced on this machine.

Nothing is created or changed, so it is safe to ask before running anything: a host can refuse a policy that this machine cannot enforce as written instead of discovering it from a run. The mechanisms named here are the ones a run will use.

示例

Refuse to run unless the machine enforces memory exactly:

report = sandbox.capabilities()
if report.memory not in ("cgroup", "rlimit_as"):
    raise RuntimeError(report.warnings)

备注

A report is about this machine, not about the policy alone: the same policy reports cgroup on a Linux host with delegated controllers and watchdog on macOS.

返回类型:

ResourceReport

wrap(command, *, cwd=None, env=None)

Return command rewritten to run inside this sandbox, without running it.

参数:

• command (str | Sequence[str]) -- A shell string, run through /bin/sh -c, or the program followed by its arguments, run directly.

• cwd (str | PathLike[str] | None) -- Working directory; the current directory by default.

• env (Mapping[str, str] | None) -- Variables added to the policy's inherited environment.

抛出:

• SandboxUnavailableError -- If the backend cannot run here.

• SandboxPolicyError -- If the command, directory or policy is invalid.

返回类型:

SandboxedCommand

async run(command, *, cwd=None, env=None, stdin=None, timeout=60.0, max_output_bytes=1048576)

Run command inside the sandbox and collect what it produced.

The process runs in its own process group. A timeout, a cancellation of the awaiting task, and the end of the run each stop the whole group, so no helper process outlives its command.

参数:

• command (str | Sequence[str]) -- A shell string or the program followed by its arguments.

• cwd (str | PathLike[str] | None) -- Working directory; the current directory by default.

• env (Mapping[str, str] | None) -- Variables added to the policy's inherited environment.

• stdin (str | bytes | None) -- Input to feed the process; it gets no input otherwise.

• timeout (float | None) -- Seconds before the run is killed; None waits without limit.

• max_output_bytes (int) -- Bytes kept of each of standard output and error.

抛出:

• SandboxUnavailableError -- If the backend cannot run here.

• SandboxPolicyError -- If an argument or the policy is invalid.

• SandboxError -- If the process could not be started.

• asyncio.CancelledError -- When the caller is cancelled, after the process group has been stopped.

返回类型:

SandboxResult