Core Concepts

Enforcement gates vs observability traces

Most LLM tools (LangSmith, Langfuse, Helicone) are observability platforms — they show you what happened after the fact. The Governance SDK is different: it provides enforcement gates that run before the LLM call fires.

The HMAC chain

Every audit row stores an HMAC-SHA256 computed over its own immutable fields plus the previous row's HMAC. This creates a chain where modifying any byte of any past row breaks that row's HMAC and every row that follows it.

The HMAC secret lives in your secrets manager (GOVERNANCE_AUDIT_SECRET). Without it, an attacker with database access cannot forge valid rows. With it, the chain is verifiable from the database alone — no trust in the application required.

The HMAC covers every immutable field: event_id, session_id, agent_id, parent_event_id, kind, model, input_hash, output_hash, metadata, prev_hash, and created_at. Modifying any of these — not just the chain link — breaks the verification.

Opt-in modules and startup signals

Every module is opt-in via a GovernanceConfig flag. When disabled, the module is not constructed at all — accessing it raises AttributeError, not a silent no-op. This is an architectural invariant: a disabled module is loudly absent, never quietly present-but-inert.

# Lightweight deployment — drop loop detection and presence tracking
async with GovernanceSDK(
    database_url="postgresql://...",
    enable_loop=False,
    enable_presence=False,
) as sdk:
    ...  # sdk.loop and sdk.presence raise AttributeError on access

# Audit-only deployment — no wrappers, no warning
async with GovernanceSDK(
    database_url="postgresql://...",
    warn_on_no_wrappers=False,  # suppress the wrap_openai/wrap_anthropic startup warning
) as sdk:
    ...

# Full enforcement with declared-ceiling budget projection
async with GovernanceSDK(
    database_url="postgresql://...",
    default_max_tokens=4096,  # ceiling used by the projected budget gate
    verify_chain_on_read=True,  # raise on first broken HMAC link at read time
) as sdk:
    ...

warn_on_no_wrappers (default True) emits a structlog warning at sdk.start() when no wrap_openai or wrap_anthropic has been registered. This is the only startup signal that enforcement is not covering your LLM calls — silencing it in a production deployment that expects wrappers will hide a misconfiguration. Set to False only in intentionally wrapper-free deployments.

What the SDK does NOT protect against

Enforcement covers every action routed through the SDK. Deployers and security reviewers should read these carve-outs before treating the SDK as a complete security boundary:

• Direct client bypass. Any code path that calls anthropic.Anthropic() or openai.OpenAI() directly — without wrap_anthropic() or wrap_openai() — is invisible to all SDK gates. Budget, scope, and audit are all bypassed. An LLM-generated tool function that instantiates its own client is not governed.
• Process-level bypass. The SDK provides in-process enforcement gates. It does not provide kernel-level, network-level, or process-isolation-level enforcement. A second Python process that bypasses the wrappers entirely is not governed.
• Streaming cost precision. Streaming calls are budget-gated using the declared max_tokens value before the stream opens. Actual token usage is recorded from the stream's final usage object. If the API does not return a usage object, the SDK falls back to max_tokens as the tracked value — actual usage may differ.
• On-demand tampering detection only. The HMAC chain detects tampering when verification is explicitly run or with verify_chain_on_read=True. It does not alert as tampering occurs, and cannot prevent deletion of the entire chain by a privileged DBA who also controls the HMAC secret.
• HITL non-blocking mode. When a HITL gate is configured with blocking=False, the gate raises ApprovalPending and the caller is responsible for not proceeding. The SDK cannot prevent a caller who ignores ApprovalPending from proceeding anyway.
• Tool invocations inside LLM responses. Scope enforcement gates the LLM API call itself (using a sentinel action name). It does not inspect tool calls returned inside the LLM's response. Tool execution must be gated at the tool-execution layer separately.

The eight modules

Non-breaking guarantee

The SDK distinguishes between observation surfaces (which never raise) and enforcement surfaces (which raise by contract):

# OBSERVATION — never raises, even if Postgres is down:
await sdk.audit.log(event)            # returns record or placeholder
await sdk.cost.track(agent, sid)      # logs and continues
await sdk.gates.request(...)          # returns request object
await sdk.presence.heartbeat(agent)   # updates last_seen
await sdk.loop.check(agent, sid)      # returns bool

# ENFORCEMENT — raises by contract (that's the gate):
await sdk.scope.check(agent, tool="x")              # raises ScopeViolation
await sdk.cost.check_or_raise(agent, sid)            # raises BudgetExceeded
await sdk.gates.wait_for(request_id, timeout=60)     # raises ApprovalTimeout / ApprovalDenied
await sdk.loop.record_call(agent, sid, "tool_name")  # raises LoopDetected (action="raise")

Fail-closed cost gate

If cost.check_or_raise() cannot read the counter (database unreachable, query timeout), it denies the call rather than allowing it. This prevents an attacker from draining budgets by taking down the cost store. The failure is audit-logged (if the audit substrate is reachable) so operators can see what happened. Pass cost_fail_open=True at SDK init if you prefer availability over safety.

JSONL durable fallback

When the primary Postgres store is unreachable, audit events are written to a local JSONL file (~/.codeatelier_governance/audit_fallback.jsonl by default). This ensures that events survive process crashes and network partitions — no audit event is silently dropped.

Weak secret rejection

The SDK rejects audit secrets that are too short (<32 bytes) or too weak (fewer than 8 unique bytes). This catches the common "I forgot to set the env var" footgun — placeholder values like "x" * 64 are rejected at initialization with a clear error message explaining how to generate a real key.

export GOVERNANCE_AUDIT_SECRET=$(python -c 'import secrets; print(secrets.token_hex(32))')

Hidden tool policies

Scope enforcement has two layers: allowed tools block execution at call time, while hidden toolsgo further by removing tools from the LLM's context entirely. A blocked tool produces a ScopeViolation when called; a hidden tool never appears in the tool list the LLM sees, so the agent cannot even attempt to call it.

sdk.scope.register(ScopePolicy(
    agent_id="support-agent",
    allowed_tools=frozenset({"search", "reply"}),
    hidden_tools=frozenset({"delete_account", "escalate_admin"}),
))

# Before passing tools to the LLM:
visible_tools = sdk.scope.filter_tools("support-agent", all_tools)
# "delete_account" and "escalate_admin" are not in visible_tools

Built-in model pricing

The SDK ships a pricing table for 24 models from OpenAI, Anthropic, Google, Meta, and Mistral. Instead of computing and passing usd= manually, call track_usage() with the model name and token counts. The SDK auto-computes the cost.

Loop and anomaly detection

A common failure mode in agentic systems is a runaway loop — the agent calls the same tool over and over, burning tokens without making progress. The loop detection module catches this pattern using a sliding window per (session, tool) pair.

Detection is scoped to (session_id, tool_name). Tool names are normalized to lowercase so Search_Web and search_web are treated as the same tool. The sliding window only counts calls within the last window_seconds.

Agent presence

In a multi-agent deployment, knowing which agents are alive is critical. The presence module provides a heartbeat-based lifecycle with three states, stored in your existing Postgres. No background worker, no Redis pub/sub — just a table and a few methods.

# Agent startup:
await sdk.presence.heartbeat("billing-agent")  # status: Live

# During processing (call periodically):
await sdk.presence.heartbeat("billing-agent")  # still Live

# Between tasks:
await sdk.presence.mark_idle("billing-agent")  # status: Idle

# Health check (run from a cron or /health endpoint):
stale = await sdk.presence.check_stale(timeout_seconds=300)
# Agents with no heartbeat in 5 min → Unresponsive

# Graceful shutdown:
await sdk.presence.close_agent("billing-agent")  # removed

Policy hot-reload

In production, you should not need to restart your application to update a scope or budget policy. With hot-reload enabled, the SDK polls the governance_policies table at a configurable interval and atomically replaces in-memory policies when changes are detected.

sdk = GovernanceSDK(
    database_url=os.environ["DATABASE_URL"],
    hot_reload=True,           # opt-in
    hot_reload_interval=30,    # poll every 30 seconds (default)
)

# Update a policy in Postgres (e.g. via the console or a migration).
# Within 30 seconds, the running SDK picks up the change — no restart.

Session time limits

Beyond token and USD caps, you can set a maximum session duration with per_session_seconds. The SDK tracks when each session started and enforces the limit in check_or_raise(). Valid range: 0 to 86,400 seconds (24 hours).

sdk.cost.register(BudgetPolicy(
    agent_id="interview-agent",
    per_session_seconds=300,   # 5 minutes max
    per_session_usd=2.00,      # dollar cap still applies too
))

# 6 minutes later...
await sdk.cost.check_or_raise("interview-agent", session_id)
# raises BudgetExceeded: session time limit exceeded: 360s > 300s limit

Self-approval prevention

When an agent sends a heartbeat with operator_id, the console uses that field to enforce self-approval preventionon HITL gates. The operator who owns the agent cannot approve their own agent's requests. This is fail-closed:

# At agent startup, bind the agent to its operator:
await sdk.presence.heartbeat(
    "billing-agent",
    operator_id="alice@company.com",
    metadata={"version": "1.2.0"},
)

Chain fork detection

Beyond verifying each row's HMAC, the chain verification also detects forks — when two events share the same prev_hash. This catches an attacker who inserts a parallel branch into the chain (e.g. splicing a second root or injecting events at a branch point).

Sync vs Async

The SDK is async-first (GovernanceSDK), but many Python web frameworks (Flask, Django, FastAPI sync routes) run synchronous code. The GovernanceSDKSync wrapper solves this by spinning up a daemon thread with a dedicated asyncio event loop. Every async method becomes a blocking sync call via asyncio.run_coroutine_threadsafe.

import os
from codeatelier_governance import GovernanceSDKSync, AuditEvent

with GovernanceSDKSync(database_url=os.environ["DATABASE_URL"]) as sdk:
    sdk.audit.log(AuditEvent(agent_id="a", kind="tool.call"))
    sdk.scope.check("a", tool="send_email")       # blocking
    sdk.cost.check_or_raise("a", session_id)       # blocking

SSE streaming for the console

The governance console uses Server-Sent Events (SSE) for real-time event streaming. Instead of polling the API, the console frontend opens a persistent connection to GET /api/stream/events and receives audit events, presence changes, and approval updates as they happen. SSE is authenticated — the stream requires either a session cookie or an x-governance-token header.