The Agent Auditor: Observability and Compliance for Gemini 3 Workflows

A Gemini 3 agent just moved a customer from “confirmed” to “refunded” without a manager. The right business flow was executed — but was it authorized to be?

The shift from command-line tools to autonomous agents changes what “verification” means. You can’t audit a decision by reading code, because the decision came from a model reasoning over retrieved context. The Agent Auditor pattern is the missing layer that makes autonomous workflows observable, defensible, and compliant.

In this tutorial, you will learn how to build an auditing layer for multi-agent Gemini 3 systems: capturing step traces, replaying decisions, validating policy conformance, and producing regulator-ready evidence.

Why Static Logging Is Not Enough

Classic logs record what happened. An auditor needs why it was allowed to happen. Consider the thin end of the wedge:

• An agent automatically discounts an order 30% because it misread a loyalty tier.
• A support bot reroutes sensitive data to an external tool without a policy check.
• A finance agent executes a transfer based on stale account balances.

None of these throw exceptions. Every step looked valid at the moment it ran. Only the combination of decisions — a reasoning chain — reveals the problem.

The Agent Auditor treats each workflow as a governed state machine: every state transition must be both (a) recorded for replay and (b) validated against policy before it is allowed.

The Auditor Triangle: Trace, Validate, Attest

A practical agent auditing layer has three responsibilities:

Layer	Responsibility	Tooling
Trace	Record every step: prompts, tool calls, tool outputs, reasoning	OpenTelemetry + agent span model
Validate	Check each transition against a policy/rule engine	OPA / Rego, custom decision gates
Attest	Produce tamper-evident, exportable evidence of conformance	Hashed audit chain + signed report

flowchart LR
    A["Gemini 3 Agent"] --> B["Audit Proxy"]
    B --> C["Trace Store"]
    B --> D["Policy Engine"]
    D -->|allow| E["Execute Tool"]
    D -->|deny| F["Human Escalation"]
    C --> G["Attestation Report"]

The proxy sits between the agent and the tools it can call. It records every interaction and asks the policy engine for permission before letting a side-effecting tool run.

Building the Audit Proxy

Wrapping tool execution is the cleanest interception point. Every Gemini 3 function call passes through a decorator that captures arguments, consults policy, and records the outcome.

import json
import hashlib
from datetime import datetime, timezone
from typing import Any
from opentelemetry import trace

tracer = trace.get_tracer("agent.auditor")


class AuditProxy:
    def __init__(self, policy_checker):
        self.policy_checker = policy_checker
        self._chain_prefix = "GENESIS"

    def exec(self, tool_name: str, args: dict[str, Any]) -> dict[str, Any]:
        with tracer.start_as_current_span("audit.gate") as span:
            span.set_attribute("tool.name", tool_name)
            span.set_attribute("tool.args", json.dumps(args, sort_keys=True))

            decision = self.policy_checker.authorize(tool_name, args)
            span.set_attribute("policy.decision", decision)

            if decision != "allow":
                return self._record(
                    {"type": "denied", "tool": tool_name, "reason": decision},
                    span,
                )

            result = self._dispatch(tool_name, args)
            return self._record(
                {"type": "executed", "tool": tool_name, "args": args,
                 "result_summary": str(result)[:500]},
                span,
            )

    def _dispatch(self, tool_name, args):
        # Route to the real tool handler registered by the agent framework
        return self._tool_registry[tool_name](**args)

    def _record(self, event: dict, span) -> dict[str, Any]:
        event["ts"] = datetime.now(timezone.utc).isoformat()
        event["prev_hash"] = self._chain_prefix
        body = json.dumps(event, sort_keys=True).encode()
        event["hash"] = hashlib.sha256(body).hexdigest()
        self._chain_prefix = event["hash"]
        span.set_attribute("audit.chain_hash", event["hash"])
        return event

Every decision produces an immutable, hash-chained entry. Even denied attempts are recorded — rejections are some of the most important audit evidence.

Policy-as-Code with a Decision Gate

Rather than hard-coding if checks, express authorization as data. The auditor calls a checker that returns an allow/deny verdict plus a human-readable reason.

class PolicyChecker:
    def __init__(self, rules: list):
        self.rules = rules

    def authorize(self, tool_name: str, args: dict) -> str:
        for rule in self.rules:
            if rule["tool"] != tool_name:
                continue
            if rule["when"](args) and not rule["allowed"](args):
                return "deny: " + rule["reason"]
        return "allow"

Example rules in practice:

POLICIES = [
    {
        "tool": "refund_order",
        "when": lambda a: a.get("amount", 0) > 100,
        "allowed": lambda a: a.get("manager_approved") is True,
        "reason": "refunds over $100 require manager approval",
    },
    {
        "tool": "call_external_crm",
        "when": lambda a: "email" in a.get("fields", []) or "phone" in a.get("fields", []),
        "allowed": lambda _a: False,
        "reason": "PII must not leave our data plane",
    },
]

The aim is that the auditor becomes the single choke point for all risky operations, so new regulations are a config change rather than a code rewrite.

Replaying a Workflow from the Trace

Observability is only useful if you can reconstruct what happened. A trace store lets you replay a single decision end-to-end.

def replay_decision(chain_id: str) -> list[dict]:
    """Return every audited event in a chain, oldest to newest."""
    return [e for e in load_all_events() if e["chain_id"] == chain_id]

Add capability to replay with what-if arguments — feed the stored trace into a fresh Gemini 3 call with a different policy set to see whether the rejection would hold under updated rules.

Producing an Attestation Report

For compliance you can’t just hand an auditor a JSON stream. Produce a signed, structured report that asserts conformance.

import hmac

def build_report(chain, secret) -> dict:
    report = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "event_count": len(chain),
        "root_hash": chain[-1]["hash"],
        "violations": [e for e in chain if e["type"] == "denied"],
        "verified": verify_chain(chain),
    }
    report["signature"] = hmac.new(
        secret, json.dumps(report, sort_keys=True).encode(), hashlib.sha256
    ).hexdigest()
    return report

Signed reports remain valid proof that a workflow was governed, even years after the events — provided the chain itself is integrity-checked.

Practical Implementation Steps

1. Wrap tool dispatch behind an AuditProxy so nothing side-effecting runs ungated.
2. Instrument with OpenTelemetry spans tagged with workflow, business, and tool metadata.
3. Codify policies as data and route them through a decision gate.
4. Hash-chain every event for tamper evidence.
5. Expose a replay + report endpoint for your compliance team and incident response.

Conclusion

Autonomous agents earn our trust the same way human teams do: by being auditable. The Agent Auditor pattern separates acting from accounting — executing with speed while recording, validating, and attesting every decision.

Start with the smallest risky tool in your workflow. Wrap it in an audit proxy, tag it with a policy, and hash-chain the outcomes. Over time you’ll widen the net until every Gemini 3 workflow is provably governed. In the agentic era, transparency is not a nice-to-have — it is the product.