Governance Recipes

Ready-to-use policy templates, audit queries, and compliance workflows for production agent deployments.

---

Recipe 1: ABAC Policy Templates

Attribute-Based Access Control (ABAC) policies gate what each agent can read, write, or execute. Attach policies to an org via the control plane API or the dashboard.

Time-Based Access

Restrict an agent to business-hours queries only — useful for agents that should not run automated operations overnight.

{
  "policy_id": "business-hours-only",
  "description": "Allow queries only Monday–Friday 08:00–18:00 UTC",
  "effect": "deny",
  "conditions": {
    "time_of_week": {
      "outside_hours": {"start": "08:00", "end": "18:00", "timezone": "UTC"},
      "outside_days": ["Saturday", "Sunday"]
    }
  },
  "actions": ["query", "write"],
  "resources": ["*"]
}

Table Restrictions

Prevent an agent from reading or modifying sensitive tables.

{
  "policy_id": "restrict-pii-tables",
  "description": "Block access to tables containing PII",
  "effect": "deny",
  "conditions": {
    "resource.table": {
      "in": ["users_pii", "payments_raw", "health_records"]
    }
  },
  "actions": ["query", "write", "delete"],
  "resources": ["table:*"]
}

Framework Restrictions

Allow only specific agent frameworks to execute write operations, reducing the attack surface from untrusted integrations.

{
  "policy_id": "trusted-frameworks-write",
  "description": "Only LangChain and CrewAI agents may write",
  "effect": "deny",
  "conditions": {
    "agent.framework": {
      "not_in": ["langchain", "crewai"]
    }
  },
  "actions": ["write", "delete", "branch_merge"],
  "resources": ["*"]
}

When to use:

Template	Scenario
Time-based	Batch agents, scheduled pipelines, after-hours lockdown
Table restrictions	PII compliance, SOC 2, HIPAA scoping
Framework restrictions	Multi-tenant SaaS where customers bring their own agents

---

Recipe 2: CoT Chain Verification Workflow

Chain-of-thought (CoT) ledger entries are cryptographically hash-chained per session. Use this workflow to automatically detect tampering or log corruption.

Automated Chain Validation

import hatidata

client = hatidata.Client(api_key="hd_...")

def verify_session_chain(session_id: str) -> dict:
    """Replay a session and verify every hash link."""
    result = client.cot.replay_session(session_id=session_id)

    broken_links = []
    for i, step in enumerate(result.steps):
        if not step.chain_valid:
            broken_links.append({
                "step_index": i,
                "step_id": step.step_id,
                "expected_hash": step.expected_prev_hash,
                "actual_hash": step.actual_prev_hash,
            })

    return {
        "session_id": session_id,
        "total_steps": len(result.steps),
        "chain_intact": len(broken_links) == 0,
        "broken_links": broken_links,
    }

report = verify_session_chain("session-xyz-789")
if not report["chain_intact"]:
    alert_security_team(report)

Alerting on Broken Chains

import smtplib
from email.message import EmailMessage

def alert_security_team(report: dict):
    msg = EmailMessage()
    msg["Subject"] = f"[ALERT] CoT chain broken — session {report['session_id']}"
    msg["From"] = "hatidata-alerts@yourorg.com"
    msg["To"] = "security@yourorg.com"
    msg.set_content(
        f"Broken links detected:\n\n"
        + "\n".join(str(l) for l in report["broken_links"])
    )
    with smtplib.SMTP("smtp.yourorg.com") as s:
        s.send_message(msg)

---

Recipe 3: Audit Report Generation

Use the HatiData query interface to generate compliance reports directly from the audit log tables.

Daily Access Summary

-- Agent query volume by table, last 24 hours
SELECT
    agent_id,
    table_name,
    COUNT(*)          AS query_count,
    SUM(rows_scanned) AS total_rows_scanned,
    MAX(executed_at)  AS last_seen
FROM _hatidata_audit_log
WHERE executed_at >= NOW() - INTERVAL '1 day'
GROUP BY agent_id, table_name
ORDER BY query_count DESC;

Policy Violation Report

-- Denied queries by policy, last 7 days
SELECT
    policy_id,
    agent_id,
    denied_action,
    COUNT(*) AS denial_count,
    MIN(denied_at) AS first_seen,
    MAX(denied_at) AS last_seen
FROM _hatidata_policy_denials
WHERE denied_at >= NOW() - INTERVAL '7 days'
GROUP BY policy_id, agent_id, denied_action
ORDER BY denial_count DESC;

Data Access by Classification

-- Queries touching classified columns
SELECT
    al.agent_id,
    al.table_name,
    al.column_name,
    dc.classification,
    COUNT(*) AS access_count
FROM _hatidata_audit_log al
JOIN _hatidata_data_classifications dc
    ON al.table_name = dc.table_name
   AND al.column_name = dc.column_name
WHERE al.executed_at >= NOW() - INTERVAL '30 days'
GROUP BY al.agent_id, al.table_name, al.column_name, dc.classification
ORDER BY access_count DESC;

---

Recipe 4: Row-Level Security Patterns

Agent-Scoped Data

Agents only see rows they own. Apply this RLS filter at query time via the proxy.

{
  "rls_rule_id": "agent-scoped-rows",
  "table": "tasks",
  "filter": "agent_id = '{{agent.id}}'",
  "applies_to": ["query", "write"]
}

Department Isolation

Isolate rows by department so finance agents cannot access HR data and vice versa.

{
  "rls_rule_id": "department-isolation",
  "table": "employee_records",
  "filter": "department = '{{agent.attributes.department}}'",
  "applies_to": ["query"]
}

When to use:

Pattern	Scenario
Agent-scoped	Multi-tenant SaaS, per-customer agent deployments
Department isolation	Enterprise deployments with strict data segregation requirements

---

Recipe 5: Trigger-Based Compliance

PII Detection Trigger

Register a semantic trigger that fires whenever an agent stores content resembling personally identifiable information.

client.triggers.register(
    name="pii-detection",
    description="Fire when agent output may contain PII",
    concept="personal information: name, email, phone, SSN, passport, address",
    threshold=0.82,
    action={
        "type": "flag_for_review",
        "reviewer_queue": "compliance-team",
        "priority": "high",
    },
    agent_id="*",  # Apply to all agents in the org
)

Cost Control Trigger

Alert when an agent's estimated query cost exceeds a per-run threshold.

client.triggers.register(
    name="cost-spike-alert",
    description="Alert when single query cost exceeds $5",
    concept="high cost query large scan expensive operation",
    threshold=0.78,
    action={
        "type": "webhook",
        "url": "https://ops.yourorg.com/webhooks/cost-alert",
        "hmac_secret": "whsec_...",
    },
    cooldown_seconds=300,  # Suppress repeat alerts for 5 minutes
)

Configuration tips:

Parameter	Recommendation
threshold	Start at 0.80 for precision; lower to 0.70 for broader detection
cooldown_seconds	Set ≥ 60 to avoid alert storms on bursty workloads
action.type	Use flag_for_review for human-in-the-loop; webhook for automated remediation

---

Related Concepts

• Security Model
• Chain-of-Thought Ledger
• Semantic Triggers
• Memory Patterns
• Multi-Agent Coordination