Policy

zeroth.governance.policy

Policy and capability enforcement primitives.

This package lets you define what agents are and aren't allowed to do. It provides models for capabilities (like network access or file writes), policy definitions that allow or deny those capabilities, and a guard that checks policies before a node runs. The Capability enum itself is authored graph vocabulary defined in :mod:zeroth.contracts.graph.models and republished here.

CapabilityDeniedError

CapabilityDeniedError(
    *,
    node_id: str,
    required: set[Capability],
    granted: set[Capability],
)

Bases: PermissionError

Raised when an operation requires capabilities the node was not granted.

Subclasses :class:PermissionError so callers that already treat tool / memory failures as recoverable (feeding the error back to the model) keep working, while capability denials remain distinguishable by type.

missing property

missing: set[Capability]

The capabilities that were required but not granted.

PolicyGuard

PolicyGuard(
    *,
    policy_registry: PolicyRegistry | None = None,
    capability_registry: CapabilityRegistry | None = None,
)

Checks whether a node is allowed to execute, given the active policies.

Before a node runs, the guard collects all policies from the graph and the node, then compares the node's required capabilities against what those policies allow or deny. The result is an EnforcementResult that says "allow" or "deny" (with a reason).

evaluate

evaluate(
    graph: Graph,
    node: Node,
    run: object,
    input_payload: Mapping[str, Any],
) -> EnforcementResult

Decide whether a node is allowed to run under the current policies.

Gathers policies from both the graph and the node, resolves the node's required capabilities, and returns an EnforcementResult indicating ALLOW or DENY along with the effective capabilities and constraints.

evaluate_run_admission

evaluate_run_admission(
    request: RunAdmissionRequest,
) -> RunAdmissionResult

Evaluate only the policies explicitly bound to a run request.

RunAdmissionRequest

Bases: Protocol

Structural contract consumed by run admission policy evaluation.

Capability

Bases: StrEnum

A specific permission that a node might need to do its job.

Each value represents one kind of action (like reading from the network or writing to the filesystem). Authored tool bindings declare the capabilities they require, and policies use these values to control what nodes are allowed to do; the policy engine republishes the enum from :mod:zeroth.governance.policy.models.

EnforcementResult

Bases: BaseModel

The result of checking policies before a node runs.

Contains the decision (allow/deny), an optional reason if denied, and the effective set of capabilities and constraints that apply to the node's execution.

PolicyDecision

Bases: StrEnum

The outcome of a policy evaluation: either allow or deny.

PolicyDefinition

Bases: BaseModel

A named set of rules that controls what capabilities are allowed or denied.

You create one of these to describe what a particular policy permits. For example, a "read-only" policy might allow NETWORK_READ but deny NETWORK_WRITE. It can also control which secrets are visible and how strict the sandbox should be.

RunAdmissionResult

Bases: BaseModel

Policy-only result for admitting a run-creating request.

CapabilityRegistry

CapabilityRegistry()

A lookup table that maps short string refs to Capability values.

Use this to register capabilities by name so that graphs and nodes can reference them without importing the Capability enum directly.

register

register(ref: str, capability: Capability) -> Capability

Store a capability under the given ref name and return it.

resolve

resolve(ref: str) -> Capability

Look up a capability by its ref name. Raises KeyError if not found.

PolicyRegistry

PolicyRegistry()

A lookup table that maps policy IDs to full PolicyDefinition objects.

Register policies here so the PolicyGuard can find them when it evaluates a graph or node.

register

register(policy: PolicyDefinition) -> PolicyDefinition

Store a policy definition, keyed by its policy_id, and return it.

resolve

resolve(ref: str) -> PolicyDefinition

Look up a policy by its ID. Raises KeyError if not found.

parse_effective_capabilities

parse_effective_capabilities(
    ctx: Mapping[str, Any] | None,
) -> set[Capability]

Read the effective (granted) capability set out of an enforcement context.

The orchestrator stores the policy guard's effective_capabilities on the run as a JSON list of capability value strings and threads it back through enforcement_context. This turns that list into a set[Capability].

Unmapped strings (values that do not correspond to a known :class:Capability) are dropped: an unrecognized grant can never satisfy a known required capability, so dropping it keeps the result fail-closed (it grants nothing) without raising on forward-compatible extra values.

Returns an empty set when the context is missing, has no effective_capabilities key, or lists nothing. Emptiness here means "nothing granted" (deny everything that needs a capability) — it does NOT mean "skip enforcement". The decision to skip is the caller's, signalled by passing None instead of calling this helper.

require_capabilities

require_capabilities(
    required: Iterable[Capability],
    effective: set[Capability] | None,
    *,
    node_id: str,
) -> None

Assert that effective covers every capability in required.

Fail-closed: effective of None is treated as the empty set, so any non-empty required denies. When required is empty the call is a no-op (an operation that needs no capability is always allowed).

Raises :class:CapabilityDeniedError listing the missing capabilities.

apply_secret_policy

apply_secret_policy(
    env: Mapping[str, str],
    *,
    allowed_secrets: list[str],
    secret_access_enabled: bool,
) -> dict[str, str]

Filter environment variables so only policy-approved secrets get through.

Returns a dictionary containing only the env vars whose names appear in allowed_secrets. If secret access is disabled or no secrets are allowed, returns an empty dictionary.

default_capability_registry

default_capability_registry() -> CapabilityRegistry

Build a CapabilityRegistry that resolves every capability by its value.

Each :class:Capability is registered under its own string value (e.g. "memory_read" -> Capability.MEMORY_READ). This is the ref scheme the served path expects: graphs declare capability_bindings=["memory_read", ...] and the wired-in :class:PolicyGuard resolves them without any app-specific registration. Apps that use a bespoke ref scheme (e.g. "capability://memory-read") build and inject their own registry instead.