Governance Agent

A 4-phase governance deep dive that demonstrates every governance-related SDK feature: Phase 1 records 5 pre-execution policy checks via waxell_ctx.record_policy_check(), Phase 2 uses manual waxell_ctx.check_policy() for mid-execution policy evaluation with @waxell.tool and @waxell.decision, Phase 3 demonstrates the sync wrapper retry feedback loop with start_run_sync/complete_run_sync and RunCompleteResult.should_retry, and Phase 4 records governance events via client.record_events_sync().

Environment variables

This example runs in dry-run mode by default (no API key needed). For live mode, set OPENAI_API_KEY, WAXELL_API_KEY, and WAXELL_API_URL.

Architecture

Key Code

Pre-execution policy checks via waxell_ctx

Phase 1 records 5 governance policy evaluations directly on the trace using the context-level API.

@waxell.observe(agent_name="governance-evaluator", workflow_name="policy-evaluation")
async def run_pre_execution(query: str, client, *, waxell_ctx=None) -> dict:
    policies = [
        ("Budget Policy", "allow", "budget", "Within daily token budget"),
        ("Rate Limit Policy", "allow", "rate-limit", "3 of 100 hourly requests used"),
        ("Safety Policy", "warn", "safety", "High-risk domain: financial trading"),
        ("Kill Switch", "allow", "kill", "No active kill switches"),
    ]

    for policy_name, action, category, reason in policies:
        waxell_ctx.record_policy_check(
            policy_name=policy_name, action=action, category=category,
            reason=reason, phase="pre_execution", priority=10,
        )

Mid-execution manual policy check

Phase 2 shows on-demand policy evaluation between execution steps.

@waxell.observe(agent_name="governance-mid-execution", workflow_name="mid-execution-check")
async def run_mid_execution(query: str, *, waxell_ctx=None) -> dict:
    result = risk_analysis("financial_trading", "high")  # @tool

    policy_result = await waxell_ctx.check_policy()  # Manual mid-execution check
    # policy_result.action, policy_result.allowed, policy_result.should_retry

    decision = make_execution_decision(result["risk_score"])  # @decision
    compliance = compliance_check("governance-demo", "trading")  # @tool

What this demonstrates

• waxell_ctx.record_policy_check() -- 5 pre-execution governance policy evaluations with action/category/reason/phase/priority.
• waxell_ctx.check_policy() -- on-demand mid-execution policy check returning PolicyCheckResult.
• @waxell.tool -- risk_analysis and compliance_check tool calls.
• @waxell.decision -- execution route decision (continue/escalate_to_human/abort).
• @waxell.step_dec -- pre-execution and mid-execution step recordings.
• client.start_run_sync()/complete_run_sync() -- sync wrapper API for retry feedback loop.
• RunCompleteResult.should_retry/retry_feedback -- governance retry guidance from the server.
• client.record_events_sync() -- governance event recording for the event timeline.
• Auto-instrumented LLM calls -- 3 OpenAI calls captured across phases.

Run it

# Dry-run (no API key needed)
python -m app.demos.governance_agent --dry-run

# Live mode with OpenAI
OPENAI_API_KEY=sk-... python -m app.demos.governance_agent

Source

dev/waxell-dev/app/demos/governance_agent.py