"""Local approval/recovery reference. Python 3.10+, SQLite, no API calls on import. Run offline tests: python -m unittest -v test_refund_workflow.py The refund ledger is fictional: this code never transfers money. """ import argparse import hashlib import json import sqlite3 TOOLS = [{ "type": "function", "name": name, "description": description, "parameters": {"type": "object", "properties": {"order_id": {"type": "string"}}, "required": ["order_id"], "additionalProperties": False}, } for name, description in [ ("lookup_order", "Read an authorized fictional order and its refund policy."), ("refund_order", "Request a full fictional refund, subject to human approval."), ]] class Workflow: def __init__(self, path): self.db = sqlite3.connect(path, timeout=10) self.db.row_factory = sqlite3.Row self.db.executescript(""" CREATE TABLE IF NOT EXISTS sessions(id TEXT PRIMARY KEY, owner TEXT NOT NULL); CREATE TABLE IF NOT EXISTS approvals( operation TEXT PRIMARY KEY, owner TEXT, order_id TEXT, amount INTEGER, state TEXT NOT NULL CHECK(state IN ('pending','approved','denied'))); CREATE TABLE IF NOT EXISTS refunds(operation TEXT PRIMARY KEY, amount INTEGER); CREATE TABLE IF NOT EXISTS results( session TEXT, turn TEXT, call TEXT, fingerprint TEXT, result TEXT, PRIMARY KEY(session,turn,call)); """) def close(self): self.db.close() def bind(self, session, owner): # Trusted operator operation, never an endpoint accepting arbitrary user IDs. with self.db: self.db.execute("INSERT OR IGNORE INTO sessions VALUES (?,?)", (session, owner)) self.authorize(session, owner) def authorize(self, session, owner): row = self.db.execute("SELECT owner FROM sessions WHERE id=?", (session,)).fetchone() if not row or row["owner"] != owner: raise ValueError("Session not accessible") def decide(self, operation, owner, decision): if decision not in ("approved", "denied"): raise ValueError("Invalid decision") with self.db: changed = self.db.execute( "UPDATE approvals SET state=? WHERE operation=? AND owner=? AND state='pending'", (decision, operation, owner)).rowcount if not changed: raise ValueError("No pending approval owned by this user") def handle(self, session, owner, action): self.authorize(session, owner) if action.get("type") != "function_call": raise ValueError("Not a function action") if any(not isinstance(action.get(key), str) or not action[key] for key in ("turn_id", "call_id", "name")): raise ValueError("Missing action identity") identity = (session, action["turn_id"], action["call_id"]) fingerprint = json.dumps([action["name"], action.get("arguments")], sort_keys=True) # Serialize the fictional business write and tool-result ledger atomically. self.db.execute("BEGIN IMMEDIATE") try: cached = self.db.execute("SELECT * FROM results WHERE session=? AND turn=? AND call=?", identity).fetchone() if cached: if cached["fingerprint"] != fingerprint: raise ValueError("Call identity reused with different arguments") result = json.loads(cached["result"]) else: try: output = self.execute(owner, action["name"], action.get("arguments")) if output is None: # Persist approval but leave tool action pending. self.db.commit() return None result = {"success": True, "output": json.dumps(output)} except ValueError as error: result = {"success": False, "error": str(error)} self.db.execute("INSERT INTO results VALUES (?,?,?,?,?)", (*identity, fingerprint, json.dumps(result))) self.db.commit() except BaseException: self.db.rollback() raise return {"type": "agent.session.input.tool_result", "turn_id": identity[1], "call_id": identity[2], **result} def execute(self, owner, name, arguments): if name not in ("lookup_order", "refund_order"): raise ValueError("Unknown tool") if not isinstance(arguments, dict) or set(arguments) != {"order_id"}: raise ValueError("Expected only order_id") if owner != "demo-user" or arguments["order_id"] != "ORDER-100": raise ValueError("Order not found or not accessible") # Server-owned fixture and policy; neither amount nor owner comes from the model. amount, operation = 2500, "demo-user:ORDER-100:full-refund:v1" if name == "lookup_order": return {"order_id": "ORDER-100", "paid_cents": amount, "policy": "Full fictional refund of 2500 cents requires approval."} self.db.execute("INSERT OR IGNORE INTO approvals VALUES (?,?,?,?, 'pending')", (operation, owner, "ORDER-100", amount)) approval = self.db.execute("SELECT * FROM approvals WHERE operation=?", (operation,)).fetchone() if approval["state"] == "pending": return None if approval["state"] == "denied": raise ValueError("The user denied this refund") self.db.execute("INSERT OR IGNORE INTO refunds VALUES (?,?)", (operation, amount)) return {"operation": operation, "refunded_cents": amount, "currency": "USD", "fictional": True} def pending(self): return [dict(row) for row in self.db.execute("SELECT * FROM approvals WHERE state='pending'")] def drain_pending(client, workflow, session, owner): """Call after an authenticated notification or on operator-requested recovery. SDK pagination/stream recovery and webhook signature verification are caller concerns. """ workflow.authorize(session, owner) current = client.beta.agents.sessions.retrieve(session) delivered = 0 for model_action in current.required_actions or []: action = model_action.to_dict() result = workflow.handle(session, owner, action) if result is None: continue # Stable across process restarts. Separate from the business operation key. key = hashlib.sha256(json.dumps([session, result], sort_keys=True).encode()).hexdigest() client.beta.agents.sessions.events.create(session, events=[result], idempotency_key=key) delivered += 1 return {"results_submitted": delivered, "pending_approvals": workflow.pending()} if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--db", default="refund-demo.sqlite") parser.add_argument("--session", required=True) parser.add_argument("--owner", default="demo-user") parser.add_argument("command", choices=["bind", "pending", "approve", "deny", "recover", "inspect"]) parser.add_argument("--operation") args = parser.parse_args() workflow = Workflow(args.db) try: if args.command == "bind": workflow.bind(args.session, args.owner) else: workflow.authorize(args.session, args.owner) if args.command in ("approve", "deny"): workflow.decide(args.operation, args.owner, "approved" if args.command == "approve" else "denied") elif args.command in ("recover", "inspect"): from openai import OpenAI with OpenAI(max_retries=0, timeout=30.0) as client: if args.command == "recover": print(json.dumps(drain_pending(client, workflow, args.session, args.owner), indent=2)) else: for turn in client.beta.agents.sessions.turns.list(args.session, order="asc", limit=100): if turn.subagent_id is None: print(json.dumps({"root_turn": turn.id, "status": turn.status})) for item in client.beta.agents.sessions.items.list(args.session, order="asc", limit=100): print(item.to_json()) print(json.dumps({"fictional_refunds": [dict(row) for row in workflow.db.execute("SELECT * FROM refunds")]})) else: print(json.dumps(workflow.pending(), indent=2)) finally: workflow.close()