Approval List And Resolve

1"""
2Approval List And Resolve
3=============================
4
5Full approval lifecycle: pause, list, filter, resolve, delete.
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_lifecycle_test.db"
18
19
20@approval
21@tool(requires_confirmation=True)
22def delete_user_data(user_id: str) -> str:
23    """Permanently delete all data for a user. This is irreversible.
24
25    Args:
26        user_id (str): The user ID whose data should be deleted.
27    """
28    return f"All data for user {user_id} has been permanently deleted."
29
30
31@approval
32@tool(requires_confirmation=True)
33def send_bulk_email(subject: str, recipient_count: int) -> str:
34    """Send a bulk email to many recipients.
35
36    Args:
37        subject (str): Email subject.
38        recipient_count (int): Number of recipients.
39    """
40    return f"Bulk email '{subject}' sent to {recipient_count} recipients."
41
42
43# ---------------------------------------------------------------------------
44# Create Agent
45# ---------------------------------------------------------------------------
46db = SqliteDb(
47    db_file=DB_FILE, session_table="agent_sessions", approvals_table="approvals"
48)
49agent = Agent(
50    name="Admin Agent",
51    model=OpenAIResponses(id="gpt-5-mini"),
52    tools=[delete_user_data, send_bulk_email],
53    markdown=True,
54    db=db,
55)
56
57# ---------------------------------------------------------------------------
58# Run Agent
59# ---------------------------------------------------------------------------
60if __name__ == "__main__":
61    # Clean up from previous runs
62    if os.path.exists(DB_FILE):
63        os.remove(DB_FILE)
64    os.makedirs("tmp", exist_ok=True)
65
66    # Re-create after cleanup
67    db = SqliteDb(
68        db_file=DB_FILE, session_table="agent_sessions", approvals_table="approvals"
69    )
70    agent = Agent(
71        name="Admin Agent",
72        model=OpenAIResponses(id="gpt-5-mini"),
73        tools=[delete_user_data, send_bulk_email],
74        markdown=True,
75        db=db,
76    )
77
78    # === Scenario 1: Trigger a pause and approval ===
79    print("=== Scenario 1: Delete user data (triggers approval) ===")
80    run1 = agent.run("Delete all data for user U-12345")
81    assert run1.is_paused, f"Expected pause, got {run1.status}"
82    print(f"Agent paused. Run ID: {run1.run_id}")
83
84    # === Scenario 2: Trigger another pause ===
85    print("\n=== Scenario 2: Send bulk email (triggers approval) ===")
86    run2 = agent.run("Send a bulk email with subject 'Holiday Sale' to 5000 recipients")
87    assert run2.is_paused, f"Expected pause, got {run2.status}"
88    print(f"Agent paused. Run ID: {run2.run_id}")
89
90    # === List all pending approvals ===
91    print("\n=== Listing all pending approvals ===")
92    approvals_list, total = db.get_approvals(status="pending")
93    print(f"Total pending: {total}")
94    assert total == 2, f"Expected 2 pending approvals, got {total}"
95    for a in approvals_list:
96        print(
97            f"  [{a['id'][:8]}...] run={a['run_id'][:8]}... context={a.get('context')}"
98        )
99
100    # === Get count ===
101    print("\n=== Pending approval count ===")
102    count = db.get_pending_approval_count()
103    print(f"Count: {count}")
104    assert count == 2
105
106    # === Filter by run_id ===
107    print(f"\n=== Filter by run_id: {run1.run_id[:8]}... ===")
108    filtered, filtered_total = db.get_approvals(run_id=run1.run_id)
109    print(f"Found: {filtered_total}")
110    assert filtered_total == 1
111    approval1 = filtered[0]
112
113    # === Get single approval ===
114    print(f"\n=== Get approval by ID: {approval1['id'][:8]}... ===")
115    single = db.get_approval(approval1["id"])
116    assert single is not None
117    print(f"  Status: {single['status']}")
118    print(f"  Source: {single['source_type']}")
119
120    # === Resolve first approval (approve) ===
121    print("\n=== Resolving first approval (approve) ===")
122    resolved = db.update_approval(
123        approval1["id"],
124        expected_status="pending",
125        status="approved",
126        resolved_by="admin@example.com",
127        resolved_at=int(time.time()),
128    )
129    assert resolved is not None
130    assert resolved["status"] == "approved"
131    print(f"  Status: {resolved['status']}")
132    print(f"  Resolved by: {resolved['resolved_by']}")
133
134    # === Try to double-resolve (should fail due to expected_status guard) ===
135    print("\n=== Attempting double-resolve (should fail) ===")
136    double = db.update_approval(
137        approval1["id"],
138        expected_status="pending",
139        status="rejected",
140        resolved_by="hacker",
141    )
142    assert double is None, "Double-resolve should return None"
143    print("  Double-resolve correctly blocked (expected_status guard)")
144
145    # === Resolve second approval (reject) ===
146    print("\n=== Resolving second approval (reject) ===")
147    approvals2, _ = db.get_approvals(status="pending")
148    assert len(approvals2) == 1
149    approval2 = approvals2[0]
150    resolved2 = db.update_approval(
151        approval2["id"],
152        expected_status="pending",
153        status="rejected",
154        resolved_by="admin@example.com",
155        resolved_at=int(time.time()),
156    )
157    assert resolved2 is not None
158    assert resolved2["status"] == "rejected"
159    print(f"  Status: {resolved2['status']}")
160
161    # === Verify clean state ===
162    print("\n=== Final state ===")
163    final_count = db.get_pending_approval_count()
164    print(f"Pending approvals: {final_count}")
165    assert final_count == 0
166
167    all_approvals, all_total = db.get_approvals()
168    print(f"Total approvals: {all_total}")
169    assert all_total == 2
170
171    # === Continue the runs ===
172    print("\n=== Continuing run 1 (approved) ===")
173    for req in run1.active_requirements:
174        if req.needs_confirmation:
175            req.confirm()
176    result1 = agent.continue_run(run_id=run1.run_id, requirements=run1.requirements)
177    print(f"  Result: {str(result1.content)[:100]}...")
178
179    print("\n=== Continuing run 2 (rejected) ===")
180    for req in run2.active_requirements:
181        if req.needs_confirmation:
182            req.reject("Rejected by admin: too many recipients")
183    result2 = agent.continue_run(run_id=run2.run_id, requirements=run2.requirements)
184    print(f"  Result: {str(result2.content)[:100]}...")
185
186    # === Delete approvals ===
187    print("\n=== Deleting approval records ===")
188    for a in all_approvals:
189        deleted = db.delete_approval(a["id"])
190        assert deleted, f"Failed to delete approval {a['id']}"
191        print(f"  Deleted: {a['id'][:8]}...")
192
193    final_all, final_total = db.get_approvals()
194    assert final_total == 0
195    print(f"All approvals deleted. Total: {final_total}")
196
197    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 approval_list_and_resolve.py