The engine contract¶

What the engine does, what it returns, and what you may rely on when you build on it. The public interface and its versioning are in section 8.

1. What the engine does¶

The engine governs an AI agent one task at a time. Before the agent acts, it evaluates the intended behaviour against a policy set and the history of the task so far, and returns a verdict: allow, warn or block. It records what the agent then does, and sends those records to the sinks you configure. Behaviours follow the AAG vocabulary of action kinds and verbs, so the same policy reads a LangChain tool call, a Claude Code hook and a decorated Python function alike. The engine runs in the agent's own process, or beside it as kyvvu serve.

2. Concepts¶

Agent: the program being governed, identified by agent_id.
Task: one run of an agent, identified by task_id, from its task.start to its task.end or task.error.
Behaviour (Behavior): one event in a task. Its kind says what it is (the task.* kinds mark task boundaries, the step.* kinds describe what the agent does), and a step.* carries a verb, its properties, its input and, once it has completed, its output. A step with no output is intended rather than done.
Evaluation context (EvalContext): the agent identity a behaviour is evaluated for.
Policy: a rule with parameters and a severity. A policy set is loaded whole or not at all.
Verdict: the action evaluation recommends, allow, warn or block, with the outcome of each policy behind it.

3. Running a governed agent¶

A governed agent meets the engine at five points: it registers, opens a task, has each step evaluated before it acts and recorded after, and ends the task. Each point is shown below in Python, with a KyvvuRunner in the agent's own process, and then as the request a kyvvu serve sidecar takes for it (the serve HTTP API). The Python is one program, which the documentation check runs against a stand-in platform and then offline.

Register the agent¶

PlatformClient.register_agent() registers the agent with the platform and returns the agent_id the platform assigned. A KyvvuRunner built from the same settings fetches that agent's policies, and evaluate_registration() evaluates its agent_registration policies, raising KyvvuBlockedError on a block.

import os
import uuid

from kyvvu_engine import (
    ActionKind,
    AgentProfile,
    Behavior,
    EvalContext,
    KyvvuBlockedError,
    KyvvuRunner,
    KyvvuSettings,
    PlatformClient,
    Verb,
)

TOOLS = ["read_faq", "run_script"]
settings = KyvvuSettings(api_url="https://platform.kyvvu.com", api_key=os.environ["KV_API_KEY"],
                         agent_key="support-agent")
registration = PlatformClient(settings).register_agent(AgentProfile(
    name="Support agent", purpose="Answer customer questions about orders",
    owner_id="ops@example.com", declared_tools=TOOLS),
    risk_classification="minimal")
runner = KyvvuRunner(settings=settings)
context = EvalContext(agent_id=registration.agent_id, environment=settings.environment,
                      risk_classification="minimal", agent_allowed_tools=TOOLS)
runner.evaluate_registration(registration.evaluation_data, context)  # a block raises

A kyvvu serve started against the platform learns the agent_id from the platform's policy answer and prints it; send it as agent_id in every request. /register_agent evaluates the registration policies:

POST http://127.0.0.1:8080/register_agent
Content-Type: application/json

{
  "agent_data": {
    "name": "Support agent",
    "purpose": "Answer customer questions about orders",
    "owner_id": "ops@example.com",
    "declared_tools": ["read_faq", "run_script"],
    "risk_classification": "minimal"
  },
  "context": {
    "agent_id": "agent_3f9c2a7b1d4e6f80",
    "risk_classification": "minimal",
    "agent_allowed_tools": ["read_faq", "run_script"]
  }
}

Open a task¶

A task is opened by recording its task.start; until then the engine refuses every other write into it (section 6.2).

task_id = str(uuid.uuid4())
context = context.model_copy(update={"task_id": task_id})
runner.record(Behavior(agent_id=context.agent_id, task_id=task_id,
                       kind=ActionKind.task_start, step_name="task.start"))

POST http://127.0.0.1:8080/record
Content-Type: application/json

{
  "step": {
    "agent_id": "agent_3f9c2a7b1d4e6f80",
    "task_id": "task-0001",
    "kind": "task.start",
    "step_name": "task.start"
  }
}

Evaluate, act, record¶

For each step, evaluate the intended behaviour, act only if it was not blocked, and record the completed step, with its output, under the action_id the evaluation returned.

intended = Behavior(agent_id=context.agent_id, task_id=task_id, kind=ActionKind.step_resource,
                    verb=Verb.GET, step_name="read_faq", input={"query": "refund policy"})
result = runner.evaluate(intended, context)  # allow or warn; a block raises
output = {"answer": "Refunds within 30 days."}  # the agent acts
runner.record(intended.model_copy(update={"action_id": result.action_id, "output": output}))

Over HTTP, /evaluate answers with the verdict and the action_id, and /record takes the completed step under that action_id:

POST http://127.0.0.1:8080/evaluate
Content-Type: application/json

{
  "intended": {
    "agent_id": "agent_3f9c2a7b1d4e6f80",
    "task_id": "task-0001",
    "action_id": "act-1",
    "kind": "step.resource",
    "verb": "GET",
    "step_name": "read_faq",
    "input": {"query": "refund policy"}
  },
  "context": {
    "agent_id": "agent_3f9c2a7b1d4e6f80",
    "task_id": "task-0001",
    "risk_classification": "minimal",
    "agent_allowed_tools": ["read_faq", "run_script"]
  }
}

POST http://127.0.0.1:8080/record
Content-Type: application/json

{
  "step": {
    "agent_id": "agent_3f9c2a7b1d4e6f80",
    "task_id": "task-0001",
    "action_id": "act-1",
    "kind": "step.resource",
    "verb": "GET",
    "step_name": "read_faq",
    "input": {"query": "refund policy"},
    "output": {"answer": "Refunds within 30 days."}
  }
}

A blocked step¶

Without a preceding gate, the policies block the code execution. The engine records the blocked step itself (section 5), so the agent does not act and does not record it.

intended = Behavior(agent_id=context.agent_id, task_id=task_id, kind=ActionKind.step_exec,
                    step_name="run_script", input={"code": "cleanup()"})
try:
    runner.evaluate(intended, context)
except KyvvuBlockedError as blocked:
    # The engine has recorded the blocked step itself: do not record it.
    verdict = blocked.result

# The answer has "blocked": true and the step's "recorded_seq": do not /record it.
POST http://127.0.0.1:8080/evaluate
Content-Type: application/json

{
  "intended": {
    "agent_id": "agent_3f9c2a7b1d4e6f80",
    "task_id": "task-0001",
    "action_id": "act-2",
    "kind": "step.exec",
    "step_name": "run_script",
    "input": {"code": "cleanup()"}
  },
  "context": {
    "agent_id": "agent_3f9c2a7b1d4e6f80",
    "task_id": "task-0001",
    "risk_classification": "minimal",
    "agent_allowed_tools": ["read_faq", "run_script"]
  }
}

End the task¶

Record the task.end, then end the task, which sends its records to the configured sinks.

runner.record(Behavior(agent_id=context.agent_id, task_id=task_id,
                       kind=ActionKind.task_end, step_name="task.end"))
runner.end_task(task_id)

POST http://127.0.0.1:8080/record
Content-Type: application/json

{
  "step": {
    "agent_id": "agent_3f9c2a7b1d4e6f80",
    "task_id": "task-0001",
    "kind": "task.end",
    "step_name": "task.end"
  }
}

###

POST http://127.0.0.1:8080/end_task
Content-Type: application/json

{"agent_id": "agent_3f9c2a7b1d4e6f80", "task_id": "task-0001"}

The same, offline¶

From a policy file the runner needs no account and no network, and the agent_id is your own. The task runs exactly as above, with the same verdicts. kyvvu serve --policy-file owasp.yaml serves the same routes offline. run_task is the task from "Open a task" to "End the task" above as one function (walkthrough.py in the documentation samples).

runner = KyvvuRunner(policy_file="owasp.yaml", log_location="none")
context = EvalContext(agent_id="support-agent", risk_classification="minimal", agent_allowed_tools=TOOLS)
offline_run = run_task(runner, context)

4. Where policies come from¶

Sources and precedence¶

A runner loads its policy set from one source:

the platform: api_url, api_key and agent_key (KV_API_URL, KV_API_KEY, KV_AGENT_KEY);
a policy file: a manifest YAML file, policy_file (KV_POLICY_FILE);
inline: a list of policies, policies=[...];
the disk cache: the last set an earlier platform fetch wrote, policy_cache_path (KV_POLICY_CACHE_PATH).

The first source that is fully configured takes priority, in that order: the platform, then a policy file or an inline list, then the disk cache. One exception keeps an offline run offline: a policy file or inline list passed as a keyword argument wins over a platform configured only through the environment. When both are passed as arguments, or both come from the environment, the platform wins, so a stray manifest cannot replace an organisation's policy set. A runner with no source at all raises KyvvuConfigError when it is built.

import os

from kyvvu_baseline import load_baseline_policies
from kyvvu_engine import KyvvuRunner

# The platform, keeping the last good set in a disk cache.
platform = KyvvuRunner(api_url="https://platform.kyvvu.com", api_key=os.environ["KV_API_KEY"],
                       agent_key="support-agent", policy_cache_path="policy-cache.json")
# A policy file, or an inline list: no account and no network.
from_file = KyvvuRunner(policy_file="owasp.yaml", log_location="none")
inline = KyvvuRunner(policies=load_baseline_policies(), log_location="none")
# Only the disk cache an earlier platform fetch wrote.
from_cache = KyvvuRunner(policy_cache_path="policy-cache.json", log_location="none")

Fetching from the platform¶

The runner sends GET /api/v1/policies with its agent_key, over HTTPS, with the agent API key as Authorization: Bearer KvKey-....
The platform answers with the agent's complete policy set and names the agent it was served for in X-Kyvvu-Agent-Id. The runner, and kyvvu serve, govern that agent only.
With a disk cache configured, the response's ETag is cached with the set, and the next fetch sends it as If-None-Match. A 304 keeps the cached set.
A platform whose floor the client is below (MIN_ENGINE_VERSION, MIN_POLICY_SCHEMA) answers 426 client_upgrade_required, raised as KyvvuClientUpgradeRequired; the runner keeps the set it already has.
Every policy is validated, from any source, against the schema of its rule's parameters. Loading is all-or-nothing: an unknown rule_type, a failed schema validation, a duplicate policy name or a regex that does not compile raises PolicyLoadError naming the policy, and nothing from that set is loaded.
After each successful fetch the set is written to the disk cache, if one is configured. When a cold start's fetch fails, the runner loads the disk cache instead.

Refresh, stale sets and the fail mode¶

The runner refreshes a platform set every policy_ttl_seconds (KV_POLICY_TTL_SECONDS, default 300). When a refresh fails or is refused, it keeps enforcing the last good set, and reports it as stale. A disk cache that is the configured source is also stale once it is older than policy_cache_max_age_seconds (KV_POLICY_CACHE_MAX_AGE_SECONDS, default 86400).

With no usable set at all, the fail mode decides: fail_mode (KV_POLICY_FAIL_MODE) is closed by default, which blocks every step.* behaviour, and open allows them. Task boundaries are allowed either way, and a source that answers with no policies is not a failure.

Load states¶

policy_status().load_state says why the set in effect looks the way it does:

enforcing: at least one policy is loaded and enforced;
none_configured: the source answered and has no policies, so nothing is enforced;
unavailable: nothing is loaded, because no load has been attempted yet or the fetch failed with no cached set to fall back on;
stale_cache: policies are enforced, but the set is not known to match its source, because the last fetch failed or the configured disk cache is too old.

Whenever nothing is enforced, the runner says so once on stderr, naming the state.

for name, runner in [("platform", platform), ("policy file", from_file),
                     ("inline", inline), ("disk cache", from_cache)]:
    status = runner.ensure_policies()
    print(f"{name}: {status.load_state.value}, {status.policy_count} policies")
    runner.stop()

With the platform up, each runner from the sources above prints:

platform: enforcing, 15 policies
policy file: enforcing, 8 policies
inline: enforcing, 15 policies
disk cache: enforcing, 15 policies

With the platform down, the platform runner keeps enforcing its disk cache, and both report it as stale:

platform: stale_cache, 15 policies
policy file: enforcing, 8 policies
inline: enforcing, 15 policies
disk cache: stale_cache, 15 policies

5. What evaluation returns¶

PolicyEngine evaluates an intended Behavior with an EvalContext against the loaded policy set and the ordered history for that task. PolicyEngine.evaluate() and explain() read history without recording anything; callers record completed steps separately. Each applicable policy produces a PolicyResult, the per-policy outcome: whether it was violated, its severity and the details of the violation. The configured aggregator combines those results into a risk score and an allow, warn or block verdict. The default aggregator takes the highest severity among violations. An invalid policy set is refused as a whole at load time, and a rule exception counts as a violation of that policy.

The result is an EvalResult:

action: the verdict, allow, warn or block;
risk_score: the normalised violation severity, a finite number from 0.0 (no violations) to 1.0;
policies: one PolicyResult per evaluated policy, in evaluation order;
action_id: the identifier a KyvvuRunner evaluation used or generated for a step.* behaviour;
recorded_seq: the seq of the step when the engine recorded it itself, else None.

When KyvvuRunner.evaluate() blocks a step.* behaviour, the runner records the blocked step itself before raising KyvvuBlockedError: under the evaluation's action_id, with meta["kyvvu.eval"], inside the same task lock, in the one shape kyvvu_engine.blocked_step() builds (output.status: "blocked", output.error, and output.decision: "deny" on a step.gate). EvalResult.recorded_seq (on exc.result) is the seq it was given. PolicyEngine.evaluate_and_record() records a block the same way, and kyvvu serve's /evaluate returns the recorded_seq. A caller must not record a blocked step: a record() of that action_id is refused as a duplicate. A blocked step is ordinary rule history (rules opt in to counting attempts with count_attempts). If recording fails, the block still stands and recorded_seq is None.

A task boundary (task.*) is not given an action_id by the runner, and a blocked one is not recorded by the engine. How a blocked boundary is represented may change in a 1.x release.

6. Guarantees¶

6.1 Determinism¶

PolicyEngine's verdict and per-policy outcomes are a function of the Behavior being evaluated, including its timestamp, the task history, the ordered policy set and the aggregator (with the agent identity in EvalContext). The same inputs give the same verdict, on any host and at any time.

Time-based rules such as working_hours_only evaluate against the behaviour's timestamp, which the client records when it forms the intent. PolicyEngine reads no wall clock, and EvalContext has no evaluation-time field. So the time a policy was checked against is the time the trace already carries, and a preflight is judged at the time the intent was formed, not when the action later executes, even when it waits behind a slow human gate. A registration has no behaviour and therefore no evaluation time, and no registration rule reads one.

To replay a decision, evaluate the recorded behaviours, with their recorded timestamps, against a PolicyEngine loaded with the same policy set: no clock needs injecting. Construct a Behavior once or supply its timestamp explicitly, since its default timestamp reads the clock at construction. History and policy changes between calls change the inputs. Custom rule functions or aggregators that read external state can also break this guarantee.

The guarantee covers engine evaluation, not the KyvvuRunner wrapper. The runner fetches policies, logs results, and may generate a random action_id for an intended action. It is also a live entrance, so it bounds clock skew Kerberos-style: KyvvuRunner.evaluate() (and /evaluate on kyvvu serve, with 422) refuses a behaviour whose timestamp is more than max_clock_skew_seconds (KV_MAX_CLOCK_SKEW_SECONDS, default 300) from the host's UTC clock, in either direction. The skew bound applies only there: record(), restore_task(), restore_intent() and PolicyEngine replay history and accept any recorded time. A replay should compare engine verdicts and policy outcomes with fixed inputs, not runner-generated identifiers or I/O effects.

6.2 Task lifecycle¶

A task is opened by recording its task.start. KyvvuRunner refuses (TaskStateError) every other write (evaluate, record, restore_intent) into a task whose task.start it has not recorded, and a second task.start. A fresh runner rebuilding a task from a history tail that lacks its task.start calls KyvvuRunner.restore_task() first. Once a task.end or task.error is recorded the task is closing: any later write raises TaskClosedError, and only end_task() may follow. PolicyEngine applies the same closing rule and refuses a second task.start, refuses writes and intent registration after end_task(), but does not require a task.start first.

Within one runner (and one PolicyEngine), a task_id that was ended or evicted is not reopened, for its most recent 100,000 closed tasks; no claim is made across processes or beyond that bound, and a task_id must still never be reused. The stale-task sweeper's eviction is local to its runner: it ends the evicted buffer with a task.idle whose properties.task.reason is "expired", which is not a terminal boundary on the wire.

The order of seq is per agent, not per task. Several agents logging under one task_id each get their own 1, 2, 3 …, and one task order exists only when one client logs the whole task. This is a disclosed difference from AAG §3.4.

6.3 What happens to the events you send¶

Before anything is recorded, the engine checks every event. An event is refused, and nothing is stored, if: - a payload field (properties, input, output, meta) is not I-JSON (RFC 7493); - its timestamp has no time zone; - an identifier breaks the identifier grammar; - its action_id was already used in the same task, except to pair an intended step with its own completion; - it sets a meta key starting with kyvvu.; - it writes into a task that is not open.

The engine records: - every event you record; - every step.* it blocks. It records a blocked step itself, under the evaluation's action_id (callers must not record it again); - intended steps that are abandoned when their task ends.

Within one agent and one task, seq strictly increases.

Delivery to sinks is at most once and is observable. The engine sends recorded events to the sinks you configure (KV_LOG_LOCATION, KV_INCIDENT_LOCATION) through one bounded queue (KV_REPORT_QUEUE_SIZE, default 1000). KyvvuRunner.report_status() counts every report as queued, delivered, failed, or dropped because the queue was full. When the process exits, the engine waits up to KV_REPORT_EXIT_FLUSH_SECONDS (default 2) for queued reports. Events can therefore fail to reach a sink in three ways: - the queue is full; - the sink fails; - the process ends first.

In each case, either a counter records it or the process has ended. A failed platform batch is not sent again: the report is counted as failed and the engine discards its events.

What the platform stores, it has acknowledged in full. For the platform sink, the engine numbers each task's exported events seq 1, 2, 3 … with no gaps, and hashes each with the v1 chain: SHA-256 over agent_id, task_id, seq, input and the previous hash. The platform: - recomputes every hash; - refuses a batch that does not extend the chain it has stored for that task; - stores the whole batch in one transaction before it answers.

A batch the platform refuses or does not acknowledge is counted as failed. Nothing from it is stored.

What the v1 chain proves, and what it does not. It shows that the covered fields of stored events were not edited without the chain being recomputed. It does not cover kind, verb, timestamp, output, properties or meta. It is unsigned, so anyone who can write to the store can recompute it. It cannot show that events were deleted from the end of a task.

What this means for you. The engine and the platform together guarantee that a stored record is an event the agent sent, in the order it sent it, with its covered fields unchanged since then. They do not guarantee that every event the agent produced was stored. If you need completeness, monitor report_status(), or use a sink that you make durable yourself.

7. Threat model and limits¶

The engine checks an agent's proposed action against policies chosen by its caller, including rules that inspect the agent's earlier steps. It treats the action data as untrusted and reports policy violations before the caller executes the action. The caller must actually submit each action for evaluation and honor a block; actions the integration does not submit, and side effects it does not report, are not governed. It also cannot prove that reported input, output, or history reflects what happened outside the process. Policy authors, custom rule code, the configured aggregator, and the policy delivery path are trusted inputs; this library is not a sandbox for malicious policy code.

Policy sets are unsigned in 1.0. The engine enforces the policy set it is delivered without verifying who wrote it. A TLS-inspecting proxy, a compromised edge, or anyone who can write the on-disk policy cache can change the enforced set undetected.

8. Stability and versioning¶

What the 1.0 interface promises, and what a 1.x release may add, is the interface stability and versioning policy.