Skip to content

Execution units

zeroth.integrations.execution

Public API for the execution_units package.

This module re-exports all the important classes, functions, and errors from the sub-modules so you can import them directly from zeroth.integrations.execution instead of reaching into individual files.

CommandRuntimeAdapter

CommandRuntimeAdapter(
    validator: ExecutableUnitValidator | None = None,
)

Bases: BaseRuntimeAdapter

Adapter for running command-line programs as governed tools.

Use this when your executable unit is a shell command or CLI program rather than a Python function.

PythonRuntimeAdapter

PythonRuntimeAdapter(
    validator: ExecutableUnitValidator | None = None,
)

Bases: BaseRuntimeAdapter

Adapter for running Python functions directly as governed tools.

Use this when your executable unit is a native Python callable (a function or method) rather than a command-line program.

RuntimeAdapter

Bases: Protocol[InputT, OutputT]

Interface that all runtime adapters must follow.

Any class that implements this protocol can convert a manifest into a runnable governed Tool. The supports method checks compatibility, and materialize does the actual conversion.

supports

supports(manifest: ExecutableUnitManifest) -> bool

Return True if this adapter can handle the given manifest.

materialize

materialize(
    manifest: ExecutableUnitManifest,
    *,
    input_model: type[InputT],
    output_model: type[OutputT],
    handler: PythonHandler | None = None,
) -> Tool[InputT, OutputT]

Convert a manifest into a ready-to-run governed Tool.

ResourceConstraints dataclass

ResourceConstraints(
    cpu_cores: float | None = None,
    memory_mb: int | None = None,
    disk_mb: int | None = None,
    max_processes: int | None = None,
    network_access: bool | None = None,
)

Runtime constraints that a hardened sandbox should enforce when possible.

requires_hard_isolation

requires_hard_isolation() -> bool

Return True when the request needs a hardened backend to be meaningful.

ManifestValidationError

ManifestValidationError(report: ManifestValidationReport)

Bases: ValueError

Raised when a manifest has errors that prevent it from being used.

Carries the full validation report so you can inspect exactly what went wrong (e.g., missing fields, invalid values).

FreeformPayload

Bases: BaseModel

Schema-less payload for inline units.

Node-level contracts (input_contract_ref/output_contract_ref) remain the typed boundary; the unit itself accepts whatever the graph routes to it.

AdmissionController dataclass

AdmissionController(
    *,
    allowed_runtimes: Iterable[str] | None = None,
    allowed_commands: Iterable[str] | None = None,
)

Check whether a manifest is trusted and allowed to run.

admit

admit(manifest: Any) -> AdmissionResult

Return the verdict for one manifest, naming the branch that decided it.

reason is a stable snake_case code on every branch, matching the trusted_digest the admitted branch already returned. The rejected manifest's own text (its runtime, its command, its unit id) stays out of it: the reason travels into an audit record, where a producer-authored string is content rather than decision metadata, and the caller's exception message still names the binding that was refused.

AdmissionResult dataclass

AdmissionResult(admitted: bool, reason: str, digest: str)

Outcome of admitting a manifest for execution.

reason is a stable snake_case code naming the branch that decided the verdict, never a rendered message: it is promoted into the audit trail as decision metadata, which is retained where free-form text is not.

ManifestIntegrityRecord dataclass

ManifestIntegrityRecord(
    digest: str,
    signed_at: datetime | None = None,
    signer: str | None = None,
)

Integrity metadata optionally attached to a manifest.

ExecutionIOError

Bases: ValueError

Base error for all input/output problems in executable units.

ExtractedOutput dataclass

ExtractedOutput(
    payload: Any,
    stdout: str = "",
    stderr: str = "",
    exit_code: int | None = None,
    output_file: Path | None = None,
)

Holds the parsed output from an executable unit after it finishes.

The payload field contains the actual structured data extracted from the output. The other fields preserve the raw stdout, stderr, exit code, and output file path for debugging.

InjectedInput dataclass

InjectedInput(
    stdin: str | None = None,
    argv: tuple[str, ...] = (),
    env: dict[str, str] = dict(),
    input_file: Path | None = None,
)

Holds the prepared input data in whatever form the executable unit needs.

Depending on the input mode, one or more of these fields will be populated: stdin for piped JSON, argv for CLI arguments, env for environment variables, or input_file for a JSON file on disk.

InputInjectionError

Bases: ExecutionIOError

Raised when we cannot convert the input data into the required format.

OutputConversionError

Bases: ExecutionIOError

Raised when the extracted output does not match the expected data model.

OutputExtractionError

Bases: ExecutionIOError

Raised when we cannot read or parse the output from an executable unit.

ArtifactSource

Bases: BaseModel

Points to where the executable unit's code or binary lives.

The kind says what type of artifact it is, and ref is a reference string (like a module path or command name).

AuditSettings

Bases: BaseModel

Controls what gets recorded for auditing when an executable unit runs.

You can toggle whether to capture stdout, stderr, inputs, and outputs.

BuildConfig

Bases: BaseModel

How to build a project-type executable unit before running it.

Includes the build command, any extra environment variables for the build, and an optional hash of the dependency lock file for caching.

CommandArtifactSource

Bases: ArtifactSource

Artifact source for a CLI command or executable binary.

DependencySpec

Bases: BaseModel

A single dependency that an executable unit needs (e.g., a Python package).

EntryPointType

Bases: StrEnum

What kind of thing gets executed: a Python function, a CLI command, or a project.

EnvironmentVariable

Bases: BaseModel

An environment variable to set when running the executable unit.

Can hold a plain value or a reference to a secret stored elsewhere.

ExecutionMode

Bases: StrEnum

How the executable unit was onboarded into the system.

NATIVE means it is a Python function, WRAPPED_COMMAND means it wraps a CLI tool, PROJECT means it is a full project with build steps, INLINE means the source was authored directly (e.g. in the Studio's code node) and always runs as a sandboxed subprocess, and REPOSITORY means the unit was declared by a governed GitHub repository's .zeroth.yaml and runs from a verified staged checkout.

InlineSourceArtifactSource

Bases: ArtifactSource

Artifact source whose code IS the artifact: authored source text.

ref carries the content digest (sha256:...) so the unit's identity is content-addressed — editing the source changes the identity, and a published graph pins its code cryptographically.

InlineUnitManifest

Bases: ExecutableUnitManifestBase

Manifest for source authored inline (the Studio code node).

The source text travels in the artifact source itself and is materialized into the sandbox working directory at run time. Inline units always run as sandboxed subprocesses — there is no in-process path for authored code.

InputMode

Bases: StrEnum

How data gets passed into an executable unit.

For example, JSON_STDIN pipes JSON to the process's standard input, CLI_ARGS passes data as command-line flags, and ENV_VARS sets environment variables.

NativeUnitManifest

Bases: ExecutableUnitManifestBase

Manifest for executable units that are plain Python functions.

Use this when you want to run a Python callable directly, without spawning a subprocess. Requires a callable_ref pointing to the function.

OutputMode

Bases: StrEnum

How data gets read back from an executable unit.

For example, JSON_STDOUT reads JSON from standard output, OUTPUT_FILE_JSON reads from a file, and EXIT_CODE_ONLY just captures the process exit code.

ProjectArchiveArtifactSource

Bases: ArtifactSource

Artifact source for a packaged project archive that needs building.

ProjectUnitManifest

Bases: ExecutableUnitManifestBase

Manifest for executable units that are full projects needing a build step.

Use this when the executable unit is a project archive that must be built (e.g., compiled or installed) before it can run.

PythonModuleArtifactSource

Bases: ArtifactSource

Artifact source for a Python module (e.g., 'mypackage.mymodule').

ResourceLimits

Bases: BaseModel

Limits on how many resources an executable unit can use.

These are hints for the sandbox, like max CPU cores, memory, timeout, number of processes, and whether network access is allowed.

RunConfig

Bases: BaseModel

How to actually run the executable unit.

Includes the command to execute, what directory to run it in, and any extra environment variables.

RuntimeLanguage

Bases: StrEnum

The runtime family used to execute the unit (Python, shell command, or project).

WrappedCommandUnitManifest

Bases: ExecutableUnitManifestBase

Manifest for executable units that wrap a CLI command or script.

Use this when you want to run an existing command-line tool as an executable unit. The system handles passing input and reading output.

ExecutableUnitAdmissionError

ExecutableUnitAdmissionError(
    message: str,
    *,
    audit_record: Mapping[str, Any] | None = None,
)

Bases: ExecutableUnitExecutionError

Raised when an executable unit fails admission control before execution.

ExecutableUnitBinding dataclass

ExecutableUnitBinding(
    manifest_ref: str,
    manifest: ExecutableUnitManifest,
    input_model: type[BaseModel],
    output_model: type[BaseModel],
    python_handler: PythonHandler | None = None,
    allowed_env_keys: tuple[
        str, ...
    ] = _DEFAULT_ALLOWED_ENV_KEYS,
    metadata: dict[str, Any] = dict(),
)

Links a manifest to its input/output models and optional Python handler.

This is what you register in the ExecutableUnitRegistry. It bundles together everything the runner needs to execute a particular unit.

ExecutableUnitError

Bases: RuntimeError

Base error for anything that goes wrong when running an executable unit.

ExecutableUnitExecutionError

Bases: ExecutableUnitError

Raised when an executable unit crashes or fails during build or run.

ExecutableUnitNotFoundError

Bases: ExecutableUnitError

Raised when you try to run a manifest ref that has not been registered.

ExecutableUnitRegistry

ExecutableUnitRegistry()

A lookup table that maps manifest ref strings to their bindings.

Register bindings here so the runner can find them by name later.

register

register(
    binding: ExecutableUnitBinding | str,
    manifest: ExecutableUnitManifest | None = None,
    *,
    input_model: type[BaseModel] | None = None,
    output_model: type[BaseModel] | None = None,
    handler: PythonHandler | None = None,
    allowed_env_keys: Sequence[
        str
    ] = _DEFAULT_ALLOWED_ENV_KEYS,
    metadata: Mapping[str, Any] | None = None,
) -> ExecutableUnitBinding

Add an executable unit binding to the registry.

You can pass a pre-built ExecutableUnitBinding, or pass a ref string along with a manifest and models to build one automatically.

get

get(manifest_ref: str) -> ExecutableUnitBinding

Look up a binding by its ref string. Raises if not found.

has

has(manifest_ref: str) -> bool

Return True if a binding with this ref string exists.

list

list() -> dict[str, ExecutableUnitBinding]

All registered bindings by manifest ref (shallow copy; used by /v1/manifests).

ExecutableUnitRunner

ExecutableUnitRunner(
    registry: ExecutableUnitRegistry | None = None,
    *,
    sandbox_manager: SandboxManager | None = None,
    python_adapter: PythonRuntimeAdapter | None = None,
    secret_resolver: SecretResolver | None = None,
    admission_controller: AdmissionController | None = None,
    project_materializer: ProjectMaterializer | None = None,
)

The main class that actually runs executable units.

It looks up bindings from the registry, validates input, runs the unit (either as a Python function or as a sandboxed subprocess), extracts the output, and returns a structured result.

run_manifest_ref async

run_manifest_ref(
    manifest_ref: str,
    payload: BaseModel | Mapping[str, Any],
    *,
    enforcement_context: Mapping[str, Any] | None = None,
    operation_identity: OperationIdentity | None = None,
) -> ExecutableUnitRunResult

Run an executable unit by looking it up in the registry by ref string.

declares_side_effect

declares_side_effect(manifest_ref: str) -> bool | None

Whether the registered manifest declares that it has side effects.

None means "unknown" -- an unregistered ref, or an inline unit whose source travels in the graph and has no manifest at all. Callers treat unknown as side-effecting: guarding a read-only unit merely writes a receipt nobody needs, whereas skipping the guard on a real side effect is the correctness hole this whole subsystem exists to close.

run async

run(
    manifest_ref: str,
    payload: BaseModel | Mapping[str, Any],
    *,
    enforcement_context: Mapping[str, Any] | None = None,
    operation_identity: OperationIdentity | None = None,
) -> ExecutableUnitRunResult

Shortcut for run_manifest_ref. Run a unit by its ref string.

run_binding async

run_binding(
    binding: ExecutableUnitBinding,
    payload: BaseModel | Mapping[str, Any],
    *,
    enforcement_context: Mapping[str, Any] | None = None,
    operation_identity: OperationIdentity | None = None,
    read_only_paths: Sequence[str] = (),
) -> ExecutableUnitRunResult

Run an executable unit from a binding directly.

Validates the input, then dispatches to either native Python execution or sandboxed subprocess execution depending on the manifest type. read_only_paths names sandbox-relative subtrees the backend should remount read-only (ZER-37); it defaults to empty -- the repository execution phase populates it -- and is ignored by native units, which run no sandbox at all.

run_inline_source async

run_inline_source(
    unit_id: str,
    source: str,
    payload: BaseModel | Mapping[str, Any],
    *,
    timeout_seconds: int | None = None,
    enforcement_context: Mapping[str, Any] | None = None,
    operation_identity: OperationIdentity | None = None,
) -> ExecutableUnitRunResult

Run inline source authored in a graph node, binding it on demand.

A Studio code node has no registry entry: its source travels in the graph, so the binding is synthesized here and runs through the same sandboxed subprocess path as a registered unit. The runtime drives this seam through its ExecutableUnitRunner protocol instead of importing the inline helpers directly.

ExecutableUnitRunResult dataclass

ExecutableUnitRunResult(
    manifest_ref: str,
    input_data: dict[str, Any],
    output_data: dict[str, Any],
    sandbox_result: SandboxExecutionResult | None = None,
    extracted_output: ExtractedOutput | None = None,
    audit_record: dict[str, Any] = dict(),
)

The result of running an executable unit.

Contains the input that was sent, the output that came back, and details about the sandbox execution and audit trail.

DockerSandboxConfig dataclass

DockerSandboxConfig(
    container_name: str = "zeroth-sandbox",
    docker_binary: str = "docker",
    workspace_root: str = "/tmp/zeroth-sandbox",
    hardened: bool = True,
    run_as_user: str | None = None,
    max_output_bytes: int = 1048576,
)

Settings for the Docker container used as a sandbox.

Includes the container name, the Docker binary path, and where files go inside the container.

EnvironmentCacheManager

EnvironmentCacheManager()

Stores prepared sandbox environments in memory so they can be reused.

This avoids rebuilding the same environment setup every time a unit runs with the same configuration.

get

get(cache_key: str) -> SandboxEnvironment | None

Look up a cached environment by its key. Returns None if not found.

put

put(
    cache_key: str,
    environment: Mapping[str, str],
    *,
    metadata: Mapping[str, Any] | None = None,
) -> SandboxEnvironment

Store an environment in the cache and return it.

resolve

resolve(
    cache_key: str,
    builder: Callable[
        [], Mapping[str, str] | SandboxEnvironment
    ],
    *,
    metadata: Mapping[str, Any] | None = None,
) -> SandboxEnvironment

Get from cache if available, otherwise build, cache, and return it.

snapshot

snapshot() -> dict[str, SandboxEnvironment]

Return a copy of all cached environments.

SandboxBackendMode

Bases: StrEnum

Where sandboxed commands actually run.

LOCAL runs directly on the host machine, DOCKER runs inside a container, and AUTO picks Docker if available, falling back to local.

SandboxBackendUnavailableError

Bases: RuntimeError

Raised when the requested backend (e.g., Docker) is not running or accessible.

SandboxConfig dataclass

SandboxConfig(
    backend: SandboxBackendMode = SandboxBackendMode.LOCAL,
    docker: DockerSandboxConfig = DockerSandboxConfig(),
    strictness_mode: SandboxStrictnessMode = SandboxStrictnessMode.STANDARD,
    allow_untrusted_local_development: bool = False,
    sidecar_url: str | None = None,
)

Top-level config that picks which backend to use and Docker settings.

SandboxEnvironment dataclass

SandboxEnvironment(
    cache_key: str,
    variables: dict[str, str],
    metadata: dict[str, Any] = dict(),
)

A snapshot of a prepared execution environment.

Contains the cache key (for looking it up later), the environment variables to use, and any extra metadata.

SandboxExecutionResult dataclass

SandboxExecutionResult(
    command: tuple[str, ...],
    returncode: int,
    stdout: str,
    stderr: str,
    workdir: str,
    environment: dict[str, str],
    timed_out: bool = False,
    duration_seconds: float | None = None,
    cache_key: str | None = None,
    backend: str = SandboxBackendMode.LOCAL.value,
    container_name: str | None = None,
    stdout_truncated: bool = False,
    stderr_truncated: bool = False,
)

Everything that came back from running a command in the sandbox.

Includes the command that was run, its exit code, stdout/stderr output, how long it took, and which backend was used.

SandboxManager

SandboxManager(
    *,
    base_env: Mapping[str, str] | None = None,
    cache_manager: EnvironmentCacheManager | None = None,
    config: SandboxConfig | None = None,
    command_runner: Callable[..., CompletedProcess[str]]
    | None = None,
    process_factory: Callable[..., Popen[bytes]]
    | None = None,
    container_inspector: Callable[[str], bool]
    | None = None,
    sidecar_client: Any | None = None,
)

Manages running commands in isolated sandbox environments.

Handles environment preparation, caching, and dispatching to either local subprocess execution or Docker container execution. This is the main entry point for sandboxed command execution.

cache_manager property

cache_manager: EnvironmentCacheManager

Access the environment cache manager for this sandbox.

prepare_environment

prepare_environment(
    *,
    allowed_env_keys: Sequence[str] | None = None,
    overlay: Mapping[str, str] | None = None,
    cache_identity: Mapping[str, Any]
    | Sequence[Any]
    | None = None,
    runtime: str = "local-subprocess",
    runtime_version: str | None = None,
    dependency_manifest: Mapping[str, Any]
    | Sequence[Any]
    | None = None,
    build_config: Mapping[str, Any]
    | Sequence[Any]
    | None = None,
    sandbox_policy: Mapping[str, Any]
    | Sequence[Any]
    | None = None,
) -> SandboxEnvironment

Build (or retrieve from cache) a sandbox environment for a unit.

Computes a cache key from the runtime and dependency info, then either returns a cached environment or builds a new one.

run

run(
    command: Sequence[str],
    *,
    input_text: str | None = None,
    timeout_seconds: float | None = None,
    allowed_env_keys: Sequence[str] | None = None,
    overlay_env: Mapping[str, str] | None = None,
    working_directory: str | Path | None = None,
    runtime_version: str | None = None,
    dependency_manifest: Mapping[str, Any]
    | Sequence[Any]
    | None = None,
    build_config: Mapping[str, Any]
    | Sequence[Any]
    | None = None,
    sandbox_policy: Mapping[str, Any]
    | Sequence[Any]
    | None = None,
    cache_identity: Mapping[str, Any]
    | Sequence[Any]
    | None = None,
    resource_constraints: ResourceConstraints | None = None,
) -> SandboxExecutionResult

Run a command in a sandboxed environment.

Prepares the environment, creates a temp directory, and dispatches to either local or Docker execution depending on the config.

SandboxPolicyViolationError

Bases: SandboxBackendUnavailableError

Raised when a required isolation or enforcement level cannot be satisfied.

SandboxStrictnessMode

Bases: StrEnum

How strongly the sandbox should insist on hardened isolation.

SandboxTimeoutError

SandboxTimeoutError(
    *,
    command: Sequence[str],
    timeout_seconds: float | None,
    stdout: str = "",
    stderr: str = "",
)

Bases: TimeoutError

Raised when a sandboxed process takes longer than its allowed timeout.

__str__

__str__() -> str

Format a readable error message showing the command and timeout.

ExecutableUnitValidator

Checks manifests for problems before they are used.

Validates common fields (ID, version, contracts) and then runs mode-specific checks depending on whether the manifest is native, wrapped command, project, or repository.

validate

validate(
    manifest: ExecutableUnitManifest,
) -> ManifestValidationReport

Validate a manifest and return a report of all issues found.

validate_or_raise

validate_or_raise(
    manifest: ExecutableUnitManifest,
) -> ManifestValidationReport

Validate a manifest and raise an error if there are any problems.

ValidationCode

Bases: StrEnum

Named codes for each type of validation problem.

Using codes instead of free-form strings makes it easy to check for specific problems programmatically.

build_docker_resource_flags

build_docker_resource_flags(
    constraints: ResourceConstraints | None,
) -> list[str]

Translate supported resource constraints into Docker CLI flags.

build_inline_binding

build_inline_binding(
    node_id: str,
    source: str,
    *,
    timeout_seconds: int | None = None,
) -> ExecutableUnitBinding

Binding for a code node, ready for ExecutableUnitRunner.run_binding.

The manifest ref embeds the node id and the source digest so audit records name both the step and the exact code that ran.

build_inline_manifest

build_inline_manifest(
    unit_id: str,
    source: str,
    *,
    timeout_seconds: int | None = None,
    input_contract_ref: str = "contract://inline-freeform",
    output_contract_ref: str = "contract://inline-freeform",
) -> InlineUnitManifest

Synthesize the manifest for one code node's source.

inline_source_digest

inline_source_digest(source: str) -> str

Content-addressed identity for authored source text.

compute_manifest_digest

compute_manifest_digest(manifest: Any) -> str

Compute a stable digest for a manifest, excluding embedded integrity metadata.

convert_output

convert_output(
    output_model: type[BaseModel],
    extracted: ExtractedOutput | Any,
) -> BaseModel

Validate extracted output data against a Pydantic model.

Takes the raw payload from extract_output and makes sure it matches the expected output schema. Returns a validated Pydantic model instance.

extract_output

extract_output(
    mode: OutputMode | str,
    *,
    stdout: str,
    stderr: str = "",
    exit_code: int | None = None,
    output_file_path: Path | None = None,
) -> ExtractedOutput

Parse raw output from an executable unit into structured data.

Supports multiple modes: reading JSON from stdout, finding a tagged JSON line in stdout, reading a JSON file, capturing plain text, or just using the exit code.

inject_input

inject_input(
    mode: InputMode | str,
    payload: BaseModel | Mapping[str, Any],
    *,
    input_file_path: Path | None = None,
    env_prefix: str = _INPUT_ENV_PREFIX,
) -> InjectedInput

Convert structured data into the format an executable unit expects.

For example, if the mode is "json_stdin", this serializes the payload as JSON and puts it in the stdin field. If the mode is "cli_args", it converts each key-value pair into --key value arguments.

validate_staged_manifest

validate_staged_manifest(
    document: RepoManifestDocument, staged_root: Path
) -> RepoManifestValidationReport

Check a parsed manifest's paths against the staged checkout it describes.

Parameters:

Name Type Description Default
document RepoManifestDocument

A document parse_manifest_document accepted.

required
staged_root Path

The directory the repository was staged into.

required

Returns:

Type Description
RepoManifestValidationReport

A report of every path that escapes the checkout, names the wrong kind

RepoManifestValidationReport

of filesystem object, or is missing.

build_sandbox_environment

build_sandbox_environment(
    base_env: Mapping[str, str] | None,
    *,
    allowed_env_keys: Sequence[str] | None = None,
    overlay: Mapping[str, str] | None = None,
) -> dict[str, str]

Build a restricted set of environment variables for sandbox execution.

Starts from the base environment, keeps only the allowed keys, then adds any overlay variables on top. This prevents leaking sensitive environment variables into sandboxed processes.

compute_environment_cache_key

compute_environment_cache_key(
    *,
    runtime: str,
    runtime_version: str | None = None,
    dependency_manifest: Mapping[str, Any]
    | Sequence[Any]
    | None = None,
    build_config: Mapping[str, Any]
    | Sequence[Any]
    | None = None,
    sandbox_policy: Mapping[str, Any]
    | Sequence[Any]
    | None = None,
    identity: Mapping[str, Any]
    | Sequence[Any]
    | None = None,
) -> str

Create a unique hash key for an execution environment setup.

Two environments with the same runtime, dependencies, build config, and policy will always produce the same key. This lets us cache and reuse environments instead of rebuilding them every time.

docker_container_running

docker_container_running(
    container_name: str,
    *,
    docker_binary: str = "docker",
    command_runner: Callable[..., CompletedProcess[str]]
    | None = None,
) -> bool

Check if a Docker container with the given name is currently running.

Returns False if Docker is not installed or the container does not exist.