from __future__ import annotations
from copy import deepcopy
from kernite import evaluate_execute
def apply_reason_code_fixes(payload: dict, reason_codes: list[str]) -> dict:
next_payload = deepcopy(payload)
if "required_currency" in reason_codes and "currency" not in next_payload:
next_payload["currency"] = "USD"
if "amount_over_limit" in reason_codes:
next_payload["amount"] = min(float(next_payload.get("amount", 0)), 1000)
return next_payload
def guarded_agent_write(base_request: dict, max_attempts: int = 3) -> dict:
current_request = deepcopy(base_request)
result = {"data": {"decision": "denied", "reason_codes": []}}
for attempt in range(1, max_attempts + 1):
result = evaluate_execute(
current_request,
idempotency_key=f"agent-attempt-{attempt}",
)
decision = result["data"]["decision"]
reason_codes = list(result["data"]["reason_codes"])
if decision == "approved":
return {"ok": True, "governance": result, "request": current_request}
remediated_payload = apply_reason_code_fixes(
current_request["payload"],
reason_codes,
)
if remediated_payload == current_request["payload"]:
return {"ok": False, "governance": result, "request": current_request}
current_request["payload"] = remediated_payload
return {"ok": False, "governance": result, "request": current_request}