Audit Approval Overview

1"""
2Audit Approval Overview
3=============================
4
5Overview: @approval vs @approval(type="audit") in the same agent.
6"""
7
8import os
9import time
10
11from kern.agent import Agent
12from kern.approval import approval
13from kern.db.sqlite import SqliteDb
14from kern.models.openai import OpenAIResponses
15from kern.tools import tool
16
17DB_FILE = "tmp/approvals_test.db"
18
19
20@approval
21@tool(requires_confirmation=True)
22def critical_action(action: str) -> str:
23    """Execute a critical action that requires pre-approval.
24
25    Args:
26        action (str): The action to execute.
27
28    Returns:
29        str: Result of the action.
30    """
31    return f"Executed critical action: {action}"
32
33
34@approval(type="audit")
35@tool(requires_confirmation=True)
36def sensitive_action(action: str) -> str:
37    """Execute a sensitive action that is logged after completion.
38
39    Args:
40        action (str): The action to execute.
41
42    Returns:
43        str: Result of the action.
44    """
45    return f"Executed sensitive action: {action}"
46
47
48# ---------------------------------------------------------------------------
49# Create Agent
50# ---------------------------------------------------------------------------
51db = SqliteDb(
52    db_file=DB_FILE, session_table="agent_sessions", approvals_table="approvals"
53)
54agent = Agent(
55    model=OpenAIResponses(id="gpt-5-mini"),
56    tools=[critical_action, sensitive_action],
57    markdown=True,
58    db=db,
59)
60
61# ---------------------------------------------------------------------------
62# Run Agent
63# ---------------------------------------------------------------------------
64if __name__ == "__main__":
65    # Clean up from previous runs
66    if os.path.exists(DB_FILE):
67        os.remove(DB_FILE)
68    os.makedirs("tmp", exist_ok=True)
69
70    # Re-create after cleanup
71    db = SqliteDb(
72        db_file=DB_FILE, session_table="agent_sessions", approvals_table="approvals"
73    )
74    agent = Agent(
75        model=OpenAIResponses(id="gpt-5-mini"),
76        tools=[critical_action, sensitive_action],
77        markdown=True,
78        db=db,
79    )
80
81    # Step 1: Run critical action - creates a pending approval record BEFORE execution
82    print("--- Step 1: Running critical action (@approval) ---")
83    run1 = agent.run("Execute the critical action: deploy to production.")
84    print(f"Run status: {run1.status}")
85    assert run1.is_paused, f"Expected paused, got {run1.status}"
86    print("Agent paused as expected.")
87
88    # Verify required approval record was created in DB (pending, before execution)
89    print("\n--- Step 2: Verifying required approval record in DB ---")
90    required_approvals, required_total = db.get_approvals(approval_type="required")
91    print(f"Required approvals (pending): {required_total}")
92    assert required_total >= 1, (
93        f"Expected at least 1 required approval, got {required_total}"
94    )
95    required_approval = required_approvals[0]
96    print(f"  Approval ID:    {required_approval['id']}")
97    print(f"  Status:         {required_approval['status']}")
98    print(f"  Approval type:  {required_approval['approval_type']}")
99    assert required_approval["status"] == "pending", (
100        f"Expected status 'pending', got {required_approval['status']}"
101    )
102    assert required_approval["approval_type"] == "required", (
103        f"Expected type 'required', got {required_approval['approval_type']}"
104    )
105
106    # Confirm and continue the critical action
107    print("\n--- Step 3: Confirming and continuing critical action ---")
108    for requirement in run1.active_requirements:
109        if requirement.needs_confirmation:
110            print(f"  Confirming tool: {requirement.tool_execution.tool_name}")
111            requirement.confirm()
112
113    run1 = agent.continue_run(
114        run_id=run1.run_id,
115        requirements=run1.requirements,
116    )
117    print(f"Run status after continue: {run1.status}")
118    assert not run1.is_paused, "Expected run to complete, but it's still paused"
119
120    # Resolve the required approval in DB
121    resolved = db.update_approval(
122        required_approval["id"],
123        expected_status="pending",
124        status="approved",
125        resolved_by="admin_user",
126        resolved_at=int(time.time()),
127    )
128    assert resolved is not None, "Approval resolution failed"
129    print(f"  Resolved required approval: status={resolved['status']}")
130
131    # Step 4: Run sensitive action - creates an audit approval record AFTER execution
132    print("\n--- Step 4: Running sensitive action (@approval audit) ---")
133    run2 = agent.run("Execute the sensitive action: export user reports.")
134    print(f"Run status: {run2.status}")
135    assert run2.is_paused, f"Expected paused, got {run2.status}"
136    print("Agent paused as expected (confirmation required).")
137
138    # Confirm and continue the sensitive action
139    print("\n--- Step 5: Confirming and continuing sensitive action ---")
140    for requirement in run2.active_requirements:
141        if requirement.needs_confirmation:
142            print(f"  Confirming tool: {requirement.tool_execution.tool_name}")
143            requirement.confirm()
144
145    run2 = agent.continue_run(
146        run_id=run2.run_id,
147        requirements=run2.requirements,
148    )
149    print(f"Run status after continue: {run2.status}")
150    assert not run2.is_paused, "Expected run to complete, but it's still paused"
151
152    # Verify logged approval record was created in DB
153    print("\n--- Step 6: Verifying logged approval record in DB ---")
154    logged_approvals, logged_total = db.get_approvals(approval_type="audit")
155    print(f"Logged approvals: {logged_total}")
156    assert logged_total >= 1, f"Expected at least 1 logged approval, got {logged_total}"
157    logged_approval = logged_approvals[0]
158    print(f"  Approval ID:    {logged_approval['id']}")
159    print(f"  Status:         {logged_approval['status']}")
160    print(f"  Approval type:  {logged_approval['approval_type']}")
161    assert logged_approval["status"] == "approved", (
162        f"Expected status 'approved', got {logged_approval['status']}"
163    )
164    assert logged_approval["approval_type"] == "audit", (
165        f"Expected type 'audit', got {logged_approval['approval_type']}"
166    )
167
168    # Step 7: Query DB filtering by approval_type to show separation
169    print("\n--- Step 7: Querying by approval_type to verify separation ---")
170    required_list, required_count = db.get_approvals(approval_type="required")
171    logged_list, logged_count = db.get_approvals(approval_type="audit")
172
173    print(f"  Required approvals: {required_count}")
174    assert required_count == 1, f"Expected 1 required approval, got {required_count}"
175
176    print(f"  Logged approvals:   {logged_count}")
177    assert logged_count == 1, f"Expected 1 logged approval, got {logged_count}"
178
179    print(
180        f"  Required record: type={required_list[0]['approval_type']}, status={required_list[0]['status']}"
181    )
182    print(
183        f"  Logged record:   type={logged_list[0]['approval_type']}, status={logged_list[0]['status']}"
184    )
185
186    print("\n--- All checks passed! ---")

1# Clone and setup repo
2git clone https://github.com/kern-ai/kern.git
3cd kern/cookbook/02_agents/11_approvals
4
5# Create and activate virtual environment
6./scripts/demo_setup.sh
7source .venvs/demo/bin/activate
8
9python audit_approval_overview.py