zemtik-govern 0.5.0.0

Security-first modular Python wrapper around Microsoft AGT

zemtik-govern — Governance for LangChain, LangGraph, and Microsoft AGT

Govern every tool call through a fixed pipeline — identity → policy → audit — with a prompt-injection screen folded into the policy seam. Fail-closed by design: any fault during governance blocks the tool and is audited, never silently allowed.

zemtik-govern is a security-first Python wrapper around Microsoft AGT (agent-os-kernel + agentmesh-platform). It is the moat between an autonomous agent and the privileged tools it can call.

What it guarantees

• Deny-by-default. AGT's native evaluator allows when no rule matches; zemtik-govern overrides that — no matching rule is a deny. A tool is reachable only if a rule names it.
• Fail-closed. Any exception in identity or policy is caught, audited as a system denial, and re-raised. The wrapped tool never runs on a governance fault.
• Prompt-injection guard. A poisoned tool argument (ignore all previous instructions…, role-play hijacks, delimiter smuggling) is denied before the tool runs, even when policy would allow the call. The deny names the offending field; the raw payload is never echoed into logs.
• Decision budget. A hung identity or policy seam is bounded by a per-call deadline. A breach is a fail-closed system denial (DecisionBudgetExceeded), not a leaked allow.
• Tamper-evident audit. Every outcome — allow, deny, or system error — is recorded on a Merkle-chained, HMAC-signed trail that verifies independently and survives a process crash. A durable file trail requires an HMAC secret in $ZEMTIK_AUDIT_SECRET; the in-memory trail does not.
• Idempotency. A duplicate idempotency_key replays the cached decision; a key reused with a different payload is an audited conflict, never a false replay. Backing caches are bounded (LRU + TTL) against unbounded growth.
• Safe rollout. shadow mode observes would-denies without enforcing; a per-guard shadow stance lets a new guard run observe-only for one release before it enforces. A kill-switch reverts to a prior governed fallback — never to allow-all.

The API is async throughout (every seam, govern, and proxy) and requires Python 3.11+.

Two ways in: use the LangChain Quick Start if your tools are LangChain/LangGraph; use the AGT-native API to govern any async callable directly.

LangChain Quick Start

pip install 'zemtik-govern[langchain]'   # quote the extras for zsh

govern.yaml — allow exactly one tool by name:

mode: strict
audit_sink: memory
rules:
  - name: allow-add-numbers
    condition:
      field: action
      operator: eq
      value: add_numbers
    action: allow

The snippet below wires a stub policy that allows every call — it demonstrates wiring only, not enforcement. For real deny-by-default policy and the injection guard, build the governor from config (GovernanceRegistry.from_config, see the AGT-native API below) and load govern.yaml.

from zemtik_govern.core import ZemtikGovern
from zemtik_govern.identity import AgentRef
from zemtik_govern.langchain import govern_tool
from zemtik_govern.protocols import Decision
from langchain_core.tools import tool

class _Policy:
    async def identify(self, s): return AgentRef(did=f"did:x:{s}")
    async def evaluate(self, c): return Decision(allowed=True, action=c.action, matched_rule="allow-add-numbers", reason="ok")
    async def write(self, e): return "e"
gov = ZemtikGovern(identity=_Policy(), policy=_Policy(), audit=_Policy())

@tool
def add_numbers(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

result = govern_tool(add_numbers, govern=gov).invoke({"a": 3, "b": 4})
print(result)  # 7

Debug with full governance logs:

ZEMTIK_DEV=1 python examples/langchain_minimal.py

For a drop-in LangGraph ToolNode replacement, see examples/langgraph_toolnode.py.

Full integration guide: docs/integrations/langchain.md

Governing a real LLM agent

The snippet above wires governance with a hand-built policy and no model in the loop. Wire the same gov/tools into a real agent — GovernedToolNode governs every tool call the model itself decides to make, identically whether the caller is a test or a live LLM:

from langchain_openai import ChatOpenAI
from zemtik_govern.langchain import GovernedToolNode

node = GovernedToolNode(tools, govern=gov)              # governs every call the model makes
model = ChatOpenAI(model="gpt-5.4-nano").bind_tools(tools)

ai_message = model.invoke(messages)                      # a real LLM call
if ai_message.tool_calls:
    result = node({"messages": [*messages, ai_message]})  # each tool call passes through govern() first

For the full, runnable version of this loop — real gpt-5.4-nano calls, a governed banking-agent tool set, a 15-prompt injection battery, and (when LANGFUSE_PUBLIC_KEY/LANGFUSE_SECRET_KEY are set) the LLM generations and governance spans landing under one shared Langfuse trace — see sandbox/e2e_openai_governed.py under Sandbox & Demos below.

AGT-native API

Every tool invocation passes identity → policy → audit in fixed order — any fault is a system denial; the guarded tool never runs on a governance error.

pip install zemtik-govern

Copy zemtik.example.yaml to zemtik.yaml and edit it, then wire and call. A request is three inputs: action (the tool being called, what policy matches on), subject (the caller, resolved to a DID by identity), and payload (the arguments, screened by the injection guard).

import asyncio
from zemtik_govern import AGTBoundary, GovernanceConfig, GovernanceContext, GovernanceRegistry

async def main():
    config = GovernanceConfig.load("zemtik.yaml")
    gov = GovernanceRegistry.from_config(config, AGTBoundary()).build()

    ctx = GovernanceContext(action="tool.run", subject="agent-1", payload={"q": "hello"})
    decision = await gov.govern(ctx)   # raises GovernanceDenied if blocked

asyncio.run(main())

Or wrap a callable so it is governed on every call (inside an async context):

governed_fn = gov.proxy(my_tool, action="tool.run", subject="agent-1")
result = await governed_fn(q="hello")

In strict/enforce mode the injection guard is always on. By default it uses AGT's own vetted detection rules (no config needed); set injection_rules_path only to pin a version or diverge — policies/prompt-injection.yaml is a ready-to-edit snapshot of those defaults. A file audit_sink requires an HMAC secret in $ZEMTIK_AUDIT_SECRET. See zemtik.example.yaml for the full annotated configuration. To watch the injection guard deny a poisoned argument, run scenario S11 in sandbox/qa_demo.py (S14 for the per-guard shadow stance, S16 for the output seam redacting PII; see Sandbox & Demos).

Security model

Operational modes

Mode	Denials enforced	Policy required	Audit required
strict	Yes	Yes	Yes
enforce	Yes	Yes	Yes
shadow	No (observe only)	No	Yes

A new guard can also run observe-only independent of the mode via a per-guard injection: {mode: shadow} or budget: {mode: shadow} stance — useful to watch would-denies for one release before enforcing.

Exceptions

Every failure is a typed, catchable exception carrying a stable .code and .guard, and an .audit_id that correlates back to its audit entry. The two enforcement exceptions raise only in enforcing modes — under shadow (or a per-guard shadow stance) the would-deny is observed and audited but never raised, so the tool runs:

Exception	Meaning
GovernanceDenied	Policy (or the injection guard) denied. Raised in strict/enforce; in shadow it is audited, not raised. Carries the Decision.
DecisionBudgetExceeded	A seam outran the decision budget; fail-closed. Raised when budget enforces; under budget: {mode: shadow} the breach is logged, not raised. Carries .limit_seconds / .elapsed_seconds.
GovernanceError	A system fault in a seam; the tool was blocked.
GovernanceNotConfigured	Insecure startup config; raised at boot, not request time.
OutputGovernanceDenied	An output rail blocked a tool's return value (read-classified tool, or an unscreenable/non-JSON return); fail-closed. Carries .rail and .audit_id. Raised in strict/enforce; a shadow rail observes a would-deny instead.

Output seam (opt-in, proxy() only)

Beyond the input pipeline, an optional post-invocation output seam screens a tool's return value through PII rails — enabled with output_screening: true and only inside proxy() (direct govern()/govern_sync() callers are input-only). The deny shape splits by the tool's tool_io_map classification, because by the time output is screened a write has already executed:

Audit event	Effect	Caller sees
output_allowed	clean	the real value
output_denied_raised	read-classified hit	OutputGovernanceDenied raised
output_denied_redacted	write-classified hit (side effect already ran)	a RedactedOutput sentinel (HIGH-severity audit row)

RedactedOutput is a poison sentinel: str()/repr() are safe (<output redacted: audit_id=…>) but attribute/item/iter access raises RedactedOutputAccessError. Wrap every governed result in gov.unwrap(result) for one uniform contract — it returns a clean value and raises OutputGovernanceDenied on a redacted one. The rail names itself and its tunable knob in the deny, never the offending value (no-echo). Watch it live with scenario S16 in sandbox/qa_demo.py. See Configuration Reference and the Integration Guide for the rails / tool_io_map config and the observe→enforce rollout.

Observability & evals (optional, off by default)

pip install 'zemtik-govern[langfuse]'   # zero-cost when not enabled — this extra
                                         # is only imported if observability.enabled

zemtik-govern can emit governance-pipeline telemetry to Langfuse — self-hosted or cloud — and run a governance-decision evals harness. Both are off by default, isolated behind one boundary (only one module ever imports langfuse), and fail open: a down, misconfigured, or unreachable Langfuse backend degrades to a no-op and never blocks a decision — the opposite failure mode from governance itself, which stays fail-closed throughout.

observability:
  enabled: false        # master switch — off by default
  provider: langfuse     # only provider this wrapper wires today
  sample_rate: 1.0       # 0.0–1.0, decided once at the trace root
  mask_payloads: true    # spans carry masked facts only; false refused in strict/enforce
  emit_subject: hmac     # hmac | omit | raw — "raw" refused in strict/enforce
  emit_rule_names: true  # false emits an opaque rule id instead of the semantic name

Enabling it turns each govern() call into a masked, nested trace (govern → identity/policy/audit, plus an injection/output annotation) — no raw payload, matched pattern, or output value ever reaches a span; the caller's subject defaults to a keyed HMAC digest, never the raw value. LANGFUSE_PUBLIC_KEY/LANGFUSE_SECRET_KEY/LANGFUSE_HOST are read from the environment only, never YAML. A LangChain agent additionally gets its LLM generation and governed tool calls under one shared trace via zemtik_govern.langchain.langfuse_callback().

The evals package (zemtik_govern.evals) scores governance-decision quality — does the policy/injection engine actually produce the right allow/deny verdict — against a ground-truth dataset seeded from this repo's own live-proven cases (the injection battery, the policy matrix, PII-adjacent actions), with a deterministic exact-match evaluator plus an optional LLM-as-judge and calibration report:

python sandbox/e2e_openai_governed.py --eval    # needs neither Langfuse nor an OpenAI key

Full reference: docs/observability.md. Design rationale: ADR 002 (the Langfuse version pin) and ADR 003 (the fail-open contract and data-egress posture).

Documentation

• LangChain Integration Guide — govern_tool, GovernedToolNode, LangSmith, error reference
• Architecture — three-seam design, idempotency, fail-closed guarantee, deny-by-default moat
• Integration Guide — quickstart, usage patterns, custom seams, error reference
• Configuration Reference — YAML fields, mode matrix, env vars, startup validation
• Operations Guide — deployment checklist, durable audit, kill-switch, shadow rollout, monitoring
• API Reference — full public API (classes, protocols, exceptions)
• Sandbox & Demos — runnable demos: three-seam scenarios, the injection battery, audit forensics, staged shadow → enforce cutover, a real gpt-5.4-nano agent governed end to end
• Observability — optional Langfuse tracing/evals extension (OFF by default; fail-open telemetry, self-hosted first)

Sandbox & Demos

Runnable demos live in sandbox/ and exercise the real pipeline against the live AGT stack:

source .venv/bin/activate

# Three-seam scenarios S1–S16: allow, deny, fail-closed, idempotency, the
# injection guard, the decision budget, error codes/audit_id, per-guard shadow,
# and the output seam redacting PII (S16)
ZEMTIK_AUDIT_SECRET=qa-test-secret python sandbox/qa_demo.py

# Audit trail: verify Merkle/HMAC, extract proofs, detect tampering
ZEMTIK_AUDIT_SECRET=audit-secret python sandbox/auditor.py

# Staged dogfood cutover: shadow -> enforce, kill-switch revert, audit integrity (no API key)
ZEMTIK_AUDIT_SECRET=dogfood-secret python sandbox/dogfood_cutover.py

# A real gpt-5.4-nano agent governed end to end, plus a 15-prompt injection
# battery and deterministic module probes — including the output seam redacting
# PII from a tool's return (needs [langchain,openai] + an OpenAI key)
uv pip install -e ".[dev,langchain,openai]"
cp .env.example .env   # then set OPENAI_API_KEY in .env (gitignored)
python sandbox/e2e_openai_governed.py

# --eval: the evals package (governance-decision ground truth, exact-match +
# optional LLM-as-judge + calibration) against the SAME real governor — needs
# neither Langfuse nor an OpenAI key to run
python sandbox/e2e_openai_governed.py --eval

See docs/sandbox.md for what each demo proves and its outputs.

Development

source .venv/bin/activate
pytest
ruff check src/

See CONTRIBUTING.md for the full dev setup and three-seam contract.

Status: v0.4.0.0 — see CHANGELOG.md. License: Apache-2.0.