Skip to content

kyvvu_engine Python API

kyvvu-engine — standalone policy evaluation engine for AI governance.

The public Python interface is exactly kyvvu_engine.__all__. Import its symbols from the top level, for example::

from kyvvu_engine import PolicyEngine, Behavior, EvalContext

Submodules are internal implementation details. Behaviour templates are not part of the engine: they are the kyvvu-templates distribution, imported as kyvvu_templates.

Interface stability

See the stability policy for public interface versioning and the policy rule contract.

kyvvu_engine.AAG_TASK_START_VERSION

AAG_TASK_START_VERSION = '0.7.0'

kyvvu_engine.AUTO_LOG_LOCATION

AUTO_LOG_LOCATION = 'auto'

kyvvu_engine.Action

Decision emitted by the engine after evaluating all applicable policies.

Values

allow: No violations; the intended step may proceed. warn: One or more non-critical violations; caller may proceed with caution. block: One or more critical violations; the intended step should be prevented.

Attributes:

kyvvu_engine.Action.allow

allow = 'allow'

kyvvu_engine.Action.block

block = 'block'

kyvvu_engine.Action.warn

warn = 'warn'

kyvvu_engine.ActionKind

The 12 atomic behavior types accepted by the engine.

Two namespaces. The four task.* types mark task boundaries; the eight step.* types describe agent behavior within a task. The split names a real asymmetry rather than a routing difference: a boundary marker carries no verb and neither input nor output, while a behavioral step carries the full {properties, input, output} triplet, which is what makes an absent output readable as an intended action.

Both are emitted on one path and evaluated at step_execution. TASK_BOUNDARY_KINDS names the four for callers that need to tell them apart — prefer it over matching on the task. prefix.

Attributes:

kyvvu_engine.ActionKind.step_credential

step_credential = 'step.credential'

kyvvu_engine.ActionKind.step_exec

step_exec = 'step.exec'

kyvvu_engine.ActionKind.step_gate

step_gate = 'step.gate'

kyvvu_engine.ActionKind.step_message

step_message = 'step.message'

kyvvu_engine.ActionKind.step_model

step_model = 'step.model'

kyvvu_engine.ActionKind.step_resource

step_resource = 'step.resource'

kyvvu_engine.ActionKind.step_self

step_self = 'step.self'

kyvvu_engine.ActionKind.step_unknown

step_unknown = 'step.unknown'

kyvvu_engine.ActionKind.task_end

task_end = 'task.end'

kyvvu_engine.ActionKind.task_error

task_error = 'task.error'

kyvvu_engine.ActionKind.task_idle

task_idle = 'task.idle'

kyvvu_engine.ActionKind.task_start

task_start = 'task.start'

kyvvu_engine.AgentBindingError

Raised when a caller uses an engine or runner bound to another agent.

kyvvu_engine.AgentProfile

Caller supplied agent registration details, without runtime settings.

Functions:

Attributes:

kyvvu_engine.AgentProfile.declared_tools

declared_tools: list[str] | None = None

kyvvu_engine.AgentProfile.default_maintainer

default_maintainer() -> AgentProfile

Use the owner as maintainer when no maintainer was supplied.

kyvvu_engine.AgentProfile.description

description: str | None = None

kyvvu_engine.AgentProfile.maintainer_id

maintainer_id: str | None = None

kyvvu_engine.AgentProfile.metadata

metadata: dict[str, Any] = Field(default_factory=dict)

kyvvu_engine.AgentProfile.name

name: str | None = None

kyvvu_engine.AgentProfile.owner_id

owner_id: str | None = None

kyvvu_engine.AgentProfile.purpose

purpose: str | None = None

kyvvu_engine.AgentProfile.store_metadata

store_metadata: bool = False

kyvvu_engine.AgentProfile.validate_metadata

validate_metadata(value: Any) -> dict[str, Any]

Reject metadata that would change in an I-JSON round trip.

kyvvu_engine.Behavior

A single agent step, used for both intended and completed steps.

Intended steps (pre-execution) have output=None and seq=None. Completed steps (post-execution) have output populated and a seq assigned when the step is recorded.

input, output, properties and meta accept only I-JSON (RFC 7493) values, recursively: dict with str keys, list, str, bool, None, int with magnitude at most 2**53 - 1, and finite float. Anything else -- NaN, a larger integer, a tuple, set, bytes, datetime or Decimal -- is refused with a ValueError naming its path (input.items[2]: float NaN is not I-JSON ...), because it could not be exported, hashed or audited without being altered.

Identifiers are refused with a ValueError naming the field and the rule (task_id 'run/42': use letters, digits and . _ : - only (1-100 characters)), because from 1.1 they are hashed, signed and used in lock and storage keys.

Fields

agent_id: Registered agent identifier. Letters, digits and . _ : -, 0-255 characters (empty for an offline, unregistered runner). task_id: Current task identifier. The engine uses this to key its internal history tracker. Letters, digits and . _ : -, 1-100 characters. A task.start may carry meta["parent_task_id"], which follows the same rule. action_id: Correlates a step outcome with its evaluation. Optional on intent and may be absent on task boundary markers. Unique per (agent_id, task_id): the one legal reuse is pairing an intended step with its own completion. A duplicate is refused with ValueError. Letters, digits and . _ : -, 1-255 characters. aag_version: The AAG vocabulary version, on task.start only. Letters, digits and . + -, 1-64 characters. timestamp: When the agent initiated this step, recorded by the client. Defaults to the UTC clock at construction, so build an intended step when the intent is formed. It is also the time time-based rules (such as working_hours_only) evaluate against: a preflight is judged at the time the intent was formed, not when the action later executes. KyvvuRunner.evaluate() refuses one further than max_clock_skew_seconds from the host clock; PolicyEngine and KyvvuRunner.record() do not, so recorded history replays as is. seq: Orders a task's recorded steps: strictly increasing per (agent_id, task_id), assigned by the recording client. Consumers must not assume it is contiguous, starts at a particular value, or reflects wall-clock order; a value supplied by the caller may be ignored. None for an intended step. kind: The atomic behavior category. Set by the template. verb: Direction/mutation intent. None for step types that carry no verb (step.exec, step.gate, step.unknown, and the four task.* boundary types). step_name: Human-readable label for this step (e.g. "chat_gpt-4o"). Any I-JSON text of 1-255 characters with no control characters (U+0000-U+001F, U+007F-U+009F). input: Input data for this step. Visible to rule functions. output: Output data. Populated for completed steps; None for intended steps. A completed step.gate carries decision (pass or deny). properties: Nested key-value object carrying contextual information. Standard groups: target, auth, data, model, exec, idle, self, message, transport, raw. A step.gate carries bare kind (human or guardrail). Custom groups are allowed and passed through unchanged. meta: Correlation and provenance fields outside the behavior itself. Keys starting with kyvvu. are reserved for what the engine and the Python client write (kyvvu.eval, kyvvu.manifests); the runner refuses a caller-supplied kyvvu.* meta key in evaluate() and record().

Functions:

Attributes:

kyvvu_engine.Behavior.aag_version

aag_version: str | None = None

kyvvu_engine.Behavior.action_id

action_id: str | None = None

kyvvu_engine.Behavior.agent_id

agent_id: str

kyvvu_engine.Behavior.input

input: dict[str, Any] | None = None

kyvvu_engine.Behavior.kind

kind: ActionKind

kyvvu_engine.Behavior.meta

meta: dict[str, Any] | None = None

kyvvu_engine.Behavior.model_config

model_config = ConfigDict(extra='forbid')

kyvvu_engine.Behavior.output

output: dict[str, Any] | None = None

kyvvu_engine.Behavior.properties

properties: dict[str, Any] = Field(default_factory=dict)

kyvvu_engine.Behavior.seq

seq: int | None = None

kyvvu_engine.Behavior.step_name

step_name: str

kyvvu_engine.Behavior.task_id

task_id: str

kyvvu_engine.Behavior.timestamp

timestamp: AwareDatetime = Field(default_factory=lambda: datetime.now(timezone.utc))

kyvvu_engine.Behavior.verb

verb: Verb | None = None

kyvvu_engine.Behavior.with_checked_payloads

with_checked_payloads() -> Behavior

Re-check the payload fields of a behavior built without validation.

model_copy(update=...) does not run validators, so a payload set that way (the documented intended.model_copy(update={"output": ...}) pattern, or an engine-injected meta entry) has not been checked. Recording calls this so that nothing non-I-JSON reaches the audit trail by that route.

Returns:

  • Behavior – A copy whose payloads are plain I-JSON data.

The timestamp is re-checked too: a naive datetime set that way would otherwise be exported in the host's local time.

Raises:

  • ValueError – If a payload value is not I-JSON, an identifier is outside its grammar, or the timestamp has no UTC offset.

kyvvu_engine.ChainHead

ChainHead(*, seq: int = 0, hash: str | None = None) -> None

The last link of one (agent_id, task_id) chain.

Attributes:

  • seq (int) – seq of the last chained step; 0 before the first.
  • hash (str | None) – That step's hash; None before the first.

kyvvu_engine.ChainHead.hash

hash: str | None = None

kyvvu_engine.ChainHead.seq

seq: int = 0

kyvvu_engine.ClientUpgradeInfo

ClientUpgradeInfo(code: str, detail: str = '', min_engine_version: str | None = None, min_policy_schema: int | None = None, upgrade: str = 'pip install -U kyvvu') -> None

What the platform said when it refused this client.

Attributes:

  • code (str) – client_upgrade_required or unsupported_wire_version.
  • detail (str) – The platform's own sentence.
  • min_engine_version (str | None) – The oldest engine the platform serves, if stated.
  • min_policy_schema (int | None) – The oldest policy schema it serves, if stated.
  • upgrade (str) – The command that fixes it, as the platform phrased it.

Functions:

  • as_dict – A JSON-ready copy, for a status report.

kyvvu_engine.ClientUpgradeInfo.as_dict

as_dict() -> dict[str, Any]

A JSON-ready copy, for a status report.

kyvvu_engine.ClientUpgradeInfo.code

code: str

kyvvu_engine.ClientUpgradeInfo.detail

detail: str = ''

kyvvu_engine.ClientUpgradeInfo.min_engine_version

min_engine_version: str | None = None

kyvvu_engine.ClientUpgradeInfo.min_policy_schema

min_policy_schema: int | None = None

kyvvu_engine.ClientUpgradeInfo.upgrade

upgrade: str = 'pip install -U kyvvu'

kyvvu_engine.EvalContext

Agent identity and execution context passed to every evaluate() call.

All fields are pre-fetched by the caller (API layer or Python client). The engine performs no I/O to retrieve these values.

Fields

agent_id: Stable string identifier for the agent (not the DB integer PK). task_id: Canonical identifier for a single agent task run. Groups all the steps recorded for one logical task. environment: Deployment environment: "development", "staging", or "production". risk_classification: EU AI Act risk tier: "high", "limited", "minimal", "unacceptable", or "unclassified". Used by classification-aware rules. user_id: Integer ID of the authenticated user who owns the agent's API key. agent_name, agent_purpose, agent_owner, agent_allowed_tools: Agent metadata pre-fetched by the API layer; forwarded to rule functions via RuleContext. manifests: Provenance records for the manifests whose policies are in effect for this agent. None when no manifest-sourced policies apply. project_root: Absolute path of the project/session working directory, when known. Supplied by integrations that run inside a project tree (e.g. kyvvu-claude populates it from the session cwd). Consumed by allowlist-containment rules such as path_within_root to block any write target that does not resolve under this root. None when the integration has no notion of a project root.

There is deliberately no evaluation-time field. Time-based rules evaluate against the behaviour's own timestamp, recorded by the client when the intent was formed, so the time a policy was checked against is the time the trace carries. The former now field is refused like any other unknown key.

Attributes:

kyvvu_engine.EvalContext.agent_allowed_tools

agent_allowed_tools: list[str] | None = None

kyvvu_engine.EvalContext.agent_id

agent_id: str

kyvvu_engine.EvalContext.agent_name

agent_name: str | None = None

kyvvu_engine.EvalContext.agent_owner

agent_owner: str | None = None

kyvvu_engine.EvalContext.agent_purpose

agent_purpose: str | None = None

kyvvu_engine.EvalContext.environment

environment: str = 'development'

kyvvu_engine.EvalContext.manifests

manifests: list[ManifestRef] | None = None

kyvvu_engine.EvalContext.model_config

model_config = ConfigDict(extra='forbid')

kyvvu_engine.EvalContext.project_root

project_root: str | None = None

kyvvu_engine.EvalContext.risk_classification

risk_classification: str | None = None

kyvvu_engine.EvalContext.task_id

task_id: str | None = None

kyvvu_engine.EvalContext.user_id

user_id: int | None = None

kyvvu_engine.EvalResult

Aggregate result returned by PolicyEngine.evaluate().

Fields

risk_score: Normalised violation severity in [0.0, 1.0]. 0.0 means no violations; 1.0 means maximum criticality. action: Recommended action derived from risk_score: allow | warn | block. policies: One PolicyResult per evaluated policy, in evaluation order. action_id: The action_id a KyvvuRunner evaluation used or generated for a step.* behaviour; None for a task boundary and for a pure PolicyEngine evaluation. recorded_seq: Set when the engine itself recorded the evaluated step: a blocked step.* behaviour is recorded by KyvvuRunner.evaluate() and PolicyEngine.evaluate_and_record() (decision D1), and this is the seq it was given. None when nothing was recorded. A caller must not record a step whose result carries a recorded_seq.

Consumers must ignore fields they do not know: later 1.x releases may add optional fields to this model.

Attributes:

kyvvu_engine.EvalResult.action

action: Action = Action.allow

kyvvu_engine.EvalResult.action_id

action_id: str | None = None

kyvvu_engine.EvalResult.policies

policies: list[PolicyResult] = Field(default_factory=list)

kyvvu_engine.EvalResult.recorded_seq

recorded_seq: int | None = None

kyvvu_engine.EvalResult.risk_score

risk_score: float = 0.0

kyvvu_engine.HttpExporter

HttpExporter(url: str, client: PlatformClient | None, *, generic: bool = False) -> None

POST steps in Kyvvu batch format to an HTTP endpoint.

Functions:

  • chain_head – Return the chain head after the last batch the API accepted.
  • export_task – POST the task's steps in Kyvvu batch format.
  • shutdown – No-op — each export opens and closes its own connection.

Configure the HTTP batch sink.

Parameters:

  • url (str) – Full URL of the batch-ingest endpoint.
  • client (PlatformClient | None) – Platform connection used to send batches.
  • generic (bool) – Send to url as a generic JSON endpoint.

kyvvu_engine.HttpExporter.chain_head

chain_head(agent_id: str, task_id: str) -> ChainHead | None

Return the chain head after the last batch the API accepted.

None when no batch of this task has been accepted by this exporter (or its head was forgotten, see MAX_CHAIN_HEADS). A caller that outlives the exporter (kyvvu-claude, a process per hook) persists this and passes it back as chain_from.

kyvvu_engine.HttpExporter.export_task

export_task(agent_id: str, task_id: str, steps: list[dict[str, Any]], *, idempotency_key: str | None = None, chain_from: ChainHead | None = None, chained: bool = True) -> bool

POST the task's steps in Kyvvu batch format.

More than MAX_BATCH_STEPS steps are sent as several batches of at most that many, in order, stopping at the first that fails. The runner's single report worker calls this, so batches of one task are never interleaved.

Each step is sent with seq and hash (the v1 hash chain), computed here over the steps exactly as sent by chain_steps: seq continues the task's chain head contiguously, whatever the steps carried. The chain starts at chain_from, else at the head this exporter remembers for the task, else at the task's first step. With no head and a first step whose seq is above 1 (a continuation exported elsewhere before), the export is sent unchained, as if chained=False. The head advances only when the API acknowledges a batch, so a batch that failed is chained identically when it is sent again. The API's hash_tail must equal the head; a difference is logged as a warning.

Sends User-Agent: kyvvu-engine/<version>, X-Kyvvu-Wire-Version: 1 (wire format 1: the unsigned batch) and the client headers of client_headers. A version refusal from the platform logs the upgrade warning.

Network and HTTP errors are logged and swallowed (a flush failure must not crash the host agent); an auth failure (401/403) additionally prints a one-time stderr hint.

Parameters:

  • agent_id (str) – Agent identifier.
  • task_id (str) – Task identifier.
  • steps (list[dict[str, Any]]) – List of step dicts to send.
  • idempotency_key (str | None) – Optional key for retrying this batch. When the steps are split, batch n (from 1) uses "<key>-<n>".
  • chain_from (ChainHead | None) – The task's chain head when this export began, for a caller that keeps it outside this exporter.
  • chained (bool) – False sends the steps as given, without hash: the API then numbers and hashes them itself, as before. For a caller that has lost its chain head mid-task.

Returns:

  • bool – True when the API accepted every batch, False otherwise. The
  • bool – HTTP and network errors below are caught rather than raised -- a
  • bool – telemetry sink must not take its caller down -- which used to make
  • bool – a dropped batch indistinguishable from an accepted one.

kyvvu_engine.HttpExporter.shutdown

shutdown() -> None

No-op — each export opens and closes its own connection.

kyvvu_engine.KyvvuBlockedError

KyvvuBlockedError(result: EvalResult, step_name: str = '') -> None

Raised by KyvvuRunner.evaluate() when the engine returns Action.block.

For a step.* behaviour the runner has already recorded the blocked step when this is raised (decision D1): result.recorded_seq is its seq and result.action_id its action_id. The caller must not record it again; a second record() of that action_id is refused as a duplicate. recorded_seq is None for a blocked task boundary, which is not recorded.

Attributes:

Retain the denying evaluation and format its user-facing step message.

kyvvu_engine.KyvvuBlockedError.result

result = result

kyvvu_engine.KyvvuBlockedError.step_name

step_name = step_name

kyvvu_engine.KyvvuClientUpgradeRequired

KyvvuClientUpgradeRequired(message: str, info: ClientUpgradeInfo) -> None

The platform refused this client as too old to serve.

Raised when the policy fetch (or registration) answers 426 with code: client_upgrade_required. A PolicyLoadError, because it is one: the platform's current set cannot be loaded by this client. That puts it under the single policy-load tolerance point -- a runner that already holds a validated set (in memory, or in its on-disk cache) keeps enforcing it, reported as stale_cache with stale_cause=client_upgrade_required; only a cold start with nothing to fall back on reaches the caller.

The message is the one line a person needs, never a list of param errors.

Attributes:

Attach the platform's compatibility requirements to the refusal.

kyvvu_engine.KyvvuClientUpgradeRequired.info

info = info

kyvvu_engine.KyvvuConfigError

Raised when KyvvuRunner is misconfigured (e.g. missing API key when needed).

kyvvu_engine.KyvvuRunner

KyvvuRunner(settings: KyvvuSettings | None = None, *, platform: PlatformClient | None = None, **kwargs: Any) -> None

Add policy fetch, batch logging, and incident webhooks to the pure engine.

Wraps PolicyEngine.

One runner governs one agent: its first evaluation or record binds the agent ID, and another agent's calls raise KyvvuConfigError.

Quickstart::

# From kwargs (test / script)
runner = KyvvuRunner(api_url="http://localhost:8000", api_key="KvKey-...")

# From a shared settings object
settings = KyvvuSettings(api_url="...", api_key="...")
runner = KyvvuRunner(settings=settings)

# Pure env-var driven (production containers)
runner = KyvvuRunner()

Important: production deployments must call sweep_stale_tasks periodically (e.g. every 5 minutes from a background thread) to evict abandoned task buffers. Tasks that crash before end_task() is called will otherwise accumulate in memory indefinitely.

Parameters:

  • settings (KyvvuSettings | None) – Pre-built KyvvuSettings object. When provided, keyword overrides are merged on top.
  • **kwargs (Any) – Any KyvvuSettings field name may be passed to override individual settings (e.g. api_url, api_key, agent_key).

Raises:

  • KyvvuConfigError – If the resolved settings name no policy source at all (see policy_source_kind), or if an unknown keyword argument is passed. A runner that can never load a policy would enforce nothing, and that is a configuration mistake worth reporting where the configuration is supplied rather than on the first governed step.

A runner needs no credentials when its policies are local::

from kyvvu_baseline import load_baseline_policies

runner = KyvvuRunner(
    policies=load_baseline_policies(),   # or policy_file="my.yaml"
    log_location="none",
)

Functions:

  • abandon_open_intents – Record each open intent of task_id as output.status='abandoned'.
  • announce_once – Emit a warning once per process using the runner's warning ledger.
  • check_agent – Refuse agent_id if this runner governs another agent; bind nothing.
  • counts_by_enforcement_point – Return the enforced policy count per enforcement point.
  • create_app – Build the optional HTTP server with a shared runner in app.state.
  • end_task – Close a task: evict engine history and flush the step buffer.
  • ensure_policies – Load or refresh the policy set through the configured source.
  • evaluate – Preflight check for intended before it executes.
  • evaluate_registration – Evaluate agent_registration policies in-process after register_agent().
  • fetch_policies – Load the client's fetched set and report whether its cache was persisted.
  • flush – Wait for queued incident and trace reports to be attempted.
  • is_task_closed – Whether task_id is closing or closed in this runner.
  • is_task_open – Whether task_id has started in this runner and not been closed by end_task.
  • load_policies – Load policies into the engine as if a fetch had produced them.
  • load_report – Return the engine's report for the most recent policy load.
  • policy_count – Return the number of policies currently being enforced.
  • policy_status – Return the current status of the policy cache.
  • record – Settle an intent by action_id and buffer its outcome for logging.
  • report_incident – Route an incident payload to the configured incident sink.
  • report_status – Return the outbound report queue's counters.
  • restore_intent – Replay a persisted open intent without evaluating it a second time.
  • restore_task – Mark task_id open in this runner without recording a task.start.
  • signal_policy_state – Announce the current load outcome on stderr, at most once.
  • stop – Stop background threads and shut down exporters.
  • sweep_stale_tasks – Evict abandoned tasks that have exceeded their maximum idle age.

Attributes:

Configure a runner, applying keyword overrides to its supplied settings.

kyvvu_engine.KyvvuRunner.abandon_open_intents

abandon_open_intents(task_id: str) -> list[Behavior]

Record each open intent of task_id as output.status='abandoned'.

An intent is open from an allowed evaluate until record settles it. Call this before recording a closing boundary of your own so that every step of the task precedes it. record does this itself for task.end and task.error, and end_task for a task closed without either, so a caller rarely needs to.

Returns:

  • list[Behavior] – The abandoned steps as recorded, in evaluation order.

kyvvu_engine.KyvvuRunner.announce_once

announce_once(signature: str, message: str) -> None

Emit a warning once per process using the runner's warning ledger.

kyvvu_engine.KyvvuRunner.check_agent

check_agent(agent_id: str) -> None

Refuse agent_id if this runner governs another agent; bind nothing.

For a caller that wants to refuse a foreign agent before doing anything else -- kyvvu serve does, so that a wrong agent_id is told so rather than told its task is not open. Refuses an id other than the served one, or other than the one the runner is bound to. Fetches nothing: an API runner that has not loaded its policies yet judges only against a binding it already has.

Parameters:

  • agent_id (str) – The agent_id the caller names.

Raises:

  • AgentBindingError – If the runner governs another agent.

kyvvu_engine.KyvvuRunner.counts_by_enforcement_point

counts_by_enforcement_point() -> dict[str, int]

Return the enforced policy count per enforcement point.

policy_count() says how much is loaded; this says how much of it applies to a running agent. A set that is entirely agent_registration is checked once and governs nothing thereafter, and a single total cannot report that -- which is how a diagnostic ends up printing "1 policy in effect (enforcing)" about an agent that enforces nothing at every step it takes.

Returns:

  • dict[str, int] – Mapping of enforcement_point to policy count; points with no
  • dict[str, int] – policies are absent.

kyvvu_engine.KyvvuRunner.create_app

create_app(*, agent_key: str | None = None, api_url: str | None = None, api_key: str | None = None, **settings_kwargs: Any) -> FastAPI

Build the optional HTTP server with a shared runner in app.state.

kyvvu_engine.KyvvuRunner.end_task

end_task(task_id: str, agent_id: str | None = None) -> None

Close a task: evict engine history and flush the step buffer.

Records each remaining open intent as output.status='abandoned' before eviction; recording task.end or task.error already did so before the boundary, which keeps the boundary last. end_task() is the one call still accepted for a closing task (one whose task.end/task.error is recorded). After this the runner refuses any write into task_id (a task_id is never reused; the runner remembers its most recent 100,000 closed ids). Calls engine.end_task(task_id) then, if a log endpoint is configured, hands all buffered steps to the background report worker and returns: it does not wait for the network. Use flush to wait for delivery. Network errors are logged and swallowed by the worker.

Exporters receive a batch of a task's steps; a task may be delivered in several batches, and is delivered at the latest by end_task().

Parameters:

  • task_id (str) – Task identifier to close.
  • agent_id (str | None) – Caller identity, when available. Refused if it differs from the agent already bound to this runner.

kyvvu_engine.KyvvuRunner.ensure_policies

ensure_policies(*, raise_on_not_found: bool = False) -> PolicyStatus

Load or refresh the policy set through the configured source.

The public form of the runner's own lazy-load step. It dispatches on policy_source_kind, so it works for a cache-only configuration where fetch_policies would raise, and it respects the TTL: calling it repeatedly is cheap.

Callers that want to know at startup whether anything will be enforced should call this rather than fetch_policies, which is the API transport primitive and nothing more.

Returns:

Raises:

  • KyvvuConfigError – Only from the API path, and only when the API is the configured source but its credentials are incomplete.

kyvvu_engine.KyvvuRunner.evaluate

evaluate(intended: Behavior, context: EvalContext) -> EvalResult

Preflight check for intended before it executes.

Ensures policies are loaded (fetching if the TTL has expired), then delegates to evaluate. An allowed or warned step.* preflight registers an open intent, visible to later policy evaluations until record settles it. Evaluation and registration are atomic for concurrent calls on the same task.

The one synchronous network wait is the FIRST policy fetch (a cold runner, or an expired cache with no refresh thread yet): it blocks the caller for up to http_timeout_seconds when the platform is unreachable, then falls back to the disk cache or inline set. After that first fetch the policy refresh runs on its own thread and evaluate() never waits on the network: incident reports go through the bounded report queue, not this thread.

When fail_mode="closed" and policies are unavailable, action steps produce a synthetic Action.block result.

A blocked step.* behaviour is recorded here (decision D1): before raising, the runner records it, in the shape blocked_step builds, under the evaluation's action_id and with meta["kyvvu.eval"], inside the same task lock. exc.result.recorded_seq is its seq. The caller must not record it again: a record() of that action_id is refused as a duplicate. A blocked task boundary (task.*) is not recorded; how one is represented is not part of the 1.0 contract.

While the registration is blocked (see evaluate_registration), every step.* behaviour is blocked without evaluating its step policies, by a result naming the registration policies, and recorded the same way. No step incident is reported for it: the registration incident already was. Task boundaries are evaluated as usual.

Evaluation time is intended.timestamp: time-based rules read the time the client recorded when it formed the intent, so build the behaviour then. A preflight is judged at that time, not when the action later executes. A timestamp more than max_clock_skew_seconds (KV_MAX_CLOCK_SKEW_SECONDS, default 300) from this host's UTC clock, in either direction, is refused with ValueError before the task is touched or policies load. record, restore_task and restore_intent are not bounded, because they replay recorded history; to re-judge a recorded behaviour, evaluate it with a PolicyEngine.

Lifecycle: an evaluate() of anything but task.start needs the task's task.start recorded first (or restore_task); a task.start for a started task, and any write into a closing or closed task, is refused.

Parameters:

  • intended (Behavior) – The step the agent intends to execute. output must be None (pre-execution). Its payloads are re-checked as I-JSON. Its action_id, when given, is unique per (agent_id, task_id): the one legal reuse is recording this step's completion with record. One is generated when omitted. meta keys starting with kyvvu. are reserved and refused.
  • context (EvalContext) – Agent identity and execution metadata.

Returns:

Note When `action == Action.warn`, a `warnings.warn()` is emitted and the incident webhook fires (if configured). Execution continues. Only `Action.block` raises `KyvvuBlockedError`.

Raises:

  • KyvvuBlockedError – If result.action == Action.block.
  • KyvvuConfigError – If the behavior and context name different agents, or this runner is already bound to another agent. With the platform as the source, any agent_id but the one it serves the policy set for is refused; on a cold runner the policies are loaded first, so that refusal precedes the first binding.
  • TaskClosedError – If the task is closing or closed in this runner.
  • TaskStateError – If the task has not started (for anything but a task.start), or a task.start names a started task.
  • ValueError – If action_id is already used in the task, a payload is not I-JSON, meta carries a reserved kyvvu. key, or intended.timestamp is more than max_clock_skew_seconds from this host's clock (see _check_clock_skew).

kyvvu_engine.KyvvuRunner.evaluate_registration

evaluate_registration(agent_data: dict[str, Any], context: EvalContext) -> EvalResult

Evaluate agent_registration policies in-process after register_agent().

Ensures policies are loaded (fetching if TTL has expired), then delegates to evaluate_registration. Raises KyvvuBlockedError if the engine returns Action.block.

The registration is held, and re-evaluated when its policies change. The runner keeps agent_data and context, and the digest of the agent_registration policies they were evaluated against. A later load whose digest differs (a manifest assigned to the running agent, a severity edited) evaluates them again, with the same consequences as here except that nothing is raised: violations are reported as a registration incident, and a block puts the runner in the registration-blocked state. A block here does too. While it holds, evaluate refuses every step.* behaviour with a block naming the registration policies, recorded like any other block; policy_status reports it as registration_blocked. An evaluation that does not block, here or after a later load, clears it.

Parameters:

  • agent_data (dict[str, Any]) – Agent profile dict with current registration metadata.
  • context (EvalContext) – Agent identity and classification metadata.

Returns:

  • EvalResult – class:~kyvvu_engine.schemas.EvalResult with aggregate outcome.

Raises:

  • KyvvuBlockedError – If action is block.
  • AgentBindingError – If the platform is the policy source and serves the set for another agent_id than context.agent_id.
  • ValueError – If agent_data is not I-JSON. Registration policies read it and an incident may quote it, so it is refused rather than silently converted.

kyvvu_engine.KyvvuRunner.fetch_policies

fetch_policies(*, raise_on_not_found: bool = False) -> bool

Load the client's fetched set and report whether its cache was persisted.

kyvvu_engine.KyvvuRunner.flush

flush(timeout: float | None = None) -> bool

Wait for queued incident and trace reports to be attempted.

end_task() and evaluate() only enqueue. A short-lived script that must not lose its trace calls this before exiting; it is also called automatically at interpreter exit with report_exit_flush_seconds as the deadline. A hard kill (SIGKILL) can still lose queued reports.

Parameters:

  • timeout (float | None) – Seconds to wait; None waits until the queue is empty.

Returns:

  • bool – True when everything queued so far has been attempted,
  • bool – False when the deadline passed first.

kyvvu_engine.KyvvuRunner.instance_id

instance_id: str

The resolved instance identifier for this runner.

kyvvu_engine.KyvvuRunner.is_task_closed

is_task_closed(task_id: str) -> bool

Whether task_id is closing or closed in this runner.

Closing: its task.end or task.error is recorded and only end_task may follow. Closed: ended or evicted. Either way any further write raises TaskClosedError. The memory of closed ids is bounded (the most recent 100,000), so False does not prove an id is unused.

kyvvu_engine.KyvvuRunner.is_task_open

is_task_open(task_id: str) -> bool

Whether task_id has started in this runner and not been closed by end_task.

Still True while the task is closing, so that end_task can finish it.

kyvvu_engine.KyvvuRunner.load_policies

load_policies(policies: list[dict[str, Any]], *, source: PolicySourceKind = PolicySourceKind.disk_cache, stale: bool = False, stale_cause: StaleCause | None = None, served_agent_id: str | None = None) -> PolicyStatus

Load policies into the engine as if a fetch had produced them.

For callers that obtain policies themselves — a hook process reading its own on-disk cache, a test, an embedder with a bespoke transport — and would otherwise have to reach into runner._engine and set runner._policies_loaded_at by hand. It performs the whole bookkeeping a fetch does: loads the set, derives manifest provenance, marks the cache fresh, records the source, and announces the outcome if it is worth announcing.

The set is treated as current unless the caller says otherwise. A caller passing a set it just obtained is asserting that this is what the source says, so load_policies([p]) reports enforcing and load_policies([]) reports none_configured — not stale_cache and unavailable. Reporting a freshly refreshed set as stale would make the idiom this class documents (refuse to run unless load_state is enforcing) refuse a perfectly governed session.

It deliberately does not touch last_fetch_succeeded: nothing was fetched, so the fetch state stays whatever the last real attempt left it as.

Parameters:

  • policies (list[dict[str, Any]]) – Raw policy dicts in the shape the API returns.
  • source (PolicySourceKind) – Where the caller got them from, recorded as source.
  • stale (bool) – Pass True when the caller knows the set is older than its source — a cache it could not refresh, for instance. The reported state then becomes stale_cache.
  • stale_cause (StaleCause | None) – Why it is stale, reported as stale_cause; caller_marked when stale is set without one. Ignored when stale is False.
  • served_agent_id (str | None) – The platform agent_id the set was served for, when the caller knows it -- a hook process reading a cache that recorded the X-Kyvvu-Agent-Id of the fetch that wrote it, say. The runner then binds to it exactly as its own fetch would have: any other agent_id is refused with AgentBindingError, rather than evaluated against a set none of whose agent-specific policies apply to it. None, the default, leaves the binding as it is.

Returns:

Raises:

  • ValueError – If served_agent_id is given and is not a non-empty agent_id in the identifier grammar. Nothing is loaded.
  • PolicyLoadError – If the engine refuses the set. Nothing is loaded.

kyvvu_engine.KyvvuRunner.load_report

load_report() -> PolicyLoadReport

Return the engine's report for the most recent policy load.

The transport-aware view is policy_status, whose load_state can also say stale_cache; this is the engine's own accounting of offered / enforced.

kyvvu_engine.KyvvuRunner.policy_count

policy_count() -> int

Return the number of policies currently being enforced.

The public equivalent of runner._engine.policy_count(). Zero means this runner is enforcing nothing; policy_status says why.

kyvvu_engine.KyvvuRunner.policy_status

policy_status() -> PolicyStatus

Return the current status of the policy cache.

Returns:

  • PolicyStatus – class:~kyvvu_engine.schemas.PolicyStatus with policy count,
  • PolicyStatus – fetch timestamps, staleness, source, and TTL info.

kyvvu_engine.KyvvuRunner.record

record(step: Behavior, context: EvalContext | None = None) -> Behavior

Settle an intent by action_id and buffer its outcome for logging.

An outcome without an open intent is appended as a replayed step.

Parameters:

  • step (Behavior) – A completed Behavior with output populated. Its action_id is unique per (agent_id, task_id): it either pairs the step with the open intent evaluate registered under the same id, or names a step not recorded before. meta keys starting with kyvvu. are reserved and refused; the runner adds kyvvu.eval and kyvvu.manifests itself.
  • context (EvalContext | None) – Optional caller context, checked against the step's agent.

seq orders a task's recorded steps: strictly increasing per (agent_id, task_id), assigned by the recording client. Consumers must not assume it is contiguous, starts at a particular value, or reflects wall-clock order; a value supplied by the caller may be ignored.

Lifecycle: recording task.start opens the task; anything else needs it open. Recording task.end or task.error makes the task closing: every later record/evaluate/restore_intent for it raises TaskClosedError, and only end_task may follow. A blocked step is recorded by evaluate itself; do not record it here.

The step's timestamp is not checked against the host clock: a step recorded here may be history replayed from an earlier process (kyvvu-claude rebuilds each hook's runner this way), so any recorded time is accepted as it is.

Returns:

  • Behavior – The recorded step, with seq assigned.

Raises:

  • ValueError – If a payload value (including the injected meta["kyvvu.eval"] record) is not I-JSON, the timestamp is naive, the action_id was already recorded in the task (including by a block evaluate recorded), or meta carries a reserved kyvvu. key.
  • KyvvuConfigError – If the step and context name different agents, or this runner is already bound to another agent, including the agent_id the platform serves its policies for. Nothing is fetched here: that agent_id is checked only once a load has learned it.
  • TaskClosedError – If the task is closing or closed.
  • TaskStateError – If the task has not started (for anything but a task.start), or a task.start names a started task.

kyvvu_engine.KyvvuRunner.report_incident

report_incident(payload: dict[str, Any]) -> None

Route an incident payload to the configured incident sink.

This is the single incident entry point shared by the engine's own _maybe_report_* helpers and by external consumers (e.g. kyvvu-claude). The destination follows the location + format model (incident_location / incident_format, defaulting to the trace sink). No-op when the incident sink is disabled.

Best-effort and off the caller's thread: a network sink is fed through the bounded report queue and sent by the background worker, so a slow platform cannot delay the step that raised the incident. A full queue drops the incident and counts it (report_status). Local sinks (stdout, file) run inline.

Parameters:

  • payload (dict[str, Any]) – The incident payload dict (agent_id, enforcement_point, action, risk_score, violations, optional task_id/action_id/step_name/kind, timestamp). It must be I-JSON: it is checked here, before it reaches any sink (OTLP included), and a payload that is not is dropped with a warning rather than altered.

kyvvu_engine.KyvvuRunner.report_status

report_status() -> ReportStatus

Return the outbound report queue's counters.

dropped is the number of reports refused because the queue was full (report_queue_size); each overflow is also a WARNING.

kyvvu_engine.KyvvuRunner.restore_intent

restore_intent(intended: Behavior) -> None

Replay a persisted open intent without evaluating it a second time.

The task must be open (its task.start recorded, or restore_task), and not closing or closed. Its timestamp is the time it was evaluated against when it was proposed, and is not checked against the host clock here.

Raises:

  • ValueError – If the intent has an output, a payload is not I-JSON, its action_id is in use, or meta carries a reserved kyvvu. key.
  • TaskClosedError – If the task is closing or closed.
  • TaskStateError – If the task has not started, or the intent is a task boundary.

kyvvu_engine.KyvvuRunner.restore_task

restore_task(task_id: str, agent_id: str) -> None

Mark task_id open in this runner without recording a task.start.

For rebuilding a task's rule history in a fresh runner when the history being replayed does not begin with the task's task.start (it was recorded by an earlier process, and the replayed history is a bounded tail). After this, replay the history with record and restore_intent, then evaluate as usual. Do not replay a task.start after it: the task has started, and a task.start is refused. Nothing is recorded or exported by this call.

The task's earlier steps were exported elsewhere, so this runner does not hold the head of its hash chain: what it exports for the task is sent without hash and the platform numbers and chains it.

This rebuilds one runner's view of a task; it says nothing about how a task is continued across processes, which the audit chain planned for 1.1 will define.

Parameters:

  • task_id (str) – The task to mark open.
  • agent_id (str) – The agent the task belongs to; checked against any existing binding but does not bind an empty engine, and is refused if this runner is bound to another agent, or if the platform serves its policies for another agent_id.

Raises:

  • TaskClosedError – If the task is closing or closed in this runner.
  • TaskStateError – If the task has already started in this runner.
  • ValueError – If an identifier is outside its grammar.

kyvvu_engine.KyvvuRunner.settings

settings: KyvvuSettings

The resolved settings for this runner instance.

kyvvu_engine.KyvvuRunner.signal_policy_state

signal_policy_state() -> None

Announce the current load outcome on stderr, at most once.

Public because a caller that loads policies through its own transport (kyvvu serve at boot, kyvvu-claude's hooks) should be able to make the same announcement without reaching into the runner. Idempotent and cheap: safe to call whenever a caller thinks the state may have moved.

Guarded twice, for two different reasons:

  • Per runner, by load generation, so this is cheap to call from the evaluate path: an unchanged policy set is never re-examined.
  • Per process, by outcome signature, so several reloads with the same result do not repeat one paragraph until it stops being read. The signature includes the agent_key and the source as well as the state and counts, because two different agents both reporting none_configured — the default state of every new account, which is the case this signal exists for — are two ungoverned agents, not one repeated message.

Destination is stderr, matching the one-time auth hint in HttpExporter: this has to reach someone running python agent.py, who has configured no logging handler and would never see a logger.warning.

This method cannot raise. It is reached from evaluate() via _ensure_policies, and one of its callers (kyvvu-claude's PreToolUse hook) converts any exception on that path into an allow verdict. An unwritable stderr — a closed fd, or a BrokenPipeError when a hook subprocess's parent has gone away — would therefore let the report of non-enforcement cause non-enforcement, blocking a step that a perfectly valid policy was about to block. Everything below is wrapped: a failure to announce degrades to a logger.debug and never to a second write on the same broken sink.

kyvvu_engine.KyvvuRunner.stop

stop() -> None

Stop background threads and shut down exporters.

Safe to call multiple times or before the threads have started.

kyvvu_engine.KyvvuRunner.sweep_stale_tasks

sweep_stale_tasks(max_age_seconds: int | None = None) -> int

Evict abandoned tasks that have exceeded their maximum idle age.

Tasks that never call end_task() retain engine history and, when logging is enabled, an exporter buffer. The background sweeper calls this automatically every sweep_interval_seconds.

Eviction is local to this runner: it frees this process's memory and says nothing about whether the agent's task is over. Open intents are recorded as abandoned, then a task.idle with properties.task.reason = "expired" ends the evicted buffer. It is not a terminal boundary on the wire: read it as "this runner stopped tracking the task", not as the end or failure of the task. (A task whose task.end/task.error is already recorded gets no task.idle.) This runner refuses later writes into the evicted task_id with TaskClosedError.

When sweep_flush_on_evict is enabled (default), a best-effort batch post is attempted for each evicted task before discarding the buffer.

Parameters:

  • max_age_seconds (int | None) – Tasks idle longer than this (in seconds) are evicted. Defaults to settings.task_max_age_seconds (3600).

Returns:

  • int – Number of tasks evicted.

kyvvu_engine.KyvvuSettings

KyvvuSettings(*, api_url: str = (lambda: os.getenv('KV_API_URL', 'https://platform.kyvvu.com').rstrip('/'))(), api_key: str | None = (lambda: os.getenv('KV_API_KEY'))(), agent_key: str | None = (lambda: os.getenv('KV_AGENT_KEY'))(), declared_tools: list[str] | None = (lambda: [name.strip() for name in os.getenv('KV_DECLARED_TOOLS', '').split(',') if name.strip()] or None)(), incident_location: str | None = (lambda: os.getenv('KV_INCIDENT_LOCATION'))(), incident_format: str | None = (lambda: os.getenv('KV_INCIDENT_FORMAT'))(), instance_id: str | None = (lambda: os.getenv('KV_INSTANCE_ID'))(), log_payloads: str = (lambda: os.getenv('KV_LOG_PAYLOADS', 'full'))(), environment: str = (lambda: os.getenv('KV_ENVIRONMENT', 'production'))(), policy_ttl_seconds: int = (lambda: int(os.getenv('KV_POLICY_TTL_SECONDS', '300')))(), registration_ttl: float | None = (lambda: parse_ttl(os.getenv('KV_REGISTRATION_TTL')))(), http_timeout_seconds: int = (lambda: int(os.getenv('KV_HTTP_TIMEOUT_SECONDS', '10')))(), report_queue_size: int = (lambda: int(os.getenv('KV_REPORT_QUEUE_SIZE', '1000')))(), report_exit_flush_seconds: float = (lambda: float(os.getenv('KV_REPORT_EXIT_FLUSH_SECONDS', '2')))(), task_max_age_seconds: int = (lambda: int(os.getenv('KV_TASK_MAX_AGE_SECONDS', '3600')))(), sweep_interval_seconds: int = (lambda: int(os.getenv('KV_SWEEP_INTERVAL_SECONDS', '300')))(), sweep_enabled: bool = (lambda: os.getenv('KV_SWEEP_ENABLED', 'true').lower() in ('true', '1', 'yes'))(), sweep_flush_on_evict: bool = (lambda: os.getenv('KV_SWEEP_FLUSH_ON_EVICT', 'true').lower() in ('true', '1', 'yes'))(), max_clock_skew_seconds: int = (lambda: int(os.getenv('KV_MAX_CLOCK_SKEW_SECONDS', '300')))(), log_level: str = (lambda: os.getenv('KV_LOG_LEVEL', 'INFO').upper())(), fail_mode: str = (lambda: os.getenv('KV_POLICY_FAIL_MODE', 'closed').lower())(), policies: list[dict[str, Any]] | None = None, policy_file: str = (lambda: os.getenv('KV_POLICY_FILE', ''))(), explicit_fields: frozenset[str] = frozenset(), policy_cache_path: str = (lambda: os.getenv('KV_POLICY_CACHE_PATH', ''))(), policy_cache_max_age_seconds: int = (lambda: int(os.getenv('KV_POLICY_CACHE_MAX_AGE_SECONDS', '86400')))(), log_location: str | None = (lambda: os.getenv('KV_LOG_LOCATION', AUTO_LOG_LOCATION))(), log_format: str = (lambda: os.getenv('KV_LOG_FORMAT', 'kv'))(), template_location: str | None = (lambda: os.getenv('KV_TEMPLATE_LOCATION'))()) -> None

Resolved configuration for a KyvvuRunner instance.

Instantiate with keyword arguments to override individual settings, or call with no arguments to read entirely from the environment. Every field is keyword-only, so a field can be added in any position.

Attributes:

  • api_url (str) – Base URL of the Kyvvu API.
  • api_key (str | None) – Bearer API key for authentication.
  • agent_key (str | None) – Agent key used to fetch agent-specific policies.
  • declared_tools (list[str] | None) – Tool names allowed for this agent's steps. Defaults to the comma-separated KV_DECLARED_TOOLS environment variable.
  • incident_location (str | None) – WHERE incidents go (same vocabulary as log_location). None inherits the trace sink location (with the …/api/v1/incidents kv path substituted for …/api/v1/logs/batch). See effective_incident_location.
  • incident_format (str | None) – HOW incidents are formatted ("kv", "json", "otlp"). None inherits the trace format (log_format).
  • instance_id (str | None) – Human-readable instance prefix. A random suffix is appended at each runner construction, so the instance id is not stable across restarts and is not an identity. None generates a UUID4.
  • log_payloads (str) – "full" or "metadata_only". Controls whether batch log payloads include raw input/output or redacted metadata.
  • environment (str) – Deployment environment label forwarded to EvalContext.
  • policy_ttl_seconds (int) – Seconds before the policy cache is refreshed.
  • registration_ttl (float | None) – Seconds before a registration cache entry expires, or None for no expiry.
  • http_timeout_seconds (int) – Per-request HTTP timeout in seconds.
  • report_queue_size (int) – Bound of the outbound report queue (incidents and trace flushes). Overflow drops and counts. Default: 1000.
  • report_exit_flush_seconds (float) – Deadline for the atexit flush of queued reports. Default: 2.
  • task_max_age_seconds (int) – Seconds before an abandoned task buffer is evicted by sweep_stale_tasks(). Default: 3600.
  • sweep_interval_seconds (int) – How often the background sweeper runs (in seconds). Default: 300 (5 minutes).
  • sweep_enabled (bool) – Whether the background sweeper thread is started automatically. Default: True. Set to False for tests or deployments that call sweep_stale_tasks() manually. The runner warns that task memory is unbounded when this is off, except when it is named in explicit_fields (passed as a KyvvuRunner keyword), which marks it deliberate.
  • sweep_flush_on_evict (bool) – Whether to attempt a batch log post for evicted tasks before discarding their buffers. Default: True.
  • max_clock_skew_seconds (int) – How far a behaviour's timestamp may be from the host clock, in either direction, for KyvvuRunner.evaluate() to accept it. Replay paths are not bounded. Must be positive. Default: 300.
  • log_level (str) – Python log level name for kyvvu / kyvvu_engine loggers. Applied when the settings object is instantiated. Default: "INFO".
  • fail_mode (str) – "closed" (default) or "open". When policies are unavailable, closed blocks action steps; open allows them. Task-boundary markers and an empty policy set are allowed.
  • policies (list[dict[str, Any]] | None) – An inline policy set, in the shape load_policies takes. Keyword-only in practice: there is no KV_* variable for it, because a policy set is not an environment-variable-shaped value. None means "not a source"; an explicitly empty list is a source that offers nothing, which loads as none_configured rather than being silently ignored.
  • policy_file (str) – One or more manifest YAML paths, comma-separated, loaded as the policy source with no credentials and no network.
  • explicit_fields (frozenset[str]) – Which settings the caller supplied by name, as opposed to defaults resolved from the environment. Provenance, not a knob: it is what lets this class honour the precedence its own module docstring promises ("explicit keyword arguments" outrank "environment variables") when choosing a policy source, which it otherwise cannot, because by the time the fields are populated a value from a kwarg and a value from KV_* look identical. Populated by KyvvuRunner from its own keyword arguments and by create_app from its overrides. Empty for a directly constructed KyvvuSettings, which therefore keeps the plain resolution order. Also read for sweep_enabled: an explicit False silences the sweep-disabled warning.
  • policy_cache_path (str) – File path for on-disk policy cache. Empty string disables disk caching.
  • policy_cache_max_age_seconds (int) – Maximum age of the on-disk cache before a staleness warning is emitted. Default: 86400 (24h).
  • log_location (str | None) – WHERE logs go. URL → HTTP POST, file path → JSONL, "stdout" → print, "none"/"" → disabled, "auto" → the platform when it is the policy source, else "stdout" (see effective_log_location). Default: "auto".
  • log_format (str) – HOW logs are formatted: "kv" (default), "json", or "otlp".
  • template_location (str | None) – Custom behavior template path, or None to use each integration's packaged template.

Functions:

kyvvu_engine.KyvvuSettings.agent_key

agent_key: str | None = field(default_factory=lambda: os.getenv('KV_AGENT_KEY'))

kyvvu_engine.KyvvuSettings.api_key

api_key: str | None = field(default_factory=lambda: os.getenv('KV_API_KEY'))

kyvvu_engine.KyvvuSettings.api_url

api_url: str = field(default_factory=lambda: os.getenv('KV_API_URL', 'https://platform.kyvvu.com').rstrip('/'))

kyvvu_engine.KyvvuSettings.declared_tools

declared_tools: list[str] | None = field(default_factory=lambda: [name.strip() for name in os.getenv('KV_DECLARED_TOOLS', '').split(',') if name.strip()] or None)

kyvvu_engine.KyvvuSettings.effective_incident_format

effective_incident_format: str

Resolve the incident format, inheriting the trace format.

Returns:

  • str – incident_format if explicitly set, otherwise
  • str – effective_log_format ("kv", "json", or "otlp").

kyvvu_engine.KyvvuSettings.effective_incident_location

effective_incident_location: str | None

Resolve the incident location, inheriting the trace sink.

When incident_location is explicitly set, it is resolved with the same none/off/"" → None and "auto" handling as effective_log_location. Every document describing this setting says it shares the trace sink's vocabulary, so auto has to mean something here too: without it KV_INCIDENT_LOCATION=auto fell through to the file branch and appended incidents to a file named auto, with no error.

When incident_location is unset, the location is derived from effective_log_location: a trailing /api/v1/logs/batch (the trace HTTP path) is stripped to recover the API base, which the incident exporter then extends with /api/v1/incidents for the kv format. Non-URL trace locations ("stdout", file paths) are returned verbatim. Returns None when the trace sink is disabled.

Returns:

  • str | None – Resolved incident location string, or None if disabled.

kyvvu_engine.KyvvuSettings.effective_log_format

effective_log_format: str

Resolve the log format.

Returns:

  • str – Format string: "kv", "json", or "otlp".

kyvvu_engine.KyvvuSettings.effective_log_location

effective_log_location: str | None

Resolve the log location.

"auto" -- the default -- means the trace follows the policy source: when the platform is what tells this agent what it may do, the platform gets the record of what it did; otherwise the trace stays on stdout.

The old default was the literal "stdout", which made the obvious configuration -- Kyvvu(api_url=..., api_key=..., agent_key=...) -- register the agent over HTTP and then print every step. Nothing errored, the Python client reported success, and the platform stored no audit trail at all, which for a compliance product is the one failure that matters. It was found against stage as five registered agents and SELECT count(*) FROM log_entries returning 0.

The condition is policy_source_kind is api, not merely "an api_key is set", and the difference is deliberate. api_key lives in the environment of anyone who has ever used the platform, so keying on it would make the credential-free offline path -- KyvvuRunner(policy_file="my-policies.yaml") -- start POSTing a local trial run's trace to platform.kyvvu.com in every such shell. An offline path that is offline only in a clean shell is not one, and that direction of the mistake ships data rather than withholding it.

Returns:

  • str | None – Resolved location string, or None if logging is disabled
  • str | None – (log_location is None, "none", "off", or empty).

kyvvu_engine.KyvvuSettings.effective_policies_endpoint

effective_policies_endpoint() -> str

Return the policies URL.

kyvvu_engine.KyvvuSettings.environment

environment: str = field(default_factory=lambda: os.getenv('KV_ENVIRONMENT', 'production'))

kyvvu_engine.KyvvuSettings.explicit_fields

explicit_fields: frozenset[str] = frozenset()

kyvvu_engine.KyvvuSettings.fail_mode

fail_mode: str = field(default_factory=lambda: os.getenv('KV_POLICY_FAIL_MODE', 'closed').lower())

kyvvu_engine.KyvvuSettings.has_policy_source

has_policy_source() -> bool

Return True when this configuration can reach some policy source.

Convenience wrapper over policy_source_kind. False means the configuration can never load a policy, so anything built from it would enforce nothing — which is why KyvvuRunner refuses it at construction rather than at the first evaluate().

kyvvu_engine.KyvvuSettings.http_timeout_seconds

http_timeout_seconds: int = field(default_factory=lambda: int(os.getenv('KV_HTTP_TIMEOUT_SECONDS', '10')))

kyvvu_engine.KyvvuSettings.incident_format

incident_format: str | None = field(default_factory=lambda: os.getenv('KV_INCIDENT_FORMAT'))

kyvvu_engine.KyvvuSettings.incident_location

incident_location: str | None = field(default_factory=lambda: os.getenv('KV_INCIDENT_LOCATION'))

kyvvu_engine.KyvvuSettings.instance_id

instance_id: str | None = field(default_factory=lambda: os.getenv('KV_INSTANCE_ID'))

kyvvu_engine.KyvvuSettings.log_format

log_format: str = field(default_factory=lambda: os.getenv('KV_LOG_FORMAT', 'kv'))

kyvvu_engine.KyvvuSettings.log_level

log_level: str = field(default_factory=lambda: os.getenv('KV_LOG_LEVEL', 'INFO').upper())

kyvvu_engine.KyvvuSettings.log_location

log_location: str | None = field(default_factory=lambda: os.getenv('KV_LOG_LOCATION', AUTO_LOG_LOCATION))

kyvvu_engine.KyvvuSettings.log_payloads

log_payloads: str = field(default_factory=lambda: os.getenv('KV_LOG_PAYLOADS', 'full'))

kyvvu_engine.KyvvuSettings.max_clock_skew_seconds

max_clock_skew_seconds: int = field(default_factory=lambda: int(os.getenv('KV_MAX_CLOCK_SKEW_SECONDS', '300')))

kyvvu_engine.KyvvuSettings.missing_policy_source_settings

missing_policy_source_settings() -> list[str]

Name the settings whose absence leaves this configuration sourceless.

Only meaningful when has_policy_source is False. Each entry names both the keyword argument and the environment variable, because either is a legitimate way to supply it and an error that names only one sends half the readers to the wrong place.

Returns:

  • list[str] – Human-readable "kwarg= / KV_VAR" strings for every missing
  • list[str] – piece of the API source, in the order a caller would set them.

kyvvu_engine.KyvvuSettings.policies

policies: list[dict[str, Any]] | None = None

kyvvu_engine.KyvvuSettings.policy_cache_max_age_seconds

policy_cache_max_age_seconds: int = field(default_factory=lambda: int(os.getenv('KV_POLICY_CACHE_MAX_AGE_SECONDS', '86400')))

kyvvu_engine.KyvvuSettings.policy_cache_path

policy_cache_path: str = field(default_factory=lambda: os.getenv('KV_POLICY_CACHE_PATH', ''))

kyvvu_engine.KyvvuSettings.policy_file

policy_file: str = field(default_factory=lambda: os.getenv('KV_POLICY_FILE', ''))

kyvvu_engine.KyvvuSettings.policy_source_kind

policy_source_kind() -> PolicySourceKind

Return which policy source this configuration can reach.

This answers a question about configuration, not about runtime state, and it deliberately performs no I/O. In particular it does not check whether policy_cache_path currently points at an existing, parseable file: whether a cache file has been written yet is runtime state that changes under the process, while __init__ is judging whether the operator supplied a workable configuration. Making construction depend on a stat() would also make it fail on a first run and succeed on the second, and — where a runner is built on an enforcement path that converts exceptions into "allow" (kyvvu-claude's hooks) — would turn a loud "0 policies" into a silent permit.

Whether a source actually answered is a separate question, reported as PolicyLoadState on policy_status.

Resolution order — the first source that is fully configured wins:

  1. local, but only when it was asked for explicitly and the API was not (see below).
  2. api — api_url and api_key and agent_key.
  3. local — an inline policies list, or a non-empty policy_file.
  4. disk_cache — a non-empty policy_cache_path.
  5. none — nothing; such a configuration can never enforce anything.

Step 1 exists because this module's own documented precedence is "explicit keyword arguments" over "environment variables", and source selection was the one place that ignored it. KyvvuRunner(policy_file= "my-policies.yaml") — the credential-free entry point exists to provide, and the first thing in the engine README's quickstart — became network-dependent and enforced nothing for any reader with KV_API_KEY exported, which is every reader who has ever used the platform. A documented offline path that is offline only in a clean shell is not one.

It is deliberately narrow, and cannot demote a credentialed deployment:

  • Both supplied explicitly -> api still wins. A caller naming both gets the authenticated, current set.
  • Both merely present in the environment -> api still wins. A manifest dropped on a production host cannot displace the org's policy set with a stale local file, which is the hazard that keeps api first.

local sits above disk_cache because the two mean different things: a caller who names a manifest file is naming the source, while a cache path names a place an earlier successful fetch may have written to. It sits below api so that adding a local file to a credentialed deployment cannot quietly demote it off the authenticated, current policy set — the credential-free path is for the trial, not a way to shadow production policy.

policies=[] — an explicitly empty list — resolves to local, not to none. It is a source that answered with nothing, which loads as none_configured and says so loudly; treating it as "no source" would report the same configuration as a construction-time mistake instead.

Returns:

kyvvu_engine.KyvvuSettings.policy_ttl_seconds

policy_ttl_seconds: int = field(default_factory=lambda: int(os.getenv('KV_POLICY_TTL_SECONDS', '300')))

kyvvu_engine.KyvvuSettings.registration_ttl

registration_ttl: float | None = field(default_factory=lambda: parse_ttl(os.getenv('KV_REGISTRATION_TTL')))

kyvvu_engine.KyvvuSettings.report_exit_flush_seconds

report_exit_flush_seconds: float = field(default_factory=lambda: float(os.getenv('KV_REPORT_EXIT_FLUSH_SECONDS', '2')))

kyvvu_engine.KyvvuSettings.report_queue_size

report_queue_size: int = field(default_factory=lambda: int(os.getenv('KV_REPORT_QUEUE_SIZE', '1000')))

kyvvu_engine.KyvvuSettings.sweep_enabled

sweep_enabled: bool = field(default_factory=lambda: os.getenv('KV_SWEEP_ENABLED', 'true').lower() in ('true', '1', 'yes'))

kyvvu_engine.KyvvuSettings.sweep_flush_on_evict

sweep_flush_on_evict: bool = field(default_factory=lambda: os.getenv('KV_SWEEP_FLUSH_ON_EVICT', 'true').lower() in ('true', '1', 'yes'))

kyvvu_engine.KyvvuSettings.sweep_interval_seconds

sweep_interval_seconds: int = field(default_factory=lambda: int(os.getenv('KV_SWEEP_INTERVAL_SECONDS', '300')))

kyvvu_engine.KyvvuSettings.task_max_age_seconds

task_max_age_seconds: int = field(default_factory=lambda: int(os.getenv('KV_TASK_MAX_AGE_SECONDS', '3600')))

kyvvu_engine.KyvvuSettings.template_location

template_location: str | None = field(default_factory=lambda: os.getenv('KV_TEMPLATE_LOCATION'))

kyvvu_engine.MAX_COLLECTION_ITEMS

MAX_COLLECTION_ITEMS = 50

kyvvu_engine.MAX_DICT_ITEMS

MAX_DICT_ITEMS = 50

kyvvu_engine.MAX_RECURSION_DEPTH

MAX_RECURSION_DEPTH = 5

kyvvu_engine.MAX_REPR_LENGTH

MAX_REPR_LENGTH = 500

kyvvu_engine.MAX_STRING_LENGTH

MAX_STRING_LENGTH = 2000

kyvvu_engine.PATH_LIST_SEPARATOR

PATH_LIST_SEPARATOR = ','

kyvvu_engine.POLICY_SCHEMA_VERSION

POLICY_SCHEMA_VERSION = 1

kyvvu_engine.PlatformClient

PlatformClient(settings: KyvvuSettings, *, instance_id: str | None = None, allow_anonymous_reports: bool = False) -> None

Own agent registration and policy traffic, with their disk caches.

Registration error types are available on this class for callers that translate transport failures into their own public exceptions.

Functions:

Attributes:

Validate and retain the platform connection settings.

kyvvu_engine.PlatformClient.APIError

APIError = KyvvuAPIError

kyvvu_engine.PlatformClient.KeyRevokedError

KeyRevokedError = KyvvuKeyRevokedError

kyvvu_engine.PlatformClient.RateLimitError

RateLimitError = KyvvuRateLimitError

kyvvu_engine.PlatformClient.TimeoutError

TimeoutError = KyvvuTimeoutError

kyvvu_engine.PlatformClient.cached_registration_info

cached_registration_info(agent_key: str) -> tuple[str, datetime] | None

Return a cached agent's ID and registration time, when available.

kyvvu_engine.PlatformClient.cached_registration_response

cached_registration_response() -> dict[str, Any] | None

Return this connection's cached platform response, if one matches.

kyvvu_engine.PlatformClient.fetch_policies

fetch_policies() -> PolicyFetchResult

Fetch the complete agent set and record the result in one cache envelope.

kyvvu_engine.PlatformClient.instance_id

instance_id: str

The instance identity sent with platform policy requests.

kyvvu_engine.PlatformClient.register_agent

register_agent(profile: AgentProfile, *, risk_classification: str | None = None) -> Registration

Register profile, or return a matching cached identity.

risk_classification is the agent's EU AI Act class (high, limited, minimal). It is not part of AgentProfile because it belongs to the client configuration, like the environment taken from the settings; it is sent with the registration and seen by the registration policies.

kyvvu_engine.PlatformClient.registration_cache_directory

registration_cache_directory() -> Path

Return the directory used for cached agent registrations.

kyvvu_engine.PlatformClient.send_incident

send_incident(incident: dict[str, Any], *, endpoint: str | None = None, require_created: bool = True) -> None

Send an incident through the platform connection.

kyvvu_engine.PlatformClient.send_steps

send_steps(agent_id: str, task_id: str, steps: list[dict[str, Any]], *, idempotency_key: str | None = None, endpoint: str | None = None, require_created: bool = True) -> dict[str, Any]

Send one batch and return its acknowledgement on the expected HTTP status.

kyvvu_engine.PlatformClient.settings

settings = settings

kyvvu_engine.PolicyEngine

PolicyEngine(aggregator: Callable[[list[PolicyResult]], EvalResult] | None = None) -> None

Stateful, pure-Python policy evaluation engine.

One engine instance is bound to one agent on its first evaluation or write. The engine is safe to use from multiple threads (the internal PathTracker uses a lock; the PolicyStore is replaced atomically on load_policies).

Typical call sequence per agent step::

engine = PolicyEngine()
engine.load_policies(policies_from_api)

# Pre-execution preflight
result = engine.evaluate(intended_behavior, context)
if result.action == Action.block:
    raise PolicyBlockedError(result)

# Execute the step …
output = run_tool(...)

# Post-execution recording
completed = intended_behavior.model_copy(update={"output": output})
engine.record(completed)

At task end::

engine.end_task(task_id)

Operational notes

  • Rule exceptions are violations. If a rule function raises, its policy is violated at its configured severity, with the error in the violation details. Evaluation continues with the remaining policies.
  • Regex patterns in field_matches_regex policy params use linear-time RE2 matching. Unsupported syntax, including lookaround and backreferences, is refused at policy load.
  • One engine per agent. The engine is not designed to be shared across agents. KyvvuRunner creates a fresh engine and is configured with a single agent_key.

Functions:

  • action_id_in_use – Whether action_id is already open or recorded in task_id.
  • applicable_policy_count – Return how many loaded policies apply to agent_id.
  • check_agent – Refuse another agent without binding an empty engine.
  • check_context_agent – Refuse a behavior evaluated under another agent's context.
  • counts_by_enforcement_point – Return the loaded policy count per enforcement point.
  • end_task – Evict all history for task_id from memory and close it.
  • evaluate – Evaluate all applicable policies against an intended step.
  • evaluate_and_record – Evaluate, then record the step, including a blocked one.
  • evaluate_registration – Evaluate agent_registration policies against agent metadata.
  • explain – Return a human-readable explanation of the evaluation result.
  • get_history – Return the completed steps for task_id, oldest first.
  • get_manifests – Return the distinct manifest provenance records for loaded policies.
  • get_open_intents – Return the task's still-unsettled runner preflights.
  • is_task_closed – Whether task_id is closing (task.end/task.error recorded) or closed.
  • load_policies – Replace the engine's policy set with policies.
  • load_report – Describe the outcome of the last load_policies call.
  • policy_count – Return the number of currently loaded policies.
  • record – Record a completed step in the task's history.
  • register_intent – Expose an allowed runner preflight to later policy evaluations.
  • registration_policy_digest – Return a digest of the loaded agent_registration policies.
  • unresolved_paths – Find step policy paths absent from every supplied behavior.
  • validate_rule_params – Validate that rule_type is registered and params match its schema.

Attributes:

  • agent_id (str | None) – The agent bound by the first evaluation or write, if any.

Create an empty engine using the supplied or maximum-severity aggregator.

kyvvu_engine.PolicyEngine.action_id_in_use

action_id_in_use(task_id: str, action_id: str) -> bool

Whether action_id is already open or recorded in task_id.

action_id is unique per (agent_id, task_id); the one legal reuse is recording the completion of an open intent.

kyvvu_engine.PolicyEngine.agent_id

agent_id: str | None

The agent bound by the first evaluation or write, if any.

kyvvu_engine.PolicyEngine.applicable_policy_count

applicable_policy_count(agent_id: str) -> int

Return how many loaded policies apply to agent_id.

The companion to policy_count for a caller that has to tell "loaded" from "applies to this agent": a set fetched for one agent reports its full count whichever agent_id is evaluated, and every agent-specific policy in it is skipped for any other. See count_applicable.

Parameters:

  • agent_id (str) – Agent string identifier.

Returns:

  • int – The number of loaded policies that are workspace-wide or name
  • int – agent_id.

kyvvu_engine.PolicyEngine.check_agent

check_agent(agent_id: str) -> None

Refuse another agent without binding an empty engine.

kyvvu_engine.PolicyEngine.check_context_agent

check_context_agent(behavior: Behavior, context: EvalContext) -> None

Refuse a behavior evaluated under another agent's context.

kyvvu_engine.PolicyEngine.counts_by_enforcement_point

counts_by_enforcement_point() -> dict[str, int]

Return the loaded policy count per enforcement point.

The companion to policy_count for callers that have to tell "gates every step" from "was checked once at registration". Both sets report the same total and only one of them governs a running agent.

Returns:

  • dict[str, int] – Mapping of enforcement_point to policy count; points with no
  • dict[str, int] – policies are absent.

kyvvu_engine.PolicyEngine.end_task

end_task(task_id: str, *, agent_id: str | None = None) -> None

Evict all history for task_id from memory and close it.

Call when a task finishes (on task.end or task.error). Safe to call for unknown task IDs and more than once. Afterwards the engine refuses any record or register_intent for task_id with TaskClosedError; it remembers its most recent 100,000 closed ids, so an older one may be forgotten.

Parameters:

  • task_id (str) – The completed task's identifier.
  • agent_id (str | None) – Optional caller identity, checked without binding an empty engine.

kyvvu_engine.PolicyEngine.evaluate

evaluate(intended: Behavior, context: EvalContext) -> EvalResult

Evaluate all applicable policies against an intended step.

Reads completed steps and runner-registered open intents from the internal tracker. Evaluates every policy whose enforcement point is step_execution and whose agent/classification filters match context. Returns an aggregated result.

The intended step is not added to history here. The runner registers allowed intents; record settles their outcomes.

Parameters:

  • intended (Behavior) – The step the agent is about to execute.
  • context (EvalContext) – Agent identity and execution metadata.

Returns:

  • EvalResult – class:~kyvvu_engine.schemas.EvalResult with aggregate risk
  • EvalResult – score, recommended action, and per-policy outcomes.

kyvvu_engine.PolicyEngine.evaluate_and_record

evaluate_and_record(intended: Behavior, context: EvalContext, output: Any = None) -> EvalResult

Evaluate, then record the step, including a blocked one.

Convenience wrapper that combines evaluate and record. The recorded step carries the decision record in meta["kyvvu.eval"].

  • Not blocked: the step is recorded with output attached.
  • Blocked step.* behaviour: the step is recorded, in the shape blocked_step builds (output.status = "blocked"), and output is ignored (decision D1). The caller must not record it again.
  • Blocked task boundary (task.*): nothing is recorded. How a blocked boundary is represented is not part of the 1.0 contract.

Parameters:

  • intended (Behavior) – The step the agent is about to execute.
  • context (EvalContext) – Agent identity and execution metadata.
  • output (Any) – Optional output to attach to the recorded step.

Returns:

  • EvalResult – class:~kyvvu_engine.schemas.EvalResult from evaluate(),
  • EvalResult – with recorded_seq set to the recorded step's seq when one
  • EvalResult – was recorded.

Raises:

  • TaskClosedError – If the task is closing or closed.
  • ValueError – As record.

kyvvu_engine.PolicyEngine.evaluate_registration

evaluate_registration(agent_data: dict[str, Any], context: EvalContext) -> EvalResult

Evaluate agent_registration policies against agent metadata.

Called immediately after a successful register_agent() API call. Evaluates every policy whose enforcement point is "agent_registration" and whose agent/classification filters match context.

Parameters:

  • agent_data (dict[str, Any]) – Agent profile dict (fields from the registration response: purpose, name, risk_classification, …).
  • context (EvalContext) – Agent identity and classification metadata.

Returns:

  • EvalResult – class:~kyvvu_engine.schemas.EvalResult with aggregate risk
  • EvalResult – score, recommended action, and per-policy outcomes.

kyvvu_engine.PolicyEngine.explain

explain(intended: Behavior, context: EvalContext) -> str

Return a human-readable explanation of the evaluation result.

Runs evaluate internally and formats the per-policy results into a readable summary. Compound-rule failures include nested sub-condition results when violation_details contains sub_results.

Parameters:

  • intended (Behavior) – The step to explain.
  • context (EvalContext) – Agent identity and execution metadata.

Returns:

  • str – Multi-line explanation string.

kyvvu_engine.PolicyEngine.get_history

get_history(task_id: str) -> list[Behavior]

Return the completed steps for task_id, oldest first.

Returns an empty list if the task has no recorded steps or has already been evicted via end_task.

Parameters:

  • task_id (str) – The task whose history to retrieve.

Returns:

kyvvu_engine.PolicyEngine.get_manifests

get_manifests() -> list[dict[str, Any]]

Return the distinct manifest provenance records for loaded policies.

Delegates to get_manifests. Each record carries manifest_id and manifest_name; the caller (runner) merges the fuller origin fields from manifests when present.

kyvvu_engine.PolicyEngine.get_open_intents

get_open_intents(task_id: str) -> list[Behavior]

Return the task's still-unsettled runner preflights.

kyvvu_engine.PolicyEngine.is_task_closed

is_task_closed(task_id: str) -> bool

Whether task_id is closing (task.end/task.error recorded) or closed.

kyvvu_engine.PolicyEngine.load_policies

load_policies(policies: list[dict[str, Any]]) -> None

Replace the engine's policy set with policies.

Transport (fetching from the API, reading from a file) is the caller's responsibility. Calling this a second time replaces the first set entirely.

Every failure raises; nothing is dropped. An unregistered rule_type, a failed schema validation, a duplicate policy name within the set, params that do not match the rule's params_schema (a required key missing, a key the rule does not know, a value of the wrong type, params that is not a mapping), a regex that does not compile, and the retired scope key all refuse the load. The whole set is checked before anything changes, so a refused load leaves the previous set exactly as it was.

Parameters:

  • policies (list[dict[str, Any]]) – List of dicts matching the PolicyDefinition shape.

Raises:

  • PolicyLoadError – Naming every offending policy, its rule and the reason.

kyvvu_engine.PolicyEngine.load_report

load_report() -> PolicyLoadReport

Describe the outcome of the last load_policies call.

The engine reports only what it was given: it cannot know whether a fetch failed or whether a set came from a stale disk cache, so it never returns stale_cache. The transport layer (KyvvuRunner) overlays that on top of this report.

State is derived as:

  • never loaded → unavailable
  • at least one policy enforced → enforcing
  • nothing offered → none_configured

The numbers come from one snapshot under one lock, so a report never mixes a new policy set with the previous load's accounting.

Returns:

kyvvu_engine.PolicyEngine.policy_count

policy_count() -> int

Return the number of currently loaded policies.

kyvvu_engine.PolicyEngine.record

record(step: Behavior) -> Behavior

Record a completed step in the task's history.

Call this after a step executes (and, optionally, after its output has been captured). The engine settles a matching intent by action_id or appends the step directly for replay, and assigns seq. seq orders a task's recorded steps: strictly increasing per (agent_id, task_id), assigned by the recording client. Consumers must not assume it is contiguous, starts at a particular value, or reflects wall-clock order; a value supplied by the caller may be ignored.

The payload fields are re-checked as I-JSON here, because a step built with model_copy(update=...) skipped validation.

Parameters:

  • step (Behavior) – The completed Behavior. Set output before calling this to make output data available to history-aware rules.

Returns:

  • Behavior – The recorded Behavior with seq assigned.

Lifecycle: recording task.start opens the task, and a second task.start for it is refused. Recording task.end or task.error makes it closing: any later record or intent for it raises TaskClosedError, and only end_task may follow. (Unlike KyvvuRunner, the pure engine does not require a task.start before other steps.)

Raises:

  • ValueError – If a payload value is not I-JSON, the timestamp is naive, or the step's action_id was already recorded in the task (action_id is unique per (agent_id, task_id); completing an open intent is the one legal reuse).
  • TaskClosedError – If the task is closing or closed.
  • TaskStateError – If step is a second task.start.

kyvvu_engine.PolicyEngine.register_intent

register_intent(intended: Behavior) -> None

Expose an allowed runner preflight to later policy evaluations.

evaluate and explain never register intents themselves. The runner calls this after an allowed or warned step evaluation.

Raises:

  • ValueError – If the intent's action_id is already open or settled in its task; action_id is unique per (agent_id, task_id).
  • TaskClosedError – If the task is closing or closed.

kyvvu_engine.PolicyEngine.registration_policy_digest

registration_policy_digest() -> str

Return a digest of the loaded agent_registration policies.

Two loads with the same digest would evaluate a registration the same way, so a caller holding a registration that was already evaluated needs to evaluate it again only when this changes. See digest for what it covers.

Returns:

  • str – A hex SHA-256 digest.

kyvvu_engine.PolicyEngine.unresolved_paths

unresolved_paths(behaviors: list[Behavior]) -> list[UnresolvedPath]

Find step policy paths absent from every supplied behavior.

kyvvu_engine.PolicyEngine.validate_rule_params

validate_rule_params(rule_type: str, params: dict[str, Any], enforcement_point: str | None = None) -> tuple[bool, str | None]

Validate that rule_type is registered and params match its schema.

The one validator behind every load path: load_policies, manifest parsing, the API's manifest validator and the Python client's policy_file all reach it (directly, or through check_params, which this wraps). Regex patterns must also compile, matching the check made when policies are loaded.

Parameters:

  • rule_type (str) – Rule name to look up.
  • params (dict[str, Any]) – The policy's params.
  • enforcement_point (str | None) – View used to validate full field paths, when known.

Returns:

  • bool – (True, None) when valid; (False, error_message) otherwise,
  • str | None – the message listing every problem found.

kyvvu_engine.PolicyLoadError

PolicyLoadError(message: str, errors: list[str] | None = None) -> None

A policy, or a manifest of policies, was refused at load time.

Raised for a policy whose params do not match its rule's params_schema (a required key missing, a key the rule does not know, a value of the wrong type), and for the retired scope key. It is raised on every load path - PolicyEngine.load_policies, manifest parsing, the API manifest validator's assignment check and the Python client's policy_file - and never degraded to a cached or partial policy set: a configuration that loads fewer policies than it names enforces less than it claims.

Subclasses ValueError so a caller that already guarded manifest loading with except ValueError keeps working.

Attributes:

  • errors – One message per problem found, each naming the policy, the rule and the offending key. Loading collects every problem before raising, so an author fixes the file once rather than once per run.

Keep the overall failure message and its individual validation errors.

kyvvu_engine.PolicyLoadError.errors

errors = list(errors or [])

kyvvu_engine.PolicyLoadReport

The outcome of the most recent policy load, as one legible object.

offered=15, enforcing=0 is the shape that matters: it says a populated policy set produced zero enforcement, which a bare count cannot express.

Fields

state: Which PolicyLoadState the load ended in. offered: Number of policies handed to the loader. enforcing: Number of policies actually loaded and enforced.

Attributes:

kyvvu_engine.PolicyLoadReport.enforcing

enforcing: int = 0

kyvvu_engine.PolicyLoadReport.offered

offered: int = 0

kyvvu_engine.PolicyLoadReport.state

state: PolicyLoadState = PolicyLoadState.unavailable

kyvvu_engine.PolicyLoadState

Why the policy set currently in the engine looks the way it does.

A policy count cannot distinguish situations that demand very different responses: an agent with nothing assigned, an agent whose policy fetch failed, and an agent whose entire policy set was rejected as unloadable all report zero policies and all evaluate as allow. This enum is the missing distinction.

Values

enforcing: At least one policy is loaded and being enforced. none_configured: The load succeeded and the source genuinely has no policies. Nothing is being enforced, and that is the source's answer, not a failure. unavailable: Nothing is loaded. Either no load has been attempted yet, or the fetch failed and no cached set was available to fall back on. stale_cache: Policies are being enforced, but the set is not known to match its source: the last fetch failed (whether the set in effect came from memory or from a disk cache), or the disk cache that is the configured source has outlived policy_cache_max_age_seconds. Staleness is a property of the load, not of which source the set came from — the long-lived agent whose API went down keeps enforcing what it has, and this is the state that says so.

There is no "some policies were dropped" state: a policy that cannot be loaded raises PolicyLoadError out of the load, so a set is either loaded whole or refused whole.

Attributes:

kyvvu_engine.PolicyLoadState.enforcing

enforcing = 'enforcing'

kyvvu_engine.PolicyLoadState.none_configured

none_configured = 'none_configured'

kyvvu_engine.PolicyLoadState.stale_cache

stale_cache = 'stale_cache'

kyvvu_engine.PolicyLoadState.unavailable

unavailable = 'unavailable'

kyvvu_engine.PolicyResult

Per-policy evaluation outcome within a single EvalResult.

Fields

policy_id: ID of the evaluated policy. name: Human-readable policy name. severity: Policy severity level. violated: True if the rule function returned False or raised an exception. violation_details: Rule-specific details when violated; None when the policy passed. manifest_id: Assignment row id of the manifest this policy came from; None for hand-authored policies. manifest_name: Human-readable source manifest name, when available.

Attributes:

kyvvu_engine.PolicyResult.manifest_id

manifest_id: int | None = None

kyvvu_engine.PolicyResult.manifest_name

manifest_name: str | None = None

kyvvu_engine.PolicyResult.name

name: str

kyvvu_engine.PolicyResult.policy_id

policy_id: int | None

kyvvu_engine.PolicyResult.rule_type

rule_type: str = ''

kyvvu_engine.PolicyResult.severity

severity: str

kyvvu_engine.PolicyResult.violated

violated: bool

kyvvu_engine.PolicyResult.violation_details

violation_details: dict[str, Any] | None = None

kyvvu_engine.PolicyRule

Registry and decorator factory for policy rule functions.

Rule functions are registered at module import time via the register class-method decorator. The registry is a module-level singleton so all functions defined here are available across all engine instances.

Usage::

@PolicyRule.register(
    name="my_rule",
    description="...",
    params_schema={"max": {"type": "integer", "required": True}},
    enforcement_points=["step_execution"],
)
def check_my_rule(data, params, context):
    return data.kind == params["kind"]

Functions:

  • check_params – Return every way params fails to match rule_type's schema.
  • get_all_rules – Return metadata for all registered rules (without callables).
  • get_rule_function – Return the evaluation callable for rule_name.
  • register – Return a decorator that registers a rule function under name.
  • rule_exists – Return True if rule_name is present in the registry.
  • unknown_rule_hint – Say what the caller probably meant by an unregistered rule_name.

kyvvu_engine.PolicyRule.check_params

check_params(rule_type: str, params: Any, enforcement_point: str | None = None) -> list[str]

Return every way params fails to match rule_type's schema.

Checks, in this order: a required key is present, no key falls outside the schema, each value has the declared type and lies within any enum, minimum or maximum (None on an optional param counts as unset). An object param that declares properties gets the same checks on its own keys, recursively, reported under a dotted path such as trigger_condition.op; an array param that declares items has each element checked, reported as conditions[0]. Keys starting with an underscore are engine-internal (_policy_id is injected at evaluation) and are neither required nor checked. The compound rules all_of, any_of and not carry nested {rule_type, params} conditions, and each of those is checked the same way, recursively.

An unregistered rule_type yields no finding: whether to refuse one is a separate decision (rule_exists).

Parameters:

  • rule_type (str) – Registered rule name.
  • params (Any) – The policy's params; None is read as {}.
  • enforcement_point (str | None) – View used to validate full field paths, including those in nested conditions.

Returns:

  • list[str] – Messages without the policy's name, one per problem; empty when
  • list[str] – the params are valid.

kyvvu_engine.PolicyRule.get_all_rules

get_all_rules(enforcement_point: str | None = None) -> dict[str, dict[str, Any]]

Return metadata for all registered rules (without callables).

Parameters:

  • enforcement_point (str | None) – When provided, only rules whose enforcement_points include this value are returned.

Returns:

  • dict[str, dict[str, Any]] – Dict mapping rule name to metadata dict.

kyvvu_engine.PolicyRule.get_rule_function

get_rule_function(rule_name: str) -> Callable[[RuleData, dict[str, Any], RuleContext], bool]

Return the evaluation callable for rule_name.

Raises:

  • ValueError – When rule_name is not found in the registry.

kyvvu_engine.PolicyRule.register

register(name: str, description: str, params_schema: dict[str, dict[str, Any]], enforcement_points: list[str], example_params: dict[str, Any] | None = None) -> Callable[[Callable[[_View, dict[str, Any], RuleContext], bool]], Callable[[_View, dict[str, Any], RuleContext], bool]]

Return a decorator that registers a rule function under name.

kyvvu_engine.PolicyRule.rule_exists

rule_exists(rule_name: str) -> bool

Return True if rule_name is present in the registry.

kyvvu_engine.PolicyRule.unknown_rule_hint

unknown_rule_hint(rule_name: Any) -> str

Say what the caller probably meant by an unregistered rule_name.

The closest registered names when there are any (difflib, so a transposed or dropped letter finds its rule), otherwise every valid name, so the message never leaves the author with nothing to try.

Parameters:

  • rule_name (Any) – The name that is not in the registry.

Returns:

  • str – A sentence to append to an "unknown rule_type" message.

kyvvu_engine.PolicySourceKind

Where a configuration is able to obtain policies from.

This is a question about configuration, not about runtime state: it says which source a caller could in principle reach, not whether that source answered. Resolved by policy_source_kind and reported as source once a load has happened.

Values

api: The Kyvvu platform API (api_url + api_key + agent_key). local: Policies the caller supplied itself, with no credentials and no network: an inline list (policies) or one or more manifest YAML files on disk (policy_file). This is the credential-free source — it is what pip install to a visible block runs on, and it is resolved before disk_cache because a caller who names a policy file is naming a source, whereas a cache path names a place a fetch may once have written to. disk_cache: An on-disk policy cache (policy_cache_path), written by an earlier successful API fetch. none: No policy source at all — such a configuration can never enforce anything, which is why KyvvuRunner refuses to be constructed from it.

Attributes:

kyvvu_engine.PolicySourceKind.api

api = 'api'

kyvvu_engine.PolicySourceKind.disk_cache

disk_cache = 'disk_cache'

kyvvu_engine.PolicySourceKind.local

local = 'local'

kyvvu_engine.PolicySourceKind.none

none = 'none'

kyvvu_engine.PolicyStatus

Status of the runner's policy cache.

Returned by policy_status.

Fields

loaded: True if policies have been loaded at least once. last_success: Wall-clock time of last successful policy fetch. last_attempt: Wall-clock time of last fetch attempt (success or failure). policy_count: Number of active policies in the engine. stale: True if the last fetch failed and the cache has exceeded TTL. source: Where the current policies came from: "api", "local", "disk_cache", or "none". last_fetch_at: Alias for last_success (backward compatibility). last_fetch_succeeded: Whether the most recent fetch attempt succeeded. instance_id: The runner's instance identifier. ttl_remaining_seconds: Seconds until the in-memory cache expires and a re-fetch is triggered. load_state: Which PolicyLoadState the last load ended in. This is the field that tells "no policies assigned" apart from "fetch failed" and from "the source answered with nothing"; policy_count alone cannot. policies_offered: How many policies were handed to the loader. Read together with policy_count; they differ only transiently, because a policy that cannot be loaded refuses the whole load. registration_blocked_by: Names of the agent_registration policies whose violation blocked the agent's registration when it was last evaluated, at register_agent() or again after a refresh changed the registration policies. Empty when it is not blocked. registration_blocked: Whether the registration is blocked. While it is, every step.* evaluation is refused with KyvvuBlockedError naming those policies; task boundaries are not. Cleared by a later evaluation of the registration that does not block. governed: Whether at least one validated policy is in effect. stale_cause: Why the set in effect is not known to match its source, while it is not; None when it does. client_upgrade_required is the one that means pip install -U. client_upgrade: What the platform said when it last refused this client as too old, or None. Cleared by the next accepted fetch. agent_id: The agent this runner governs: the platform agent_id the policy set was served for, when the platform is the source and has said; otherwise the agent_id the runner bound to on its first evaluation or record; None before either. With the platform as the source this is the agent_id a caller must send, and any other is refused with AgentBindingError. applicable_policy_count: How many of the loaded policies apply to agent_id: the workspace-wide ones plus those naming that agent, counted the way policy_count is. None while agent_id is. Equal to policy_count for the agent a set was fetched for; lower means some loaded policies never apply to this agent.

Attributes:

kyvvu_engine.PolicyStatus.agent_id

agent_id: str | None = None

kyvvu_engine.PolicyStatus.applicable_policy_count

applicable_policy_count: int | None = None

kyvvu_engine.PolicyStatus.client_upgrade

client_upgrade: dict[str, Any] | None = None

kyvvu_engine.PolicyStatus.governed

governed: bool

Whether the current load state has policies in effect.

kyvvu_engine.PolicyStatus.instance_id

instance_id: str = ''

kyvvu_engine.PolicyStatus.last_attempt

last_attempt: datetime | None = None

kyvvu_engine.PolicyStatus.last_fetch_at

last_fetch_at: datetime | None = None

kyvvu_engine.PolicyStatus.last_fetch_succeeded

last_fetch_succeeded: bool = False

kyvvu_engine.PolicyStatus.last_success

last_success: datetime | None = None

kyvvu_engine.PolicyStatus.load_state

load_state: PolicyLoadState = PolicyLoadState.unavailable

kyvvu_engine.PolicyStatus.loaded

loaded: bool = False

kyvvu_engine.PolicyStatus.policies_offered

policies_offered: int = 0

kyvvu_engine.PolicyStatus.policy_count

policy_count: int = 0

kyvvu_engine.PolicyStatus.registration_blocked

registration_blocked: bool

Whether the agent's registration is blocked by a policy.

kyvvu_engine.PolicyStatus.registration_blocked_by

registration_blocked_by: list[str] = Field(default_factory=list)

kyvvu_engine.PolicyStatus.source

source: str = 'none'

kyvvu_engine.PolicyStatus.stale

stale: bool = False

kyvvu_engine.PolicyStatus.stale_cause

stale_cause: StaleCause | None = None

kyvvu_engine.PolicyStatus.ttl_remaining_seconds

ttl_remaining_seconds: float = 0.0

kyvvu_engine.RegistrationView

RegistrationView(*, agent_id: str | None = None, agent_key: str | None = None, name: str | None = None, purpose: str | None = None, description: str | None = None, owner_id: str | None = None, maintainer_id: str | None = None, risk_classification: str | None = None, environment: str | None = None, declared_tools: list[str] | None = None, metadata: dict[str, Any] | None = None) -> None

The known agent fields visible to registration rules.

Functions:

  • from_agent_data – Select known fields from an agent response, ignoring extras.

Attributes:

kyvvu_engine.RegistrationView.agent_id

agent_id: str | None = None

kyvvu_engine.RegistrationView.agent_key

agent_key: str | None = None

kyvvu_engine.RegistrationView.declared_tools

declared_tools: list[str] | None = None

kyvvu_engine.RegistrationView.description

description: str | None = None

kyvvu_engine.RegistrationView.environment

environment: str | None = None

kyvvu_engine.RegistrationView.from_agent_data

from_agent_data(agent_data: dict[str, Any]) -> RegistrationView

Select known fields from an agent response, ignoring extras.

kyvvu_engine.RegistrationView.maintainer_id

maintainer_id: str | None = None

kyvvu_engine.RegistrationView.metadata

metadata: dict[str, Any] | None = None

kyvvu_engine.RegistrationView.name

name: str | None = None

kyvvu_engine.RegistrationView.owner_id

owner_id: str | None = None

kyvvu_engine.RegistrationView.purpose

purpose: str | None = None

kyvvu_engine.RegistrationView.risk_classification

risk_classification: str | None = None

kyvvu_engine.ReportStatus

ReportStatus(*, queued: int, dropped: int, failed: int, delivered: int, capacity: int) -> None

A snapshot of the outbound report queue.

Keyword-only, so a later release can add a counter without breaking a caller; read the fields by name.

Attributes:

  • queued (int) – Reports waiting, plus the one being sent, right now.
  • dropped (int) – Reports refused because the queue was full, since start.
  • failed (int) – Reports whose delivery raised, or whose sink returned False (an exporter that caught and logged its own error and reported the batch as not delivered).
  • delivered (int) – Reports handed to their sink without an exception and without a False return.
  • capacity (int) – The queue bound.

kyvvu_engine.ReportStatus.capacity

capacity: int

kyvvu_engine.ReportStatus.delivered

delivered: int

kyvvu_engine.ReportStatus.dropped

dropped: int

kyvvu_engine.ReportStatus.failed

failed: int

kyvvu_engine.ReportStatus.queued

queued: int

kyvvu_engine.RuleContext

RuleContext(context: EvalContext | None = None, enforcement_point: str = '', agent_id: str | None = None, task_id: str | None = None, current_step_id: int | None = None, task_history: list[Any] | None = None, agent: Any | None = None, user_id: int | None = None, policy_store: Any | None = None, now: datetime | None = None) -> None

Context object passed to every rule evaluation function.

Provides access to the task's step history (pre-fetched from the engine's internal PathTracker) and agent metadata (pre-fetched from EvalContext).

No I/O occurs inside this class. All data is supplied at construction.

Attributes:

  • enforcement_point – When this evaluation runs ("step_execution" | "agent_registration").
  • agent_id – String identifier of the agent being evaluated.
  • task_id – Identifier of the current task/execution run.
  • current_step_id – Integer DB id of the step being evaluated (informational).
  • now (datetime | None) – The evaluation time, in UTC: the timestamp of the behaviour being evaluated, which the engine passes in. None when there is no behaviour, as at agent_registration, where no registered rule reads the time. This class never reads the wall clock, so a verdict replays from the recorded behaviour alone.
  • hour (int | None) – UTC hour of now, or None when now is.

Functions:

Build a context from pre-fetched data; performs no I/O.

Parameters:

  • context (EvalContext | None) – Agent identity and execution metadata, when known.
  • enforcement_point (str) – The enforcement point being evaluated.
  • agent_id (str | None) – Agent identifier, used when context is None.
  • task_id (str | None) – Task identifier, used when context is None.
  • current_step_id (int | None) – Informational id of the step being evaluated.
  • task_history (list[Any] | None) – The task's rule history, oldest first.
  • agent (Any | None) – A pre-seeded agent record, if any.
  • user_id (int | None) – Owning user id, used when context is None.
  • policy_store (Any | None) – Store holding pre-compiled policy parameters.
  • now (datetime | None) – The evaluated behaviour's timestamp. Must be timezone-aware; it is normalised to UTC. None leaves the context without an evaluation time.

kyvvu_engine.RuleContext.agent_id

agent_id = context.agent_id if context else agent_id

kyvvu_engine.RuleContext.compiled_params

compiled_params(policy_id: int | None) -> dict[str, Any] | None

Return pre-compiled regex data for policy_id, or None on miss.

kyvvu_engine.RuleContext.count_steps_of_kind

count_steps_of_kind(kind: str) -> int

Count completed steps of the given kind.

Parameters:

  • kind (str) – The action kind to count (e.g. "step.model").

Returns:

  • int – Count of matching steps in history.

kyvvu_engine.RuleContext.current_step_id

current_step_id = current_step_id

kyvvu_engine.RuleContext.enforcement_point

enforcement_point = enforcement_point

kyvvu_engine.RuleContext.get_all_steps_in_task

get_all_steps_in_task() -> list[Any]

Return the full ordered step history for the current task.

kyvvu_engine.RuleContext.get_current_agent

get_current_agent() -> Any | None

Return the agent record for the current evaluation.

Prefers the pre-seeded agent argument; falls back to constructing an AgentRecord from EvalContext.

Returns:

  • Any | None – class:~kyvvu_engine.schemas.AgentRecord or None.

kyvvu_engine.RuleContext.get_previous_step

get_previous_step(required_kind: str | None = None) -> Any | None

Return the most recent step in task history that happened, or None.

"Happened" is _happened: open intents (no step number yet) and blocked, failed or abandoned outcomes are attempts, and are looked past. The last history entry is therefore not necessarily the previous step.

Parameters:

  • required_kind (str | None) – The kind the calling rule requires of the step it is looking for, if any. When it is step.gate, a gate that did not pass (pending, denied, or with no decision) is looked past as well, as _counts_for defines.

Returns:

  • Any | None – The matching history entry, or None when no step happened.

kyvvu_engine.RuleContext.hour

hour: int | None = self.now.hour if self.now is not None else None

kyvvu_engine.RuleContext.last_violation_details

last_violation_details: dict[str, Any] | None = None

kyvvu_engine.RuleContext.now

now: datetime | None = now.astimezone(timezone.utc) if now is not None else None

kyvvu_engine.RuleContext.project_root

project_root: str | None

Return the session/project root from EvalContext, or None.

Populated by integrations that run inside a project tree (e.g. kyvvu-claude sets it from the session cwd). Consumed by allowlist-containment rules such as path_within_root. None when no project root is known.

kyvvu_engine.RuleContext.task_id

task_id = context.task_id if context else task_id

kyvvu_engine.RuleView

RuleView(*, kind: str, verb: str | None, step_name: str, input: dict[str, Any] | None, properties: dict[str, Any]) -> None

The intended step fields visible to step-execution rules.

Functions:

  • from_behavior – Select the intended step fields visible to rules.

Attributes:

kyvvu_engine.RuleView.from_behavior

from_behavior(behavior: Behavior) -> RuleView

Select the intended step fields visible to rules.

kyvvu_engine.RuleView.input

input: dict[str, Any] | None

kyvvu_engine.RuleView.kind

kind: str

kyvvu_engine.RuleView.properties

properties: dict[str, Any]

kyvvu_engine.RuleView.step_name

step_name: str

kyvvu_engine.RuleView.verb

verb: str | None

kyvvu_engine.StaleCause

Why the set in effect is not known to match its source.

PolicyLoadState says that a set is stale; this says why, because the remedies differ: a network problem, a policy definition this engine refuses, and an engine the platform no longer serves are three different fixes, and only the last is pip install -U.

Values

fetch_failed: The last fetch did not complete (network, timeout, an HTTP error). set_refused: The source answered with a set this engine refuses to load (PolicyLoadError). client_upgrade_required: The platform refused this client as too old to serve (KyvvuClientUpgradeRequired). The set in effect is the last one fetched before that. cache_expired: The configured disk cache has outlived policy_cache_max_age_seconds. local_source_unreadable: The configured manifest could not be read. caller_marked: A caller of load_policies passed stale=True.

Attributes:

kyvvu_engine.StaleCause.cache_expired

cache_expired = 'cache_expired'

kyvvu_engine.StaleCause.caller_marked

caller_marked = 'caller_marked'

kyvvu_engine.StaleCause.client_upgrade_required

client_upgrade_required = 'client_upgrade_required'

kyvvu_engine.StaleCause.fetch_failed

fetch_failed = 'fetch_failed'

kyvvu_engine.StaleCause.local_source_unreadable

local_source_unreadable = 'local_source_unreadable'

kyvvu_engine.StaleCause.set_refused

set_refused = 'set_refused'

kyvvu_engine.TASK_BOUNDARY_KINDS

TASK_BOUNDARY_KINDS: frozenset[ActionKind] = frozenset({ActionKind.task_start, ActionKind.task_end, ActionKind.task_error, ActionKind.task_idle})

kyvvu_engine.TaskClosedError

Raised when a write names a task that is closing or closed.

A task is closing once its task.end or task.error is recorded: only end_task(), which finishes cleanup, may follow. It is closed by end_task(), or evicted by the stale-task sweeper. Within one runner (or PolicyEngine) a closed task_id is not reopened, for its most recent 100,000 closed tasks: a second record under the same identifier would restart seq and be indistinguishable from the first. Older ids may be forgotten, and nothing is claimed across processes; a task_id must still never be reused.

kyvvu_engine.TaskStateError

Raised when a write does not fit the task lifecycle.

A task is opened by recording its task.start. A KyvvuRunner refuses any other write (evaluate, record, restore_intent) into a task it has not opened, and refuses a second task.start for a task it has already opened. TaskClosedError is the subclass for a write into a task that is closing or closed.

kyvvu_engine.UnresolvedPath

UnresolvedPath(*, policy_name: str, param: str, path: str) -> None

A policy path absent from every inspected step.

Keyword-only, so a later release can add a field additively.

Attributes:

kyvvu_engine.UnresolvedPath.param

param: str

kyvvu_engine.UnresolvedPath.path

path: str

kyvvu_engine.UnresolvedPath.policy_name

policy_name: str

kyvvu_engine.VALID_COMBINATIONS

VALID_COMBINATIONS: set[tuple[ActionKind, Verb | None]] = {(ActionKind.task_start, None), (ActionKind.task_end, None), (ActionKind.task_error, None), (ActionKind.task_idle, None), (ActionKind.step_resource, Verb.GET), (ActionKind.step_resource, Verb.POST), (ActionKind.step_resource, Verb.PATCH), (ActionKind.step_resource, Verb.DELETE), (ActionKind.step_message, Verb.GET), (ActionKind.step_message, Verb.POST), (ActionKind.step_self, Verb.GET), (ActionKind.step_self, Verb.POST), (ActionKind.step_self, Verb.PATCH), (ActionKind.step_self, Verb.DELETE), (ActionKind.step_model, None), (ActionKind.step_credential, Verb.GET), (ActionKind.step_credential, Verb.POST), (ActionKind.step_credential, Verb.PATCH), (ActionKind.step_credential, Verb.DELETE), (ActionKind.step_exec, None), (ActionKind.step_gate, None), (ActionKind.step_unknown, None)}

kyvvu_engine.Verb

HTTP-style verb describing the direction and mutation intent of a step.

Attributes:

kyvvu_engine.Verb.DELETE

DELETE = 'DELETE'

kyvvu_engine.Verb.GET

GET = 'GET'

kyvvu_engine.Verb.PATCH

PATCH = 'PATCH'

kyvvu_engine.Verb.POST

POST = 'POST'

kyvvu_engine.__version__

Engine release version.

kyvvu_engine.aggregate_max

aggregate_max(results: list[PolicyResult]) -> EvalResult

Aggregate by taking the maximum severity among all violations.

Action thresholds: - risk_score == 0.0 → allow - 0.0 < risk_score < 1.0 → warn - risk_score == 1.0 → block

Parameters:

  • results (list[PolicyResult]) – Per-policy evaluation outcomes.

Returns:

  • EvalResult – class:EvalResult with risk_score and action from worst violation.

kyvvu_engine.aggregate_mean

aggregate_mean(results: list[PolicyResult]) -> EvalResult

Aggregate by computing the mean severity across all violations.

Parameters:

  • results (list[PolicyResult]) – Per-policy evaluation outcomes.

Returns:

  • EvalResult – class:EvalResult with mean risk score.

kyvvu_engine.aggregate_weighted_sum

aggregate_weighted_sum(results: list[PolicyResult], weights: dict[int | None, float]) -> EvalResult

Aggregate using caller-supplied per-policy weights.

Parameters:

  • results (list[PolicyResult]) – Per-policy evaluation outcomes.
  • weights (dict[int | None, float]) – Dict mapping policy_id to a numeric weight.

Returns:

  • EvalResult – class:EvalResult with weighted sum capped at 1.0.

kyvvu_engine.blocked_step

blocked_step(intended: Behavior, result: EvalResult) -> Behavior

Return intended in the one shape a blocked step is recorded in.

The step keeps every field of intended and gains:

  • action_id: the evaluation's (result.action_id), falling back to the intended step's own;
  • output: {"status": "blocked", "error": <message>}, plus "decision": "deny" on a step.gate (a denied gate never satisfies step_requires_gate). The message is the KyvvuBlockedError text, cut at 2000 characters.

KyvvuRunner.evaluate() and PolicyEngine.evaluate_and_record() record this themselves, with meta["kyvvu.eval"]; a caller uses the builder only to keep its own copy of the step (a session file, say) in the same shape. It does not record anything.

Parameters:

  • intended (Behavior) – The blocked step.* behaviour.
  • result (EvalResult) – Its blocking evaluation.

Returns:

Raises:

  • ValueError – If intended is a task boundary. How a blocked boundary is represented is not part of the 1.0 contract.

kyvvu_engine.build_exporter_from_settings

build_exporter_from_settings(settings: Any, instance_id: str = '', platform: PlatformClient | None = None) -> list[Exporter]

Build the exporter list from log_location / log_format.

Reads the resolved location and format from the settings object's effective_log_location and effective_log_format properties.

Parameters:

  • settings (Any) – A KyvvuSettings instance.
  • instance_id (str) – The runner's instance ID (passed to HttpExporter).
  • platform (PlatformClient | None) – The platform client already owned by the runner, if any.

Returns:

  • list[Exporter] – List of exporter instances (0 or 1 elements).

kyvvu_engine.build_step_dicts

build_step_dicts(steps: Iterable[Behavior | Mapping[str, Any]], *, redaction: StepRedaction = 'full') -> list[dict[str, Any]]

Build wire actions from live behaviors or persisted session history.

metadata_only keeps the Python client's key/count summaries. sha256 keeps Claude's byte-count/hash summaries (SHA-256 over RFC 8785 canonical JSON) and only structural properties. In both, the payload-derived values in meta["kyvvu.eval"] violation_details are redacted the same way: dropped ({"redacted": true}) or replaced by their digest, keeping the keys that say which rule and field failed.

timestamp is always UTC with microsecond precision and a Z suffix, for example 2026-10-14T10:00:01.000000Z.

A mapping (a persisted step) must carry kind, step_name, seq and timestamp; a missing one raises ValueError instead of being defaulted. Its agent_id and task_id are optional, because the wire action does not carry them.

Raises:

  • ValueError – If a step is not I-JSON, a mapping lacks a required key, or redaction is unknown.

kyvvu_engine.capture_dict

capture_dict(value: dict[str, Any] | None) -> dict[str, Any] | None

Coerce a captured payload dict (input / output); None stays None.

kyvvu_engine.capture_step_name

capture_step_name(value: Any, fallback: str = 'unnamed') -> str

Return a valid step_name derived from a framework-supplied name.

Control characters (U+0000-U+001F, U+007F-U+009F), surrogates and noncharacters are replaced by U+FFFD; a name longer than 255 characters is cut to 254 and ends in "…"; an empty name becomes fallback. The same input always gives the same output, and a valid name is returned unchanged.

Parameters:

  • value (Any) – A function, tool or node name (anything; non-strings use str()).
  • fallback (str) – The name used when value is empty.

Returns:

  • str – A step_name the engine accepts.

kyvvu_engine.chain_entry_hash

chain_entry_hash(agent_id: str, task_id: str, seq: int, input_data: Mapping[str, Any] | None, previous_hash: str | None = None, *, ordinal_key: Literal['seq', 'step'] = 'seq') -> str

Return the v1 chain hash of one log entry.

Parameters:

  • agent_id (str) – The platform's agent identifier (the batch's agent_id).
  • task_id (str) – Task identifier.
  • seq (int) – The entry's ordinal within its (agent_id, task_id) chain.
  • input_data (Mapping[str, Any] | None) – The entry's input as exported; None is hashed as {}, which is also what an absent input is stored as for hashing.
  • previous_hash (str | None) – The previous entry's hash. None (or empty) for the first entry, whose hash input then has no previous_hash key.
  • ordinal_key (Literal['seq', 'step']) – The name of the ordinal in the hash input. Always "seq" for a new entry; "step" only to validate entries the platform stored before migration 0054 renamed it.

Returns:

  • str – 64 lowercase hexadecimal characters (SHA-256).

kyvvu_engine.chain_steps

chain_steps(agent_id: str, task_id: str, steps: Sequence[Mapping[str, Any]], head: ChainHead | None = None) -> tuple[list[dict[str, Any]], ChainHead]

Number and hash exported steps so they extend head.

The steps are wire actions as exported (build_step_dicts output, after redaction). Each copy gets seq = the previous seq + 1, starting at head.seq + 1, so the exported seq is contiguous over exactly the steps exported, whatever the steps carried; a step whose own seq differs is logged, because the producers number their recorded steps contiguously and a difference means one was not exported. Each copy then gets hash from chain_entry_hash.

head is not changed. Commit the returned head only once the sink has accepted the batch: a retry of a batch that was not acknowledged must be chained from the same head, or it is a different batch.

Parameters:

  • agent_id (str) – The platform's agent identifier.
  • task_id (str) – Task identifier.
  • steps (Sequence[Mapping[str, Any]]) – The wire actions to chain, in order.
  • head (ChainHead | None) – The chain so far; None or ChainHead() starts a task.

Returns:

  • tuple[list[dict[str, Any]], ChainHead] – The chained copies, and the head after the last of them.

Raises:

  • ValueError – If a step's input is not I-JSON.

kyvvu_engine.check_action_id

check_action_id(value: object, field: str = 'action_id') -> str

Return value if it is a valid action_id.

Raises:

  • ValueError – Naming the field and the rule.

kyvvu_engine.check_step_name

check_step_name(value: object, field: str = 'step_name') -> str

Return value if it is a valid step_name.

Raises:

  • ValueError – Naming the field and the rule.

kyvvu_engine.check_task_id

check_task_id(value: object, field: str = 'task_id') -> str

Return value if it is a valid task_id (or parent_task_id).

Raises:

  • ValueError – Naming the field and the rule.

kyvvu_engine.client_headers

client_headers() -> dict[str, str]

The headers that announce this client to the platform.

Returns:

  • dict[str, str] – X-Kyvvu-Engine-Version, X-Kyvvu-Policy-Schema and
  • dict[str, str] – X-Kyvvu-Client always; X-Kyvvu-SDK-Version when the Python
  • dict[str, str] – client is present.

kyvvu_engine.compliant_action_id

compliant_action_id(raw: Any) -> str

Map a framework-supplied id to a valid action_id, like compliant_task_id.

kyvvu_engine.compliant_task_id

compliant_task_id(raw: Any) -> str

Map a framework-supplied id to a valid task_id, deterministically.

A valid id is returned unchanged. Anything else (a LangGraph thread id with a /, an over-long run id, ...) becomes "x-" followed by the first 32 hex digits of the SHA-256 of its UTF-8 text. Never raises; keep the raw value in properties where it carries meaning.

Parameters:

  • raw (Any) – The framework's identifier (non-strings use str()).

Returns:

  • str – A task_id the engine accepts.

kyvvu_engine.load_manifest_policies

load_manifest_policies(paths: str | Path | list[str | Path]) -> list[dict[str, Any]]

Load one or more manifests into policy dicts, refusing an invalid one.

The convenience form for every caller that wants policies rather than a validation report: the local policy source on KyvvuSettings, kyvvu try, and kyvvu serve --policy-file.

Parameters:

  • paths (str | Path | list[str | Path]) – A single path, a comma-separated list of paths (see PATH_LIST_SEPARATOR), or a list of either — every string element is comma-split, so the two spellings behave identically. Policies are returned in the order the manifests are given, so a later manifest never displaces an earlier one — every entry is loaded and the engine evaluates all of them.

Returns:

  • list[dict[str, Any]] – Policy dicts ready for
  • list[dict[str, Any]] – meth:~kyvvu_engine.engine.PolicyEngine.load_policies, each carrying
  • list[dict[str, Any]] – its manifest's name as manifest_name.

Raises:

  • PolicyLoadError – When a policy's params do not match its rule's schema or it carries the retired scope key. Never a ManifestError, so a caller that degrades around an unusable file does not degrade around this.
  • ManifestError – When a file is unreadable, or when a manifest is otherwise invalid. Refusing an invalid manifest rather than loading the entries that happened to parse is deliberate: a partially loaded policy set is a failure mode where a configuration looks populated and enforces less than it claims.

kyvvu_engine.manifest_schema

manifest_schema() -> dict[str, Any]

Return an editor schema for this installed engine's manifest contract.

Rule names and parameter schemas come from the current rule registry; policy fields and categorical values come from the engine declarations. No settings, credentials or network access are needed.

The schema is an editor aid. Use parse_manifest and PolicyEngine.load_policies for authoritative validation, including unique policy names and executable regex constraints.

Returns:

  • dict[str, Any] – A fresh JSON Schema mapping, independent of the rule registry and
  • dict[str, Any] – previous results. Mutating it cannot change policy enforcement.

kyvvu_engine.parse_manifest

parse_manifest(yaml_content: str) -> ManifestParseResult

Parse and validate a manifest YAML string. Pure: no I/O, no network.

Validation rules:

  1. The YAML must parse.
  2. Required top-level keys: name (a string) and policies (a list).
  3. Optional top-level keys: description, version.
  4. Each policy entry requires name, rule_type, severity and enforcement_point. The retired scope key is an error that names its replacement.
  5. Each policy entry may carry description, params and risk_classification.
  6. params must match the rule's params_schema: required keys present, no unknown keys, values of the declared type.
  7. rule_type must be registered in the engine's rule registry, and policy name must be unique within the manifest. Both are load errors: the engine would refuse the same file.

Parameters:

  • yaml_content (str) – Raw YAML text.

Returns:

  • A (ManifestParseResult) – class:ManifestParseResult. Never raises for bad content — a
  • ManifestParseResult – malformed manifest comes back with valid=False and populated
  • ManifestParseResult – errors.

kyvvu_engine.parse_refusal

parse_refusal(status: int, body: bytes | str | None) -> ClientUpgradeInfo | None

Recognise the platform's version refusal in an HTTP error response.

Two shapes are recognised: 426 with code: client_upgrade_required at the top level (the policy fetch and registration), and 400 with code: unsupported_wire_version nested under detail (the log batch and incident endpoints). A bare 426 with an unreadable body is still a refusal: the status alone means "upgrade", and saying nothing because the body was malformed would be the generic failure this replaces.

Parameters:

  • status (int) – The HTTP status code.
  • body (bytes | str | None) – The raw response body.

Returns:

  • ClientUpgradeInfo | None – The refusal, or None when the response is something else.

kyvvu_engine.parse_ttl

parse_ttl(value: str | None) -> float | None

Parse seconds or a duration ending in s, m, h, or d; empty means no expiry.

kyvvu_engine.placeholder_warnings

placeholder_warnings(policies: Iterable[tuple[str, Any]]) -> list[str]

Name RFC 2606 example hosts found in policy parameters.

kyvvu_engine.register_client

register_client(name: str, version: str, *, sdk_version: str | None = None) -> None

Name the integration this process is, for X-Kyvvu-Client.

The Python client registers kyvvu/<version> when it is imported; an integration built on it (kyvvu-langchain, kyvvu-crewai) registers itself afterwards and wins, because the integration package is the one a person would upgrade. kyvvu-claude, which uses the engine without the Python client, registers itself.

Parameters:

  • name (str) – The distribution name, such as kyvvu-claude.
  • version (str) – Its installed version.
  • sdk_version (str | None) – The Python client's version, when it is present. Left as it is when None, so an integration does not erase what the Python client set.

kyvvu_engine.setup_logging

setup_logging(level: str | None = None, json: bool = False, force: bool = False) -> None

Configure the root logger.

Parameters:

  • level (str | None) – Log level string ("DEBUG", "INFO", "WARNING", "ERROR"). Defaults to the KV_LOG_LEVEL environment variable, or "INFO" if that is not set.
  • json (bool) – Emit JSON-formatted lines when True. Defaults to False (human-readable plaintext).
  • force (bool) – Re-configure even if setup_logging() was already called. Useful in tests that need a specific level.

kyvvu_engine.severity_effect

severity_effect(severity: str) -> Action

Return the default aggregator's action for one violated severity.

kyvvu_engine.split_path_list

split_path_list(paths: str) -> list[str]

Split a comma-separated manifest path list, dropping empty entries.

Parameters:

Returns:

  • list[str] – The non-empty, whitespace-stripped paths, in the order given.

kyvvu_engine.to_ijson_capture

to_ijson_capture(value: Any, *, truncate: bool = False, max_string_length: int = MAX_STRING_LENGTH, max_collection_items: int = MAX_COLLECTION_ITEMS, max_dict_items: int = MAX_DICT_ITEMS, max_recursion_depth: int = MAX_RECURSION_DEPTH, max_repr_length: int = MAX_REPR_LENGTH) -> Any

Return value as I-JSON data. Never raises.

Parameters:

  • value (Any) – Any Python value captured from the host.
  • truncate (bool) – Cut long strings, collections and deep nesting with markers. When false, only the fallback representation and a 64-level cycle guard are limited.
  • max_string_length (int) – Maximum string length when truncate is true.
  • max_collection_items (int) – Maximum list, tuple or set items when truncated.
  • max_dict_items (int) – Maximum dictionary entries when truncated.
  • max_recursion_depth (int) – Maximum nesting depth when truncated.
  • max_repr_length (int) – Maximum fallback repr() length in either mode.

Returns:

  • Any – A value built only from dict (str keys), list, str,
  • Any – bool, None, int within +-(2**53 - 1) and finite float.

kyvvu_engine.upgrade_message

upgrade_message(info: ClientUpgradeInfo, fetched_at: datetime | None, *, enforcing: int | None = None, source: str | None = None) -> str

The one line that tells a person what happened and what to do.

The middle clause says what is still in effect, and there are exactly three answers: the platform's own set -- held in memory from the last fetch, or read back from the on-disk policy cache -- a configured inline set, or nothing. Each state gets one of them and never two, so a caller that does not yet know what it will hold must not print this line before it does.

Parameters:

  • info (ClientUpgradeInfo) – The refusal.
  • fetched_at (datetime | None) – When the set still being enforced was fetched, or None when that is not known (or nothing is being enforced).
  • enforcing (int | None) – How many policies are in effect. 0 means nothing is, whatever fetched_at says; None means the caller did not say.
  • source (str | None) – Where the set in effect came from, as source names it -- api or disk_cache for the platform's set, local for a configured inline set. None keeps the previous inference: with no fetch time, a non-zero count is an inline set.

Returns:

  • str – `kyvvu 0.23.0 (kyvvu-engine 0.12.0) is too old for this platform
  • str – (needs kyvvu-engine>=0.13.0); still enforcing the policies fetched at
  • str – ; pip install -U ...`

kyvvu_engine.valid_verbs_hint

valid_verbs_hint(kind: ActionKind) -> str

Say which verbs kind admits, derived from VALID_COMBINATIONS.

Parameters:

  • kind (ActionKind) – The action kind that was given a verb it does not admit.

Returns:

  • str – A sentence such as "step.resource takes: DELETE, GET, PATCH, POST." or
  • str – "step.exec takes no verb (omit it).".

kyvvu_engine.warn_upgrade_required

warn_upgrade_required(message: str, *, signature: str | None = None) -> bool

Log message as a WARNING, at most once per hour per signature.

Never raises: it is called from exporters and the policy refresh, neither of which may fail because a log handler did.

Parameters:

  • message (str) – The line to log.
  • signature (str | None) – What makes two warnings the same one. Defaults to the message; the policy path passes a stable key, because its message names a timestamp that would otherwise make every line unique.

Returns:

  • bool – Whether the warning was emitted this time.