AI-Powered Payment Recovery Agent

Build an AI agent that diagnoses failed payments and recovers revenue.

Introduction

30 Second Summary

Failed payments cost online businesses billions every year. Most of these failures are actually recoverable if you diagnose the root cause and act fast with the right strategy.

In this project, you will build RecoverAI, a Python-powered revenue recovery agent that ingests simulated Razorpay failed payment events. It diagnoses root causes, selects the mathematically optimal recovery action, and displays results in a live web dashboard.

What You'll Build

You run streamlit run dashboard.py and a polished dashboard opens in your browser, showing 12 failed payments worth 47,500 INR being diagnosed, scored, and recovered in real time.

By the end of this project, you'll have:

A live Streamlit dashboard with KPI metric cards, interactive bar charts, and expandable per-payment decision timelines you can open in any browser and demo on the spot.
An AI pipeline that catches obvious payment failures with deterministic rules and routes ambiguous cases to Gemini 2.5 Flash for structured output diagnosis you can parse and route automatically.
A tamper-evident SHA-256 hash-chained audit log with built-in verification that proves no decision record was altered.
Secret Mission: Add a customer risk scoring system that tracks cross-payment behavior and adjusts recovery aggressiveness per customer.

Are there any prerequisites?

No prior experience with payment systems or AI is needed. This project walks you through every tool and concept from scratch.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

Set Up Your Project and Connect to Gemini

RecoverAI needs a solid foundation before it can process any payment data. You need a project directory, the right Python packages, and a verified connection to Gemini AI.

In this step, you will create the project structure, install three key dependencies, and make your first API call to Gemini 2.5 Flash to prove everything is wired up. You will also add a sample webhook event so you have real data to work with in the next step.

In this step, get ready to:

Create the project directory and install dependencies.
Connect to Gemini and verify the API key works.
Add a sample Razorpay webhook event for testing.

Create your project directory and install dependencies

Every Python project starts with a dedicated folder and the right packages. You will create the recoverai directory, scaffold your empty files, and install three libraries: the google-genai SDK for calling Gemini, Pydantic for data validation, and Streamlit for building the dashboard.

Move to your Desktop by running this command:

cd ~/Desktop

Create your project folder and move into it by running these commands:

mkdir recoverai
cd recoverai

Create the project files by running this command:

touch recover.py models.py dashboard.py sample_events.json requirements.txt

Confirm the files exist by running:

ls

You should see recover.py, models.py, dashboard.py, sample_events.json, and requirements.txt in the output.

Don't see all five files?

Make sure you ran the touch command from inside the recoverai folder. Run pwd to check your current location.

Help me figure out why my files are missing

Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
Type VS Code and press Enter to open it.
Click File in the top menu bar.
Click Open Folder.
Select the recoverai folder on your Desktop, then click Open.

You should see all five files listed in the Explorer sidebar on the left.

Click on requirements.txt in the Explorer sidebar to open it.
Paste the following dependency list into the file:

google-genai==2.20.0
pydantic==2.13.4
streamlit==1.62.0

What are these three packages?

google-genai is Google's SDK for calling Gemini models. It handles authentication, request formatting, and response parsing.
pydantic validates and structures data using Python type hints. You will use it to parse messy webhook JSON into clean, typed objects.
streamlit turns Python scripts into interactive web dashboards with zero HTML or CSS.

Save requirements.txt by pressing Cmd+S (macOS) or Ctrl+S (Windows).
Install all three packages by running this command in your terminal:

pip install -r requirements.txt

You should see output ending with Successfully installed followed by a list of packages including google-genai, pydantic, and streamlit.

Seeing a permission error or command not found?

If pip is not recognized, try pip3 instead. If you get a permission error, add --user to the end of the command.

Help me fix my pip install

Connect to Gemini and verify the API key

Before building any pipeline logic, you need to confirm that your Gemini API key works. You will set the key as an environment variable and write a short test script that makes a single API call.

Set your Gemini API key as an environment variable by running the command for your operating system:

macOS

Run this in your terminal:

export GEMINI_API_KEY=your-api-key-here

Replace your-api-key-here with the API key from your Google AI Studio account.

Windows

Run this in your terminal:

set GEMINI_API_KEY=your-api-key-here

Replace your-api-key-here with the API key from your Google AI Studio account.

Where do I get a Gemini API key?

Go to Google AI Studio, sign in with your Google account, and click Create API Key. The free tier is more than enough for this project.

Click on recover.py in the VS Code Explorer sidebar to open it.
Paste this code into recover.py to create the Gemini connection test:

import json
import os
from google import genai

def main():
    # Read the API key from the environment variable
    api_key = os.environ.get("GEMINI_API_KEY", "")
    if not api_key:
        raise ValueError("Set the GEMINI_API_KEY environment variable")

    # Create the Gemini client
    client = genai.Client(api_key=api_key)

    # Make a test call to verify the connection works
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents="Respond with just the word: connected",
    )
    print(f"Gemini connected: gemini-2.5-flash")

if __name__ == "__main__":
    main()

What does this code do?

os.environ.get("GEMINI_API_KEY") reads your API key from the environment variable you just set. This keeps the key out of your source code.
genai.Client(api_key=api_key) creates a client object that handles all communication with the Gemini API.
client.models.generate_content() sends a simple prompt to gemini-2.5-flash. If the call succeeds, your API key and SDK are working correctly.
import json is not used yet. You will need it in the next step to load your webhook data.

Save recover.py.

Before you run this, what do you think will happen? Will the script print the connection message, or will it raise an error?

Run the test script by executing this command:

python recover.py

You should see Gemini connected: gemini-2.5-flash printed to the terminal. This confirms your API key is valid and the SDK is installed correctly.

Seeing a ValueError or authentication error?

If you see Set the GEMINI_API_KEY environment variable, your environment variable is not set. Re-run the export (macOS) or set (Windows) command in the same terminal session where you run the script.

If you see an API key error from Google, double-check that you copied the full key from Google AI Studio with no extra spaces.

Help me fix my Gemini connection

✔️ Awesome, I've got everything!

Great. Make sure you have saved recover.py and requirements.txt before moving on.

ⓧ I'd like to double check the full code

Here is the full recover.py so far:

import json
import os
from google import genai

def main():
    api_key = os.environ.get("GEMINI_API_KEY", "")
    if not api_key:
        raise ValueError("Set the GEMINI_API_KEY environment variable")
    client = genai.Client(api_key=api_key)
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents="Respond with just the word: connected",
    )
    print(f"Gemini connected: gemini-2.5-flash")

if __name__ == "__main__":
    main()

And here is requirements.txt:

google-genai==2.20.0
pydantic==2.13.4
streamlit==1.62.0

Create your first sample webhook event

RecoverAI processes Razorpay payment.failed webhook events. These are JSON payloads that Razorpay sends when a customer's payment fails, containing details like the payment amount, method, and error information.

Click on sample_events.json in the VS Code Explorer sidebar to open it.
Paste this first section of the webhook event into the file:

[
  {
    "entity": "event",
    "account_id": "acc_CJoeHMNpi0nC7k",
    "event": "payment.failed",
    "contains": ["payment"],
    "payload": {
      "payment": {
        "entity": {
          "id": "pay_F1a2b3c4d5e6f7",
          "entity": "payment",
          "amount": 150000,
          "currency": "INR",
          "status": "failed",
          "order_id": "order_F1a2b3c4d5e6f7",
          "invoice_id": null,
          "international": false,
          "method": "card",
          "amount_refunded": 0,
          "refund_status": null,
          "captured": false,
          "description": "Order #1001",
          "card_id": "card_F1a2b3c4d5e6f7",
          "bank": null,
          "wallet": null,
          "vpa": null,
          "email": "rahul.sharma@example.com",
          "contact": "+919876543210",
          "notes": {},

Directly below the last line you pasted, add these remaining fields to complete the event:

          "fee": null,
          "tax": null,
          "error_code": "BAD_REQUEST_ERROR",
          "error_description": "Payment was unsuccessful due to insufficient account balance",
          "error_source": "bank",
          "error_step": "payment_authorization",
          "error_reason": "insufficient_balance",
          "created_at": 1693500000
        }
      }
    },
    "created_at": 1693500001
  }
]

What is inside this webhook event?

This is a simulated Razorpay payment.failed webhook event. The payment entity is nested inside payload.payment.entity.

amount is in paise (100 paise = 1 INR), so 150000 means 1,500 INR.
error_code, error_description, and error_reason tell you why the payment failed. This one failed because of insufficient funds.
method indicates how the customer tried to pay (card, UPI, netbanking, or wallet). You will see all four in the next step.

Save sample_events.json.
Verify the JSON is valid by running this command:

python -c "import json; data = json.load(open('sample_events.json')); print(f'{len(data)} event(s) loaded successfully')"

You should see 1 event(s) loaded successfully. If you see a JSONDecodeError, double-check that both code blocks were pasted in order with no gaps or extra characters between them.

Seeing a JSONDecodeError?

The most common cause is a missing comma or bracket where the two pasted sections meet. Open sample_events.json and check that the line ending in "notes": {}, is immediately followed by "fee": null, with no blank lines or duplicate characters.

Help me fix my sample_events.json

✔️ Awesome, I've got everything!

Great. Make sure sample_events.json is saved before moving on.

ⓧ I'd like to double check the full code

Scroll up to the two JSON code blocks you pasted. Confirm that the first block (starting with [ and ending at "notes": {},) is followed immediately by the second block (starting at "fee": null, and ending with ]). Both sections together form a single valid JSON array with one event object.

Your project foundation is in place. You have a verified Gemini connection and a real webhook event to work with. Next up, you will build Pydantic models to parse all 12 payment events and get your first AI diagnosis of a failed payment.

Parse Webhooks and Get Your First AI Diagnosis

In Step 1 you proved your Gemini connection works. Now it is time to feed it real data.

Payment gateways like Razorpay send failure events as webhook payloads. These arrive as deeply nested JSON blobs with amounts in paise, optional fields that may be null, and error details buried three levels deep. Before your pipeline can do anything useful, you need Pydantic models to validate that messy JSON into clean, typed Python objects. Then you will send one of those objects to Gemini and see what an AI diagnosis looks like.

In this step, get ready to:

Define Pydantic models that parse Razorpay webhook JSON into typed Python objects.
Load and parse all 12 sample events, printing a one-line summary for each.
Send a failed payment to Gemini for a plain text diagnosis.

Define the Pydantic models

Razorpay's payment.failed webhook sends a nested JSON payload. The payment amount arrives in paise (the smallest unit of INR, where 100 paise = 1 rupee), and many fields like bank, wallet, and vpa may be null depending on the payment method.

In models.py, paste the following code to start building the PaymentEntity model:

import enum
from typing import Optional
from pydantic import BaseModel, Field

class PaymentEntity(BaseModel):
    # Core payment identification
    id: str
    entity: str = "payment"
    amount: int = Field(description="Amount in paise")
    currency: str = "INR"
    status: str = "failed"
    order_id: Optional[str] = None
    invoice_id: Optional[str] = None
    international: bool = False
    method: str
    amount_refunded: int = 0
    refund_status: Optional[str] = None
    captured: bool = False
    description: Optional[str] = None
    # Payment method details
    card_id: Optional[str] = None
    bank: Optional[str] = None
    wallet: Optional[str] = None
    vpa: Optional[str] = None
    # Customer contact info
    email: Optional[str] = None
    contact: Optional[str] = None
    notes: Optional[dict] = None
    fee: Optional[int] = None
    tax: Optional[int] = None

What does this code do?

PaymentEntity is a Pydantic model that mirrors the fields inside Razorpay's webhook payment.entity object. Each field is typed (str, int, bool) and optional fields default to None.

The enum import is not used yet. You will need it in the next step when you build the rules engine.

Still in models.py, add the error fields, a helper property, and two more models directly below the tax field:

    # Error details from Razorpay
    error_code: Optional[str] = None
    error_description: Optional[str] = None
    error_source: Optional[str] = None
    error_step: Optional[str] = None
    error_reason: Optional[str] = None
    created_at: int

    # Convert amount from paise (smallest currency unit) to rupees
    @property
    def amount_inr(self) -> float:
        return self.amount / 100

class PaymentPayload(BaseModel):
    entity: PaymentEntity

class WebhookEvent(BaseModel):
    entity: str = "event"
    account_id: str
    event: str
    contains: list[str]
    payload: dict
    created_at: int

    # Extract PaymentEntity from nested webhook payload
    def get_payment(self) -> PaymentEntity:
        return PaymentEntity(**self.payload["payment"]["entity"])

What do these models do?

The five error_* fields capture why the payment failed. Razorpay populates error_source with who is responsible (bank, customer, business) and error_step with where the failure happened (payment_authorization, payment_authentication).

The amount_inr property divides paise by 100 so you always work in rupees. WebhookEvent wraps the outer event structure and its get_payment() method digs into the nested payload dict to extract a typed PaymentEntity.

Save models.py.
Confirm the models load correctly by running this command:

python -c "from models import WebhookEvent, PaymentEntity; print('Models loaded OK')"

You should see Models loaded OK printed with no errors.

Seeing an ImportError or ModuleNotFoundError?

Make sure you are running the command from inside the recoverai folder, not a parent directory.
Check that models.py is saved and has no indentation errors. Every field inside PaymentEntity must be indented one level under the class definition.
help me fix an ImportError when loading my Pydantic models

✔️ Awesome, I've got everything!

Great. Double check you have saved models.py before moving on.

ⓧ I'd like to double check the full code

Your models.py should contain both chunks in order. Scroll up and confirm:

Chunk 1: the imports (enum, Optional, BaseModel, Field) and the PaymentEntity fields from id through tax.
Chunk 2: the error fields (error_code through error_reason), created_at, the amount_inr property, PaymentPayload, and WebhookEvent with its get_payment() method.

Load and parse the webhook events

Your models are ready. Now you need recover.py to load the 12 sample events from sample_events.json, parse each one through the Pydantic models, and print a summary so you can see what you are working with.

Replace the entire contents of recover.py with this updated version:

import json
import os
from google import genai
from models import WebhookEvent

# Load webhook events from JSON file
def load_events(filepath: str) -> list[WebhookEvent]:
    with open(filepath) as f:
        raw = json.load(f)
    return [WebhookEvent(**event) for event in raw]

def main():
    # Verify Gemini connection
    api_key = os.environ.get("GEMINI_API_KEY", "")
    if not api_key:
        raise ValueError("Set the GEMINI_API_KEY environment variable")
    client = genai.Client(api_key=api_key)
    client.models.generate_content(
        model="gemini-2.5-flash",
        contents="Respond with just the word: connected",
    )
    print("Gemini connected: gemini-2.5-flash")
    # Parse all events and print summaries
    events = load_events("sample_events.json")
    payments = [e.get_payment() for e in events]
    for p in payments:
        print(f"{p.id} | {p.amount_inr:.2f} INR | {p.method} | {p.error_description}")

if __name__ == "__main__":
    main()

What changed here?

load_events() reads the JSON file and wraps each raw event in a WebhookEvent Pydantic model. Invalid data would raise a validation error immediately.
Inside main(), get_payment() extracts each nested PaymentEntity and the for loop prints a one-line summary using the amount_inr property you just built.

Save recover.py and run it:

python recover.py

You should see Gemini connected: gemini-2.5-flash followed by 12 summary lines, each showing a payment ID, amount in INR, payment method, and error description.

Seeing a KeyError or ValidationError?

A KeyError: 'payment' means your get_payment() method is looking for the wrong key. Check that it reads self.payload["payment"]["entity"] to match the nested webhook structure.
A Pydantic ValidationError means a required field is missing or has the wrong type. Compare your PaymentEntity fields against the JSON structure in sample_events.json.
help me fix a parsing error with my Razorpay webhook Pydantic models

Send a payment to Gemini for plain text diagnosis

You can see 12 failures in your terminal. Now you want to know what an AI thinks about one of them. You will create a dedicated ai_engine.py module to keep all Gemini logic in one place, then call it from recover.py.

Create a new file called ai_engine.py in the recoverai folder and paste the following code to set up a shared Gemini client factory:

import os
from google import genai
from models import PaymentEntity

# Shared Gemini client factory
def get_client() -> genai.Client:
    api_key = os.environ.get("GEMINI_API_KEY", "")
    if not api_key:
        raise ValueError(
            "Set the GEMINI_API_KEY environment variable with your Google AI Studio key"
        )
    return genai.Client(api_key=api_key)

Why move the client into its own module?

Right now recover.py creates the Gemini client directly. Soon, dashboard.py will also need a client. get_client() centralizes the API key check so every module reuses the same factory.

Still in ai_engine.py, add this function below get_client():

# Send payment details to Gemini for plain text diagnosis
def diagnose_payment_plain_text(
    client: genai.Client, payment: PaymentEntity
) -> str:
    prompt = f"""Analyze this failed Razorpay payment.

Payment ID: {payment.id}
Amount: {payment.amount_inr} INR
Method: {payment.method}
Error: {payment.error_code} - {payment.error_description}
Reason: {payment.error_reason}

What is the root cause and what recovery action do you recommend?
"""
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
    )
    return response.text

What does this function do?

diagnose_payment_plain_text() builds an f-string prompt that fills in the payment's key details (ID, amount, method, error info). It sends the prompt to gemini-2.5-flash and returns the raw text response.

This gives you a human-readable diagnosis. You will see in the next step why plain text output creates a problem for automated pipelines.

Save ai_engine.py.
Confirm the module loads correctly by running:

python -c "from ai_engine import get_client, diagnose_payment_plain_text; print('ai_engine loaded OK')"

You should see ai_engine loaded OK printed with no errors.

✔️ Awesome, I've got everything!

Great. Double check you have saved ai_engine.py before moving on.

ⓧ I'd like to double check the full code

Your ai_engine.py should contain both chunks in order. Scroll up and confirm:

Chunk 1: the imports (os, genai, PaymentEntity) and the get_client() factory function.
Chunk 2: the diagnose_payment_plain_text() function with the f-string prompt and generate_content call.

Now you need to update recover.py to use the new ai_engine module and send one payment for diagnosis.

In recover.py, find the line from models import WebhookEvent. Add this import directly below it:

from ai_engine import get_client, diagnose_payment_plain_text

In main(), find these lines that create the client directly:

    api_key = os.environ.get("GEMINI_API_KEY", "")
    if not api_key:
        raise ValueError("Set the GEMINI_API_KEY environment variable")
    client = genai.Client(api_key=api_key)

Replace those four lines with this single line, so the client now comes from your shared factory:

    client = get_client()

Find the for loop that prints payment summaries. Add this code directly below it, at the end of main():

    # Send the first payment for a plain text AI diagnosis
    first_ambiguous = payments[0]
    diagnosis_text = diagnose_payment_plain_text(client, first_ambiguous)
    print("\nSample Gemini diagnosis (plain text):")
    print(diagnosis_text)

What did you just wire up?

recover.py now delegates client creation to get_client() from ai_engine.py. After printing the 12 payment summaries, it picks the first payment and sends it to Gemini. The raw text response shows you what Gemini thinks went wrong and what action it recommends.

Save recover.py.

Before you run this, think about what you expect to see. You already know the first payment is a 1,500.00 INR card payment that failed due to insufficient funds. What kind of response do you think Gemini will produce?

Run the updated pipeline:

python recover.py

You should see Gemini connected: gemini-2.5-flash followed by 12 summary lines, then a block of plain text from Gemini analyzing the first payment's failure. The diagnosis explains the root cause and suggests a recovery action.

Why does the plain text matter?

Read the Gemini response carefully. It is helpful for a human, but it is free-form prose. Your pipeline cannot parse "I recommend sending a payment link" from a wall of text to decide what to do next.

In the next step, you will fix this by asking Gemini to return structured JSON that your code can parse directly into a typed Python object.

Seeing a connection or API error?

Make sure your GEMINI_API_KEY environment variable is still set in this terminal session. If you opened a new terminal, you need to export it again.
If you see a rate limit error, wait 60 seconds and try again. The free tier allows 10 requests per minute.
help me fix a Gemini API connection error in my recover.py script

✔️ Awesome, I've got everything!

Great. Double check you have saved recover.py before moving on.

ⓧ I'd like to double check the full code

Your recover.py should contain these sections in order. Scroll up and confirm each is present:

Imports: json, os, genai, WebhookEvent, get_client, and diagnose_payment_plain_text.
The load_events() function that reads and parses the JSON file.
main() calling get_client(), verifying the connection, loading events, printing summaries, then sending the first payment to diagnose_payment_plain_text() and printing the result.
The if __name__ == "__main__" guard at the bottom.

Your pipeline can now parse raw webhook data into clean Python objects and get an AI opinion on a failure. But that opinion is just a wall of text. In the next step, you will build a rules engine to handle obvious cases without any AI call, and switch Gemini to structured JSON output so every diagnosis is machine-readable.

Build the Rules Engine and Structured AI Diagnosis

In Step 2, you parsed all 12 webhook events into clean Python objects and got your first AI diagnosis from Gemini. Every failed payment is now ready for processing.

But there are two problems. Sending every failure to Gemini is slow and wasteful when obvious patterns like fraud flags or insufficient funds have known fixes. And the plain text diagnosis you printed in Step 2 is just a blob of prose. You cannot extract the recovery probability or recommended actions from it programmatically. This step solves both problems: a rules engine for clear-cut cases, and structured JSON output for the rest.

In this step, get ready to:

Build a deterministic rules engine that classifies obvious failure patterns without any AI call.
Add structured Gemini output so ambiguous failures return typed, parseable diagnosis objects.
Wire both paths into the pipeline so every payment gets a machine-readable classification.

Build the deterministic rules engine

The rules engine needs a vocabulary of failure types and a result container to hold each classification decision. You will define these in models.py first, then build the classifier function in rules.py.

In models.py, scroll to the bottom of the file (below the WebhookEvent class) and add the following two models:

class FailureCategory(str, enum.Enum):
    INSUFFICIENT_FUNDS = "insufficient_funds"
    FRAUD = "fraud"
    BANK_DECLINE = "bank_decline"
    NETWORK_TIMEOUT = "network_timeout"
    INTERNATIONAL_BLOCK = "international_block"
    CARD_EXPIRED = "card_expired"
    SERVER_ERROR = "server_error"
    SESSION_EXPIRED = "session_expired"
    UNKNOWN = "unknown"


class ClassificationResult(BaseModel):
    category: FailureCategory
    confidence: float
    rule_name: Optional[str] = None
    needs_ai: bool = False
    recovery_probability: float = 0.0

What do these models do?

FailureCategory is a string enum listing every failure type your rules engine can recognise: fraud, insufficient funds, network timeouts, and more. ClassificationResult holds the outcome of a classification: the matched category, a confidence score, the rule that fired, and a needs_ai flag that tells the pipeline whether to route this payment to Gemini.

Save models.py.

Now build the classifier function that checks error fields against known patterns.

Open rules.py (currently empty) and add the following code to create the initial classifier with two patterns and a catch-all fallback:

from models import PaymentEntity, ClassificationResult, FailureCategory


def classify_payment(payment: PaymentEntity) -> ClassificationResult:
    # Normalize error fields for case-insensitive matching
    desc = (payment.error_description or "").lower()
    reason = (payment.error_reason or "").lower()
    code = (payment.error_code or "").upper()

    if "fraud" in desc or "fraud" in reason or reason == "suspected_fraud":
        return ClassificationResult(
            category=FailureCategory.FRAUD,
            confidence=0.95,
            rule_name="fraud_keyword_match",
            needs_ai=False,
            recovery_probability=0.0,
        )

    if "insufficient" in desc or reason == "insufficient_balance":
        return ClassificationResult(
            category=FailureCategory.INSUFFICIENT_FUNDS,
            confidence=0.90,
            rule_name="insufficient_funds_match",
            needs_ai=False,
            recovery_probability=0.3,
        )

    return ClassificationResult(
        category=FailureCategory.UNKNOWN,
        confidence=0.0,
        rule_name=None,
        needs_ai=True,
        recovery_probability=0.0,
    )

What does this code do?

Each if block checks the payment's error fields for known keywords. When a pattern matches, the function returns immediately with the appropriate FailureCategory and a pre-set recovery_probability.

If no pattern matches, the catch-all at the bottom returns UNKNOWN with needs_ai=True. These payments will be routed to Gemini later for structured diagnosis.

Save rules.py.

Test the rules engine by wiring it into your existing pipeline.

In recover.py, find this import line near the top:

from ai_engine import get_client, diagnose_payment_plain_text

Add a new import line directly below it:

from rules import classify_payment

Scroll to the bottom of the main() function (below the print(diagnosis_text) line) and add this classification loop:

    # Test the rules engine
    print("\n--- Rules Engine Classifications ---")
    for p in payments:
        result = classify_payment(p)
        source = "DETERMINISTIC" if not result.needs_ai else "NEEDS_AI"
        print(f"  {p.id} | {source} | {result.category.value}")

Save recover.py and run it:

python recover.py

Below the existing event summaries and plain text diagnosis, you should see a new section headed --- Rules Engine Classifications ---. Two payments show as DETERMINISTIC (the fraud and insufficient funds cases). The other 10 show as NEEDS_AI with category unknown.

Not seeing the classifications?

Make sure models.py is saved with the FailureCategory and ClassificationResult classes below WebhookEvent.
Check that your rules.py import matches exactly: from models import PaymentEntity, ClassificationResult, FailureCategory.
Confirm the classification loop is indented inside main(), not at the module level.

Still stuck? Help me debug my rules engine import or classification loop.

Ten payments falling through to UNKNOWN means the rules engine needs more patterns. Time to add them.

In rules.py, find the catch-all return ClassificationResult( at the bottom of classify_payment. Add the following three patterns directly above that fallback return:

    if reason == "international_not_allowed" or "domestic" in desc:
        return ClassificationResult(
            category=FailureCategory.INTERNATIONAL_BLOCK,
            confidence=0.95,
            rule_name="international_block_match",
            needs_ai=False,
            recovery_probability=0.0,
        )

    if "expired" in desc or reason == "card_expired":
        return ClassificationResult(
            category=FailureCategory.CARD_EXPIRED,
            confidence=0.90,
            rule_name="card_expired_match",
            needs_ai=False,
            recovery_probability=0.1,
        )

    if "timeout" in desc or reason == "upi_timeout":
        return ClassificationResult(
            category=FailureCategory.NETWORK_TIMEOUT,
            confidence=0.85,
            rule_name="timeout_match",
            needs_ai=False,
            recovery_probability=0.7,
        )

Still above the fallback return, add these four more patterns below the timeout check you just pasted:

    if code == "SERVER_ERROR" or reason == "server_error":
        return ClassificationResult(
            category=FailureCategory.SERVER_ERROR,
            confidence=0.85,
            rule_name="server_error_match",
            needs_ai=False,
            recovery_probability=0.8,
        )

    if "session" in desc or reason == "session_expired":
        return ClassificationResult(
            category=FailureCategory.SESSION_EXPIRED,
            confidence=0.80,
            rule_name="session_expired_match",
            needs_ai=False,
            recovery_probability=0.6,
        )

    if "declined" in desc or "decline" in reason:
        return ClassificationResult(
            category=FailureCategory.BANK_DECLINE,
            confidence=0.60,
            rule_name=None,
            needs_ai=True,
            recovery_probability=0.0,
        )

Why does bank_decline set needs_ai to True?

A generic "card declined" message could mean anything: the card has a spending cap, the bank flagged the transaction, or there is a temporary hold. The rules engine does not have enough context to determine the root cause. Setting needs_ai=True routes these ambiguous cases to Gemini for deeper analysis.

Save rules.py and run the pipeline again:

python recover.py

The classifications section now shows most payments as DETERMINISTIC with specific categories like fraud, network_timeout, and card_expired. Only a few ambiguous payments show NEEDS_AI.

Seeing more NEEDS_AI than expected?

Make sure every new if block is inside the classify_payment function (indented by 4 spaces), not at the module level.
Confirm the new patterns sit above the catch-all return ClassificationResult(category=FailureCategory.UNKNOWN, ...) block. Patterns below the fallback never execute.

Still stuck? Help me fix my rules.py pattern matching order.

✔️ Awesome, I've got everything!

Make sure you have saved rules.py and models.py.

ⓧ I'd like to double check the full code

Your rules.py should have, in order:

The import line: from models import PaymentEntity, ClassificationResult, FailureCategory.
The classify_payment function with three setup lines (desc, reason, code).
Pattern checks in this order: fraud, insufficient_funds, international_block, card_expired, network_timeout, server_error, session_expired, bank_decline.
The catch-all return ClassificationResult(category=FailureCategory.UNKNOWN, ...) at the very bottom of the function.

Scroll up and confirm each block is present and in order. The fraud and insufficient_funds checks should have needs_ai=False. The bank_decline and unknown checks should have needs_ai=True.

Add structured Gemini output

Look at the plain text diagnosis still printing in your terminal from Step 2. It is a paragraph of prose. Imagine trying to extract a numeric recovery_probability from that text programmatically. You would need fragile regex patterns or string splitting, and the format could change with every Gemini call.

Structured output solves this. You give Gemini a Pydantic model's JSON schema, and it returns valid JSON that maps directly to typed Python fields. No parsing guesswork.

In models.py, add the following model below ClassificationResult:

class PaymentDiagnosis(BaseModel):
    root_cause: str = Field(description="Primary reason the payment failed")
    failure_permanence: str = Field(
        description="Whether this failure is permanent, transient, or ambiguous"
    )
    recommended_actions: list[str] = Field(
        description="Ordered list of recommended recovery actions"
    )
    recovery_probability: float = Field(
        description="Estimated probability of successful recovery, 0.0 to 1.0"
    )
    explanation: str = Field(
        description="Brief explanation of the diagnosis reasoning"
    )

Why do the fields have descriptions?

The Field(description=...) values become part of the JSON schema that Gemini receives. They act as instructions telling the model what each field should contain. The more precise these descriptions are, the more consistent Gemini's structured responses become.

Save models.py.

Now update ai_engine.py to use this schema when calling Gemini.

In ai_engine.py, find the import lines at the top of the file:

import os
from google import genai
from models import PaymentEntity

Replace those three lines with the following four lines, which add the types import and PaymentDiagnosis:

import os
from google import genai
from google.genai import types
from models import PaymentEntity, PaymentDiagnosis

Below the get_client() function, add the first part of the new diagnose_payment function. This builds the prompt with all available payment context:

def diagnose_payment(
    client: genai.Client, payment: PaymentEntity
) -> PaymentDiagnosis:
    # Build detailed prompt with all available payment context
    prompt = f"""Analyze this failed Razorpay payment and provide a structured diagnosis.

Payment Details:
- Payment ID: {payment.id}
- Amount: {payment.amount_inr} INR
- Method: {payment.method}
- Error Code: {payment.error_code}
- Error Description: {payment.error_description}
- Error Reason: {payment.error_reason}
- Error Source: {payment.error_source}
- Error Step: {payment.error_step}
- International: {payment.international}
- Bank: {payment.bank}
- VPA: {payment.vpa}
- Wallet: {payment.wallet}

Determine the root cause, whether this failure is permanent/transient/ambiguous,
what recovery actions are appropriate, and the probability of successful recovery (0.0 to 1.0).
"""

Directly below the prompt (still inside the diagnose_payment function), add the Gemini call with structured output configuration:

    # Force Gemini to return JSON matching the PaymentDiagnosis schema
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            response_json_schema=PaymentDiagnosis.model_json_schema(),
            temperature=0.2,
        ),
    )
    # Parse the JSON string into a typed Pydantic object
    return PaymentDiagnosis.model_validate_json(response.text)

What makes this different from the plain text call?

Three additions force Gemini to return structured data instead of prose. response_mime_type="application/json" tells Gemini to output valid JSON. response_json_schema=PaymentDiagnosis.model_json_schema() passes the exact field names, types, and descriptions from your Pydantic model.

On the parsing side, PaymentDiagnosis.model_validate_json(response.text) converts the JSON string into a typed Python object. If Gemini returns unexpected fields or wrong types, Pydantic raises a validation error instead of silently passing bad data.

Save ai_engine.py.
Verify the new function imports correctly by running:

python -c "from ai_engine import diagnose_payment; print('Import OK')"

You should see Import OK printed with no errors.

Seeing an ImportError?

Check that models.py has PaymentDiagnosis defined and saved.
Check that the import line in ai_engine.py reads from models import PaymentEntity, PaymentDiagnosis with both names included.
Verify from google.genai import types is present. This import provides the GenerateContentConfig class used in the structured output call.

Still stuck? Help me fix my ai_engine.py imports for structured Gemini output.

Wire both paths into the pipeline

The rules engine and the structured AI diagnosis are ready. Time to connect them into a single pipeline that classifies every payment, then sends only the ambiguous ones to Gemini.

In recover.py, find the imports at the top of the file:

import json
import os
from google import genai
from models import WebhookEvent
from ai_engine import get_client, diagnose_payment_plain_text
from rules import classify_payment

Replace all of those import lines with the following updated imports:

import json
import os
from google import genai
from models import WebhookEvent, ClassificationResult
from rules import classify_payment
from ai_engine import get_client, diagnose_payment

Now replace the entire main() function (from def main(): through the last print statement) with this new version that runs the full classification and diagnosis pipeline:

def main():
    client = get_client()
    client.models.generate_content(
        model="gemini-2.5-flash",
        contents="Respond with just the word: connected",
    )
    print("Gemini connected: gemini-2.5-flash")

    events = load_events("sample_events.json")
    payments = [e.get_payment() for e in events]

    # Classify every payment with the rules engine
    classifications = {}
    for p in payments:
        result = classify_payment(p)
        classifications[p.id] = result
        source = "DETERMINISTIC" if not result.needs_ai else "NEEDS_AI"
        print(f"{p.id} | {source} | {result.category.value}")

Directly below the classification loop (still inside main()), add the AI diagnosis section that processes only the ambiguous payments:

    # Route ambiguous payments to Gemini for structured diagnosis
    needs_ai = [p for p in payments if classifications[p.id].needs_ai]
    for p in needs_ai:
        diagnosis = diagnose_payment(client, p)
        classifications[p.id].recovery_probability = diagnosis.recovery_probability
        print(f"\nAI Diagnosis for {p.id}:")
        print(f"  root_cause: {diagnosis.root_cause}")
        print(f"  failure_permanence: {diagnosis.failure_permanence}")
        print(f"  recommended_actions: {diagnosis.recommended_actions}")
        print(f"  recovery_probability: {diagnosis.recovery_probability}")
        print(f"  explanation: {diagnosis.explanation}")

What does the pipeline do now?

The pipeline runs in two phases. First, every payment goes through classify_payment which is instant and free. Payments with clear patterns (fraud, insufficient funds, timeouts) get classified without any API call.

Second, only payments where needs_ai=True get sent to Gemini. Each returns a typed PaymentDiagnosis object with real fields you can access in code: diagnosis.recovery_probability, diagnosis.root_cause, and so on.

Save recover.py.

Before you run this, think about what you expect to see. How many of the 12 payments should the rules engine handle on its own, and how many should need Gemini?

Run the updated pipeline:

python recover.py

You should see each of the 12 payments classified as DETERMINISTIC or NEEDS_AI with their category. The majority are handled by rules alone. For the few NEEDS_AI payments, you should see structured PaymentDiagnosis output with typed root_cause, failure_permanence, recommended_actions, and recovery_probability fields. No parsing errors.

Why deterministic rules before AI?

Your rules engine handles the majority of failures instantly, without using any API quota. Gemini only processes the genuinely ambiguous cases. This pattern is common in production systems. It keeps costs low, latency fast, and reserves AI for decisions that actually benefit from reasoning.

Seeing a Gemini API error?

If you see a rate limit error, wait 60 seconds and try again. The free tier allows 10 requests per minute.
If model_validate_json raises a validation error, check that PaymentDiagnosis in models.py has exactly five fields: root_cause, failure_permanence, recommended_actions, recovery_probability, and explanation.
Make sure ai_engine.py uses response_mime_type="application/json" and response_json_schema=PaymentDiagnosis.model_json_schema() inside the GenerateContentConfig.

Still stuck? Help me debug my structured Gemini output pipeline.

✔️ Awesome, I've got everything!

Make sure you have saved all four files: models.py, rules.py, ai_engine.py, and recover.py.

ⓧ I'd like to double check the full code

Your models.py should have, in order from top to bottom: PaymentEntity, PaymentPayload, WebhookEvent, FailureCategory, ClassificationResult, and PaymentDiagnosis.

Your rules.py should have one import line and the classify_payment function with pattern checks in this order: fraud, insufficient_funds, international_block, card_expired, network_timeout, server_error, session_expired, bank_decline, then the UNKNOWN fallback return.

Your ai_engine.py should have four import lines (os, genai, types, PaymentEntity and PaymentDiagnosis), then get_client(), diagnose_payment(), and diagnose_payment_plain_text().

Your recover.py imports should include ClassificationResult from models, classify_payment from rules, and diagnose_payment from ai_engine. The main() function should classify all payments first, then loop over needs_ai payments and print structured diagnosis fields.

Every payment now has a machine-readable classification. Deterministic rules handle the majority instantly, and Gemini returns typed diagnosis objects for the rest. Next up, you will add policy guardrails to block unsafe actions, expected-value scoring to pick the best recovery strategy, and a simulator to measure results.

Build the Policy Engine, EV Scoring, and Simulation

You now have a diagnosis for every failed payment. Some came from deterministic rules, others from Gemini's structured output. But a diagnosis alone does not tell you what action to take.

Without guardrails, your pipeline could retry a fraud-flagged payment or spam a customer at 3 AM. Without a way to compare recovery options mathematically, you are guessing. In this step, you will build a policy engine that blocks unsafe actions, an expected-value scorer that picks the mathematically optimal recovery, a seeded simulator, and a tamper-evident audit log.

In this step, get ready to:

Build a policy engine that blocks fraud retries, enforces retry limits, and respects quiet hours.
Build an expected-value engine that scores recovery actions and picks the highest-value option.
Build a seeded simulator and SHA-256 hash-chained audit log, then wire the full pipeline end to end.

Build the policy engine and EV scorer

Before you can write the policy engine and EV scorer, they need new model types to work with. You will add a RecoveryAction enum for the four possible actions and an EVScore model to hold scored results.

In models.py, scroll to the bottom of the file (below PaymentDiagnosis) and add these two new classes:

class RecoveryAction(str, enum.Enum):
    RETRY = "retry"
    PAYMENT_LINK = "payment_link"
    REMINDER = "reminder"
    ABANDON = "abandon"

class EVScore(BaseModel):
    action: RecoveryAction
    expected_value: float
    probability: float
    amount_inr: float
    cost: float
    blocked: bool = False
    block_reason: Optional[str] = None

What do these models represent?

RecoveryAction defines the four strategies the system can attempt: retry the payment, send a payment link, send a reminder, or abandon.

EVScore holds the expected-value calculation for one action. The blocked and block_reason fields let the policy engine mark an action as forbidden before the EV scorer picks the best one.

Save models.py.

Now you will build the policy engine. This is the guardrail layer that prevents dangerous recovery actions before they happen.

Open policy.py (currently empty) and add the imports, class definition, and first three guardrail checks:

from models import (
    FailureCategory,
    RecoveryAction,
)

class PolicyEngine:
    def __init__(self):
        self.max_retries = 3
        self.quiet_hour_start = 22
        self.quiet_hour_end = 8
        self.seen_events: set[str] = set()
        self.retry_counts: dict[str, int] = {}

    def check_duplicate(self, payment_id: str) -> tuple[bool, str]:
        if payment_id in self.seen_events:
            return True, f"Duplicate event for {payment_id}"
        self.seen_events.add(payment_id)
        return False, ""

    def check_fraud_block(self, category: FailureCategory) -> tuple[bool, str]:
        if category == FailureCategory.FRAUD:
            return True, "Fraud-flagged payments cannot be retried"
        return False, ""

    def check_permanent_block(self, category: FailureCategory) -> tuple[bool, str]:
        if category == FailureCategory.INTERNATIONAL_BLOCK:
            return True, "International payments blocked by business config"
        return False, ""

What do these checks do?

check_duplicate tracks every payment ID it has seen. If the same webhook fires twice, the second attempt is blocked.
check_fraud_block prevents any recovery attempt on fraud-flagged payments.
check_permanent_block catches failures that cannot be resolved, like international payments blocked by business configuration.

Save policy.py.
Verify the imports work by running this command:

python -c "from policy import PolicyEngine; p = PolicyEngine(); print('Policy engine loaded')"

You should see Policy engine loaded printed to the terminal.

Seeing an import error?

Make sure you saved both models.py (with the new RecoveryAction enum) and policy.py.
Check that RecoveryAction is defined at the module level in models.py, not nested inside another class.
Still stuck? Help me debug my policy engine import error.

Back in policy.py, add the retry limit, retry counter, and quiet hours checks below check_permanent_block:

    def check_retry_limit(self, payment_id: str) -> tuple[bool, str]:
        count = self.retry_counts.get(payment_id, 0)
        if count >= self.max_retries:
            return True, f"Retry limit ({self.max_retries}) exceeded"
        return False, ""

    def record_retry(self, payment_id: str):
        self.retry_counts[payment_id] = self.retry_counts.get(payment_id, 0) + 1

    def check_quiet_hours(
        self, hour: int, action: RecoveryAction
    ) -> tuple[bool, str]:
        if action in (RecoveryAction.PAYMENT_LINK, RecoveryAction.REMINDER):
            if hour >= self.quiet_hour_start or hour < self.quiet_hour_end:
                return True, f"Cannot send {action.value} during quiet hours (22:00-08:00)"
        return False, ""

What do these checks do?

check_retry_limit caps retries at 3 per payment. After that, the system must try a different strategy.
check_quiet_hours blocks payment links and reminders between 22:00 and 08:00 to avoid disturbing customers overnight.

Now add the evaluate method that runs all checks in sequence. Add this below check_quiet_hours:

    def evaluate(
        self,
        payment_id: str,
        category: FailureCategory,
        action: RecoveryAction,
        current_hour: int = 14,
    ) -> tuple[bool, str]:
        is_dup, reason = self.check_duplicate(payment_id)
        if is_dup:
            return True, reason

        blocked, reason = self.check_fraud_block(category)
        if blocked:
            return True, reason

        blocked, reason = self.check_permanent_block(category)
        if blocked:
            return True, reason

        if action == RecoveryAction.RETRY:
            blocked, reason = self.check_retry_limit(payment_id)
            if blocked:
                return True, reason

        blocked, reason = self.check_quiet_hours(current_hour, action)
        if blocked:
            return True, reason

        return False, ""

Why does evaluate run checks in this order?

The checks go from cheapest to most expensive. Duplicate detection is a set lookup. Fraud and permanent blocks are simple enum comparisons. Retry limits and quiet hours only run if the earlier checks pass. This short-circuit pattern avoids unnecessary work.

Save policy.py and verify the full class loads by running:

python -c "from policy import PolicyEngine; print('All policy checks loaded')"

You should see All policy checks loaded with no errors.

✔️ Awesome, I've got everything!

Great. Make sure you have saved policy.py before moving on.

ⓧ I'd like to double check the full code

Since policy.py spans multiple code blocks, scroll up and confirm you have all of these in order:

Imports and class definition with __init__ (max_retries, quiet hours, seen_events, retry_counts).
check_duplicate, check_fraud_block, and check_permanent_block methods.
check_retry_limit, record_retry, and check_quiet_hours methods.
evaluate method that chains all checks together.

Next up is the expected-value engine. This module calculates a score for each recovery action so the pipeline can pick the mathematically best option.

Open ev_engine.py (currently empty) and add the imports and cost constants:

from models import RecoveryAction, EVScore, FailureCategory

# Intervention cost (INR) per recovery action
ACTION_COSTS = {
    RecoveryAction.RETRY: 2.0,
    RecoveryAction.PAYMENT_LINK: 5.0,
    RecoveryAction.REMINDER: 1.0,
    RecoveryAction.ABANDON: 0.0,
}

# Multiplier applied to base recovery probability per action
PROBABILITY_ADJUSTMENTS = {
    RecoveryAction.RETRY: 1.0,
    RecoveryAction.PAYMENT_LINK: 0.8,
    RecoveryAction.REMINDER: 0.5,
    RecoveryAction.ABANDON: 0.0,
}

What is expected value?

Expected value (EV) is the formula: recovery_probability x amount_inr - intervention_cost. A retry costs 2 INR but has the highest probability multiplier (1.0). A payment link costs 5 INR but only reaches 80% of the base probability. Abandon costs nothing but recovers nothing.

Below the constants, add the score_actions and select_best_action functions:

def score_actions(
    amount_inr: float,
    recovery_probability: float,
    category: FailureCategory,
) -> list[EVScore]:
    scores = []
    for action in RecoveryAction:
        adj_probability = recovery_probability * PROBABILITY_ADJUSTMENTS[action]
        cost = ACTION_COSTS[action]
        # EV = probability of recovery * payment amount - cost to intervene
        ev = adj_probability * amount_inr - cost
        scores.append(
            EVScore(
                action=action,
                expected_value=round(ev, 2),
                probability=round(adj_probability, 4),
                amount_inr=amount_inr,
                cost=cost,
            )
        )
    scores.sort(key=lambda s: s.expected_value, reverse=True)
    return scores

def select_best_action(
    scores: list[EVScore],
) -> tuple[RecoveryAction, EVScore]:
    # Pick the first non-blocked action (scores are sorted by EV)
    for score in scores:
        if not score.blocked:
            return score.action, score
    return RecoveryAction.ABANDON, scores[-1]

What does this code do?

score_actions loops through all four recovery actions, calculates each one's expected value, and returns them sorted from highest to lowest EV.
select_best_action walks the sorted list and picks the first action that hasn't been blocked by the policy engine. If everything is blocked, it falls back to ABANDON.

Save ev_engine.py and verify it loads by running:

python -c "from ev_engine import score_actions, select_best_action; print('EV engine ready')"

You should see EV engine ready with no errors.

Seeing a NameError or ImportError?

Check that EVScore is defined in models.py (you added it earlier in this step).
Make sure models.py is saved with the RecoveryAction enum. The EV engine imports it directly.
Still stuck? Help me debug my EV engine import error.

✔️ Awesome, I've got everything!

Great. Make sure you have saved ev_engine.py before moving on.

ⓧ I'd like to double check the full code

Scroll up and confirm ev_engine.py has these sections in order:

Imports from models.
ACTION_COSTS dictionary (retry 2.0, payment_link 5.0, reminder 1.0, abandon 0.0).
PROBABILITY_ADJUSTMENTS dictionary (retry 1.0, payment_link 0.8, reminder 0.5, abandon 0.0).
score_actions function that calculates EV and returns sorted scores.
select_best_action function that picks the first non-blocked action.

Build the seeded simulator and audit log

You have guardrails and scoring, but two pieces are still missing. You need a simulator to test whether a recovery action succeeds, and an audit log to record every decision with tamper detection.

First, add the remaining model types that the simulator and audit log depend on.

In models.py, scroll to the bottom (below EVScore) and add these two classes:

class DecisionRecord(BaseModel):
    payment_id: str
    amount_inr: float
    method: str
    error_description: Optional[str] = None
    classification_source: str
    category: str
    recovery_probability: float
    policy_blocked: bool = False
    policy_block_reason: Optional[str] = None
    ev_scores: list[EVScore] = []
    selected_action: Optional[RecoveryAction] = None
    simulated_outcome: Optional[str] = None
    recovered_amount: float = 0.0
    intervention_cost: float = 0.0

class AuditEntry(BaseModel):
    index: int
    payment_id: str
    action: str
    detail: str
    timestamp: float
    previous_hash: str
    current_hash: str = ""

What do these models represent?

DecisionRecord captures the full story of one payment: its classification, EV scores, selected action, and outcome. The dashboard will use this to render per-payment timelines.
AuditEntry stores one audit log record with previous_hash and current_hash fields for SHA-256 hash chaining.

Save models.py.

Open simulator.py (currently empty) and add the recovery simulator:

import random
from models import RecoveryAction

def simulate_recovery(
    payment_id: str,
    action: RecoveryAction,
    recovery_probability: float,
) -> tuple[bool, str]:
    if action == RecoveryAction.ABANDON:
        return False, "No recovery attempted"
    # Derive a deterministic seed from the payment ID
    seed = int.from_bytes(payment_id.encode()[:8], "big")
    rng = random.Random(seed)
    roll = rng.random()
    if roll < recovery_probability:
        return True, f"Recovery successful (roll={roll:.3f} < prob={recovery_probability:.3f})"
    else:
        return False, f"Recovery failed (roll={roll:.3f} >= prob={recovery_probability:.3f})"

Why use a seeded random number?

Deriving the seed from the payment ID means the same payment always produces the same outcome. This makes the simulation reproducible. Every time you run the pipeline, you get identical results, which is critical for debugging and for the dashboard to show consistent data.

Save simulator.py and verify it loads by running:

python -c "from simulator import simulate_recovery; print('Simulator ready')"

You should see Simulator ready with no errors.

✔️ Awesome, I've got everything!

Great. Make sure you have saved simulator.py before moving on.

ⓧ I'd like to double check the full code

import random
from models import RecoveryAction

def simulate_recovery(
    payment_id: str,
    action: RecoveryAction,
    recovery_probability: float,
) -> tuple[bool, str]:
    if action == RecoveryAction.ABANDON:
        return False, "No recovery attempted"
    # Derive a deterministic seed from the payment ID
    seed = int.from_bytes(payment_id.encode()[:8], "big")
    rng = random.Random(seed)
    roll = rng.random()
    if roll < recovery_probability:
        return True, f"Recovery successful (roll={roll:.3f} < prob={recovery_probability:.3f})"
    else:
        return False, f"Recovery failed (roll={roll:.3f} >= prob={recovery_probability:.3f})"

Now for the audit log. Every decision the pipeline makes will be recorded as a hash-chained entry. If anyone tampers with a record, the chain breaks and verification fails.

Open audit.py (currently empty) and add the class structure:

import hashlib
import json
import time
from models import AuditEntry

class AuditLog:
    def __init__(self):
        self.entries: list[AuditEntry] = []

Below the __init__, add the append method that creates hash-chained entries:

    def append(self, payment_id: str, action: str, detail: str) -> AuditEntry:
        # Chain to the previous entry's hash (or zeros for the first entry)
        previous_hash = (
            self.entries[-1].current_hash if self.entries else "0" * 64
        )
        entry = AuditEntry(
            index=len(self.entries),
            payment_id=payment_id,
            action=action,
            detail=detail,
            timestamp=time.time(),
            previous_hash=previous_hash,
        )
        # Hash the entry's contents to create a tamper-evident fingerprint
        hash_input = json.dumps(
            {
                "index": entry.index,
                "payment_id": entry.payment_id,
                "action": entry.action,
                "detail": entry.detail,
                "timestamp": entry.timestamp,
                "previous_hash": entry.previous_hash,
            },
            sort_keys=True,
        )
        entry.current_hash = hashlib.sha256(hash_input.encode()).hexdigest()
        self.entries.append(entry)
        return entry

How does hash chaining work?

Each entry includes the hash of the previous entry in its own hash calculation. This creates a chain: if someone modifies entry #3, its hash changes, which means entry #4's previous_hash no longer matches. The entire chain from that point forward becomes invalid.

The first entry uses a string of 64 zeros as its previous_hash since there is nothing before it.

Below append, add the verify_chain method that checks every link:

    def verify_chain(self) -> tuple[bool, str]:
        if not self.entries:
            return True, "Empty log"
        if self.entries[0].previous_hash != "0" * 64:
            return False, "Genesis block has wrong previous hash"
        for i, entry in enumerate(self.entries):
            hash_input = json.dumps(
                {
                    "index": entry.index,
                    "payment_id": entry.payment_id,
                    "action": entry.action,
                    "detail": entry.detail,
                    "timestamp": entry.timestamp,
                    "previous_hash": entry.previous_hash,
                },
                sort_keys=True,
            )
            computed = hashlib.sha256(hash_input.encode()).hexdigest()
            if computed != entry.current_hash:
                return False, f"Hash mismatch at entry {i}"
            if i > 0 and entry.previous_hash != self.entries[i - 1].current_hash:
                return False, f"Chain break at entry {i}"
        return True, f"All {len(self.entries)} entries verified"

What does verify_chain check?

The genesis entry's previous_hash must be all zeros.
Every entry's stored hash must match a freshly computed hash of its contents.
Every entry's previous_hash must match the hash of the entry before it.

Save audit.py and test the hash chain by running:

python -c "from audit import AuditLog; log = AuditLog(); log.append('test', 'init', 'test entry'); v, m = log.verify_chain(); print(f'{v}: {m}')"

You should see True: All 1 entries verified.

Seeing an error from audit.py?

Make sure AuditEntry is defined in models.py and that you saved the file.
Check that both append and verify_chain are indented inside the AuditLog class (one level of indentation from class AuditLog:).
Still stuck? Help me debug my audit log.

✔️ Awesome, I've got everything!

Great. Make sure you have saved audit.py before moving on.

ⓧ I'd like to double check the full code

Scroll up and confirm audit.py has these sections in order:

Imports (hashlib, json, time, AuditEntry).
AuditLog class with __init__ setting entries to an empty list.
append method that creates an AuditEntry, hashes its contents with SHA-256, and chains it to the previous hash.
verify_chain method that recomputes every hash and checks the chain links.

Wire the full pipeline

Every module is built and tested. Now you will replace the current recover.py with the full pipeline that wires them all together into a single run_pipeline() function.

Clear the entire contents of recover.py and replace it with the updated imports and load_events function:

import json
import os
from google import genai
from models import (
    WebhookEvent,
    DecisionRecord,
    RecoveryAction,
    FailureCategory,
)
from rules import classify_payment
from ai_engine import get_client, diagnose_payment
from policy import PolicyEngine
from ev_engine import score_actions, select_best_action, ACTION_COSTS
from simulator import simulate_recovery
from audit import AuditLog

def load_events(filepath: str) -> list[WebhookEvent]:
    with open(filepath) as f:
        raw = json.load(f)
    return [WebhookEvent(**event) for event in raw]

What changed in the imports?

The imports now pull in every module you built in this step: PolicyEngine, score_actions, simulate_recovery, and AuditLog. They also import DecisionRecord and ACTION_COSTS for the pipeline logic.

Below load_events, add the first part of run_pipeline that sets up the pipeline and classifies all payments:

def run_pipeline() -> tuple[list[DecisionRecord], AuditLog, dict]:
    client = get_client()

    client.models.generate_content(
        model="gemini-2.5-flash",
        contents="Respond with just the word: connected",
    )

    events = load_events("sample_events.json")
    payments = [e.get_payment() for e in events]

    policy = PolicyEngine()
    audit_log = AuditLog()
    decisions: list[DecisionRecord] = []
    classifications = {}

    for p in payments:
        result = classify_payment(p)
        classifications[p.id] = result

    needs_ai = [p for p in payments if classifications[p.id].needs_ai]
    for p in needs_ai:
        diagnosis = diagnose_payment(client, p)
        classifications[p.id].recovery_probability = diagnosis.recovery_probability

What does this section do?

Initializes the Gemini client and verifies the connection.
Loads all 12 webhook events and extracts payment entities.
Creates fresh PolicyEngine, AuditLog, and classifications instances.
Classifies every payment with the rules engine, then sends ambiguous cases to Gemini for structured diagnosis.

Save recover.py and verify the imports are correct by running:

python -c "import recover; print('Imports OK')"

You should see Imports OK. If you see an error, double-check the import lines match exactly.

Continue run_pipeline by adding the main processing loop that creates decision records, runs EV scoring, and applies policy checks. Add this directly below the diagnose_payment loop:

    for p in payments:
        cls = classifications[p.id]
        source = "gemini" if cls.needs_ai else "rules"
        prob = cls.recovery_probability

        record = DecisionRecord(
            payment_id=p.id,
            amount_inr=p.amount_inr,
            method=p.method,
            error_description=p.error_description,
            classification_source=source,
            category=cls.category.value,
            recovery_probability=prob,
        )

        audit_log.append(p.id, "ingested", f"Parsed {p.method} payment {p.amount_inr} INR")
        audit_log.append(p.id, "classified", f"{source}: {cls.category.value} (prob={prob})")

        ev_scores = score_actions(p.amount_inr, prob, cls.category)

        for score in ev_scores:
            blocked, reason = policy.evaluate(p.id, cls.category, score.action)
            if blocked:
                score.blocked = True
                score.block_reason = reason

        record.ev_scores = ev_scores
        best_action, best_score = select_best_action(ev_scores)

What does this loop do?

For each payment, the loop creates a DecisionRecord, logs the ingestion and classification to the audit log, scores all four recovery actions, and runs each action through the policy engine to mark blocked ones. Finally, it picks the best non-blocked action.

Still inside the same for p in payments: loop, add the policy logging, simulation, and outcome recording. Paste this directly below best_action, best_score = select_best_action(ev_scores):

        any_blocked = any(s.blocked for s in ev_scores)
        if any_blocked:
            blocked_actions = [s for s in ev_scores if s.blocked]
            for ba in blocked_actions:
                record.policy_blocked = True
                record.policy_block_reason = ba.block_reason
                audit_log.append(p.id, "policy_blocked", f"{ba.action.value}: {ba.block_reason}")

        record.selected_action = best_action
        audit_log.append(p.id, "action_selected", f"{best_action.value} (EV={best_score.expected_value:.2f})")

        if best_action == RecoveryAction.RETRY:
            policy.record_retry(p.id)

        success, outcome_detail = simulate_recovery(p.id, best_action, prob)
        if success:
            record.simulated_outcome = "success"
            record.recovered_amount = p.amount_inr
            record.intervention_cost = ACTION_COSTS[best_action]
        else:
            record.simulated_outcome = "failed"
            record.intervention_cost = (
                ACTION_COSTS[best_action]
                if best_action != RecoveryAction.ABANDON
                else 0.0
            )

        audit_log.append(p.id, "outcome", outcome_detail)
        decisions.append(record)

What does this section do?

Logs any policy-blocked actions to the audit trail.
Records the selected action and its EV score.
Runs simulate_recovery to determine if the action succeeds, then records the outcome and cost in the decision record.

After the for loop (back at the run_pipeline indentation level), add the stats summary, return statement, and the new main() function:

    stats = {
        "total": len(decisions),
        "deterministic": sum(1 for d in decisions if d.classification_source == "rules"),
        "ai_classified": sum(1 for d in decisions if d.classification_source == "gemini"),
    }

    return decisions, audit_log, stats

def main():
    decisions, audit_log, stats = run_pipeline()

    total_at_risk = sum(d.amount_inr for d in decisions)
    successful = sum(1 for d in decisions if d.simulated_outcome == "success")
    gross_recovered = sum(d.recovered_amount for d in decisions)
    total_cost = sum(d.intervention_cost for d in decisions)
    net_recovered = gross_recovered - total_cost

    print(f"Gemini connected: gemini-2.5-flash")
    print(f"Processed {stats['total']} payments | At-risk: {total_at_risk:,.2f} INR")
    print(f"Deterministic: {stats['deterministic']} | AI: {stats['ai_classified']}")
    print(f"Recovered: {successful} payments | Net: {net_recovered:,.2f} INR")

    is_valid, msg = audit_log.verify_chain()
    print(f"Audit log: {'INTACT' if is_valid else 'TAMPERED'} - {msg}")


if __name__ == "__main__":
    main()

Why is run_pipeline a separate function?

Extracting the pipeline into run_pipeline() means both the CLI (main()) and the Streamlit dashboard can call it. The function returns decisions, audit_log, and stats so the caller can render them however it wants.

Save recover.py.

Before you run the full pipeline, think about what you expect to see. The pipeline will process all 12 payments through classification, policy checks, EV scoring, simulation, and audit logging. Do you think every payment will be recovered?

Run the full pipeline by running:

python recover.py

You should see output similar to this:

Gemini connected: gemini-2.5-flash
Processed 12 payments | At-risk: 47,500.00 INR
Deterministic: 10 | AI: 2
Recovered: X payments | Net: X,XXX.XX INR
Audit log: INTACT - All XX entries verified

What do these numbers tell you?

10 out of 12 payments were classified by deterministic rules without any AI call. Only 2 ambiguous cases went to Gemini.
The exact recovery count depends on Gemini's diagnosis probabilities, but the seeded simulator ensures consistent results between runs.
Audit log: INTACT confirms the hash chain is unbroken. Every decision was recorded and no entries were tampered with.

Seeing an error when running the pipeline?

If you see IndentationError, check that the for p in payments: loop body and the stats dictionary are at the correct indentation level inside run_pipeline().
If you see NameError, make sure all six files are saved: models.py, policy.py, ev_engine.py, simulator.py, audit.py, and recover.py.
If the GEMINI_API_KEY error appears, make sure your environment variable is still set in this terminal session.
Still stuck? Help me debug my RecoverAI pipeline.

✔️ Awesome, I've got everything!

Great. Make sure you have saved all files before moving on.

ⓧ I'd like to double check the full code

Since recover.py spans multiple code blocks, scroll up and confirm you have all of these in order:

Imports from json, os, genai, models, rules, ai_engine, policy, ev_engine, simulator, and audit.
load_events function.
run_pipeline function: client setup, event loading, classify/diagnose loops.
Main for loop: DecisionRecord creation, audit logging, EV scoring, policy evaluation, best action selection.
Policy block logging, simulation, outcome recording, decisions.append.
stats dictionary, return statement.
main function with summary prints and audit log verification.
if __name__ == "__main__": main() at the bottom.

Your pipeline now processes all 12 payments end to end: classify, diagnose, check policy, score recovery options, simulate, and log every decision in a tamper-evident audit chain. Next up, you will build a Streamlit dashboard that turns all of this into a visual report you can open in your browser and show off.

Build the Streamlit Dashboard

Your recovery pipeline works, but its output is a wall of terminal text. Nobody can glance at a terminal dump and immediately grasp how well the system recovers revenue.

In this step, you will build a Streamlit dashboard that turns your pipeline results into KPI metric cards, interactive charts, expandable payment timelines, and a verified audit log. Streamlit handles all the HTML, CSS, and JavaScript for you so your effort stays on the data, not on frontend boilerplate.

In this step, get ready to:

Display KPI metric cards showing at-risk revenue, recoveries, and net recovered INR.
Visualize recovery outcomes with a bar chart and expandable per-payment decision timelines.
Add a tabbed audit log with SHA-256 hash chain verification status.

Build the dashboard layout and KPI metrics

Streamlit apps are plain Python scripts that render widgets top-to-bottom. You write Python, and Streamlit turns it into a live web page at localhost:8501.

In dashboard.py, add the imports, page configuration, and pipeline runner by copying this code:

import streamlit as st
import pandas as pd
from models import RecoveryAction
from recover import run_pipeline

# Configure the page for a wide dashboard layout
st.set_page_config(
    page_title="RecoverAI Dashboard",
    layout="wide",
)

st.title("RecoverAI: Payment Recovery Dashboard")

# Run the full pipeline and show progress
with st.status("Running recovery pipeline...") as status:
    decisions, audit_log, stats = run_pipeline()
    status.update(label="Pipeline complete", state="complete")

What does this code do?

st.set_page_config(layout="wide") tells Streamlit to use the full browser width instead of a narrow center column.

st.status creates a mutable status container that shows "Running recovery pipeline..." while run_pipeline() processes all 12 payments, then switches to "Pipeline complete" when done.

Why Streamlit instead of Flask or FastAPI?

Streamlit is purpose-built for data visualization dashboards. It requires zero HTML, CSS, or JavaScript. You write ordinary Python and Streamlit renders interactive widgets, charts, and tables automatically.

Flask or FastAPI would require writing templates, routes, and frontend code. Your effort stays on the AI pipeline, not on frontend boilerplate.

Save dashboard.py.
Launch the dashboard by running this command in your terminal:

streamlit run dashboard.py

Your browser opens automatically to localhost:8501. You should see the title "RecoverAI: Payment Recovery Dashboard" and a status indicator that flips from "Running recovery pipeline..." to "Pipeline complete".

Dashboard not loading?

Make sure your GEMINI_API_KEY environment variable is still set in the same terminal session where you ran the command.
Check that you are running the command from inside the recoverai/ folder so Streamlit can find your project files.
Help me debug my Streamlit dashboard not loading.

Keep Streamlit running in this terminal. From now on, every time you save dashboard.py, Streamlit auto-detects the change and offers to re-run the app in your browser.

The pipeline is running, but the page is mostly blank. Time to compute KPI values and display them as metric cards.

In dashboard.py, add the following code below the status.update block. This computes summary statistics, creates three tabs, and fills the Overview tab with KPI metric cards:

# Compute KPI summary values from pipeline results
total_at_risk = sum(d.amount_inr for d in decisions)
successful = sum(1 for d in decisions if d.simulated_outcome == "success")
attempted = sum(
    1 for d in decisions
    if d.selected_action and d.selected_action != RecoveryAction.ABANDON
)
gross_recovered = sum(d.recovered_amount for d in decisions)
total_cost = sum(d.intervention_cost for d in decisions)
net_recovered = gross_recovered - total_cost
recovery_rate = (successful / stats["total"] * 100) if stats["total"] > 0 else 0

# Create three dashboard tabs
tab_overview, tab_payments, tab_audit = st.tabs(
    ["Overview", "Payment Details", "Audit Log"]
)

with tab_overview:
    col1, col2, col3, col4 = st.columns(4)
    col1.metric("Total Failed Payments", stats["total"], border=True)
    col2.metric("At-Risk Revenue", f"{total_at_risk:,.0f} INR", border=True)
    col3.metric(
        "Successful Recoveries",
        successful,
        f"{recovery_rate:.0f}% rate",
        border=True,
    )
    col4.metric(
        "Net Recovered",
        f"{net_recovered:,.0f} INR",
        f"-{total_cost:,.0f} costs",
        border=True,
    )

What does this code do?

The KPI computation lines aggregate totals from the decisions list. net_recovered subtracts intervention costs from gross recovery.
st.tabs creates three clickable tabs. Everything inside a with tab_overview: block renders only when the Overview tab is selected.
st.metric with border=True creates dashboard-style cards with a visible border. The third argument is a delta indicator (like "+33% rate").

Save dashboard.py and switch back to your browser.
Click **Rerun** in the Streamlit notification bar (or press **R** on the page) to reload.

You should see four metric cards across the top of the Overview tab showing Total Failed Payments, At-Risk Revenue, Successful Recoveries (with a percentage delta), and Net Recovered INR (with a cost delta).

Metrics not showing up?

Make sure the KPI computation lines and the with tab_overview: block are at the top level of the file (not indented inside the st.status block).
Check that all four col1.metric through col4.metric calls are indented inside the with tab_overview: block.
Why aren't my Streamlit metric cards rendering?

Add charts and payment details

The Overview tab has raw numbers, but a chart makes recovery patterns visible at a glance. You also need expandable cards so someone can drill into any individual payment's decision timeline.

In dashboard.py, find the last col4.metric(...) call inside with tab_overview:. Add the following code below it, keeping the same indentation level (still inside with tab_overview:):

    # Show how many payments were classified by rules vs AI
    st.subheader("Classification Breakdown")
    cls_col1, cls_col2 = st.columns(2)
    cls_col1.metric("Deterministic (Rules)", stats["deterministic"], border=True)
    cls_col2.metric("AI (Gemini)", stats["ai_classified"], border=True)

    # Build a bar chart comparing recovered vs unrecovered by payment method
    st.subheader("Recovery by Payment Method")
    method_data = {}
    for d in decisions:
        if d.method not in method_data:
            method_data[d.method] = {"recovered": 0.0, "failed": 0.0}
        if d.simulated_outcome == "success":
            method_data[d.method]["recovered"] += d.recovered_amount
        else:
            method_data[d.method]["failed"] += d.amount_inr

    chart_df = pd.DataFrame(
        [
            {"Method": method, "Recovered (INR)": vals["recovered"], "Unrecovered (INR)": vals["failed"]}
            for method, vals in method_data.items()
        ]
    )
    st.bar_chart(chart_df, x="Method", y=["Recovered (INR)", "Unrecovered (INR)"], stack=False)

What does this code do?

The classification breakdown shows two metric cards: how many payments the rules engine handled vs how many needed Gemini. This makes the efficiency of your deterministic rules visible.
The bar chart aggregates recovered and unrecovered INR by payment method (card, UPI, netbanking, wallet). stack=False places bars side by side instead of stacking them, making comparison easier.

Save dashboard.py and rerun in your browser.

Scroll down on the Overview tab. You should see two classification metric cards (Deterministic vs AI) and a bar chart showing recovered vs unrecovered amounts broken down by payment method.

The Payment Details tab is still empty. Each payment needs an expandable card showing its full decision timeline, from the original error through EV scoring to the simulated outcome.

In dashboard.py, add the following code after the with tab_overview: block ends (at the same indentation level as with tab_overview:). This builds the basic expander structure for each payment:

with tab_payments:
    st.subheader("Per-Payment Decision Timelines")
    for d in decisions:
        # Label each expander with payment ID, amount, method, and outcome
        outcome_icon = "+" if d.simulated_outcome == "success" else "-"
        label = f"[{d.payment_id}] {d.amount_inr:,.0f} INR via {d.method} ({outcome_icon})"
        with st.expander(label):
            st.write(f"Error: {d.error_description}")
            st.write(f"Classification: {d.classification_source.upper()} -> {d.category}")
            st.write(f"Recovery Probability: {d.recovery_probability:.2f}")
            if d.policy_blocked:
                st.warning(f"Policy Blocked: {d.policy_block_reason}")

What does this code do?

st.expander creates a collapsible section for each payment. The label includes a + or - icon so you can spot successful vs failed recoveries at a glance.
st.warning renders a yellow warning banner for policy-blocked payments (like fraud-flagged transactions that the policy engine prevented from retrying).

Save dashboard.py and rerun in your browser.
Click the **Payment Details** tab. You should see 12 collapsible rows. Click any row to expand it and see the error, classification source, and recovery probability.

Each expander shows basic info, but the EV scores table and outcome are still missing. The viewer needs to see which action the system picked and why.

In dashboard.py, find the line st.warning(f"Policy Blocked: {d.policy_block_reason}") inside the expander. Add the following code below it, keeping the same indentation level (still inside the with st.expander(label): block):

            # Show EV scores as an interactive table
            if d.ev_scores:
                ev_df = pd.DataFrame(
                    [
                        {
                            "Action": s.action.value,
                            "EV (INR)": s.expected_value,
                            "Probability": s.probability,
                            "Cost (INR)": s.cost,
                            "Blocked": "Yes" if s.blocked else "No",
                        }
                        for s in d.ev_scores
                    ]
                )
                st.dataframe(ev_df, hide_index=True)
            # Display the selected action and simulated outcome
            if d.selected_action:
                st.write(f"Selected Action: **{d.selected_action.value}**")
            outcome_text = (
                f"Outcome: {d.simulated_outcome}"
                + (f" (+{d.recovered_amount:,.0f} INR)" if d.recovered_amount > 0 else "")
            )
            if d.simulated_outcome == "success":
                st.success(outcome_text)
            else:
                st.error(outcome_text)

What does this code do?

st.dataframe renders the EV scores as a sortable, interactive table. Each row shows an action (retry, payment_link, reminder, abandon), its expected value in INR, probability, cost, and whether it was blocked by the policy engine.
st.success and st.error render green or red banners for the simulated outcome. Successful recoveries also show the recovered amount in INR.

Save dashboard.py and rerun in your browser.
Click the **Payment Details** tab and expand any payment. You should now see an EV scores table, the selected action in bold, and a green or red outcome banner.

Expander content looks wrong or incomplete?

Double check that the new code block is indented at the same level as the st.write and st.warning lines above it. Python is whitespace-sensitive, so misaligned code will either error or render outside the expander.
If you see a red Streamlit error, read the traceback line number and check for missing colons or mismatched parentheses near that line.
Help me fix my Streamlit expander content.

Add the audit log tab

The Audit Log tab proves that no decision was tampered with after the fact. It displays the SHA-256 hash chain entries and runs the verification function to confirm integrity.

In dashboard.py, add the following code after the with tab_payments: block ends (at the same indentation level as with tab_payments:):

with tab_audit:
    st.subheader("SHA-256 Hash-Chained Audit Log")
    # Verify the hash chain and display status
    is_valid, msg = audit_log.verify_chain()
    if is_valid:
        st.success(f"Audit Log Verified: INTACT - {msg}")
    else:
        st.error(f"Audit Log Verification FAILED: {msg}")

    # Display audit entries as a table with truncated hashes
    audit_df = pd.DataFrame(
        [
            {
                "Index": e.index,
                "Payment ID": e.payment_id,
                "Action": e.action,
                "Detail": e.detail,
                "Hash": e.current_hash[:16] + "...",
            }
            for e in audit_log.entries
        ]
    )
    st.dataframe(audit_df, hide_index=True)

What does this code do?

audit_log.verify_chain() recomputes every SHA-256 hash in the chain and checks that each entry links to the previous one. If any record was modified, the hashes break and st.error shows a red "FAILED" banner.
The hash column truncates each 64-character SHA-256 hash to 16 characters plus "..." so the table stays readable.

Save dashboard.py and rerun in your browser.
Click the **Audit Log** tab. You should see a green "Audit Log Verified: INTACT" banner and a table of hash chain entries below it.

Audit log showing FAILED?

This would mean something modified an audit entry after it was created. If you see FAILED, re-run the dashboard (click **Rerun**). The pipeline generates a fresh log each time, so a restart should produce an INTACT result.
Help me debug my audit log verification failure.

Before you do the final check, what do you expect to see when you click through all three tabs?

Click through the **Overview**, **Payment Details**, and **Audit Log** tabs to confirm everything is working together.

The Overview tab shows four KPI metric cards, a classification breakdown, and a bar chart of recovery by payment method. The Payment Details tab has 12 expandable timelines with EV score tables and outcome banners. The Audit Log tab has a green "INTACT" verification status and a table of hash chain entries.

✔️ Awesome, I've got everything!

Great. Double check you have saved dashboard.py.

ⓧ I'd like to double check the full code

Your dashboard.py should contain the following sections in order. Scroll up and confirm each one is present:

Imports (streamlit, pandas, RecoveryAction, run_pipeline) and st.set_page_config with layout="wide".
st.title and st.status block running the pipeline.
KPI computation (total_at_risk, successful, attempted, net_recovered, recovery_rate).
st.tabs with three tabs: Overview, Payment Details, Audit Log.
with tab_overview: containing four st.metric cards, classification breakdown, and st.bar_chart.
with tab_payments: containing a for loop with st.expander per payment, each showing error info, EV scores dataframe, selected action, and outcome banner.
with tab_audit: containing verify_chain() with st.success/st.error and a st.dataframe of audit entries.

Secret mission

Customer Risk Scoring System

Your pipeline treats every payment as an isolated event. But a customer with three recent failures is a very different recovery prospect than a first-time failure. Build a cross-payment customer risk engine that tracks failure history, assigns LOW/MEDIUM/HIGH risk scores, and adjusts recovery probabilities before they reach the EV scorer. Then add a dedicated Customer Risk tab to your Streamlit dashboard so you can see exactly how repeat-offender behavior changes recovery decisions.

Clean Up Your Resources

This project runs entirely locally with Gemini API free tier calls. There are no cloud resources, databases, or paid services creating ongoing costs.

Resources you used:

The recoverai/ project directory on your Desktop containing all Python source files and sample_events.json.
Installed Python packages: google-genai, pydantic, and streamlit.
Your Google AI Studio API key used for Gemini calls.

Keep everything running

No action needed. Choose this if you are still actively building or want to extend RecoverAI further.

Your recoverai/ directory stays on your Desktop for reference or future development.
The installed pip packages (google-genai, pydantic, streamlit) remain available for other projects.
Your Google AI Studio API key remains active. The Gemini free tier incurs no charges.

Pause - I'll come back to this later

There are no persistent processes or cloud resources to pause. Everything runs on demand.

If the Streamlit server is still running in your terminal, press Ctrl+C to stop it.
Your project files and API key stay in place. Run streamlit run dashboard.py from the recoverai/ directory whenever you want to pick back up.

Delete - I don't want to use this again

Remove all project files, uninstall the packages, and optionally revoke your API key.

If the Streamlit server is still running, press Ctrl+C to stop it.

Delete the project directory by running the command for your operating system.

macOS

rm -rf ~/Desktop/recoverai

Windows

Remove-Item -Recurse -Force "$HOME\Desktop\recoverai"

Uninstall the project's Python packages by running:

pip uninstall google-genai pydantic streamlit -y

The GEMINI_API_KEY environment variable only lasts for the current terminal session. Closing your terminal clears it. If you added it to a shell config file like .bashrc or .zshrc, remove that line.
To revoke your API key entirely, visit Google AI Studio, locate your key, and delete it. This is optional. The free tier incurs no charges even if the key stays active.

Nice Work!

You just built RecoverAI, a complete AI-powered revenue recovery agent that ingests failed payment webhooks, diagnoses root causes, and selects optimal recovery actions. You can open it in any browser and showcase every decision the system made.

Here's what you learned:

Built a payment recovery pipeline that combines a deterministic rules engine for clear-cut failures with Gemini 2.5 Flash structured output for ambiguous cases, using Pydantic models to validate both webhook payloads and AI diagnosis objects.
Implemented policy guardrails (fraud blocking, retry limits, quiet hours, duplicate detection) and an expected-value scoring engine that compares recovery actions and picks the mathematically optimal choice for each payment.
Created a Streamlit dashboard with KPI metric cards, interactive bar charts, expandable per-payment decision timelines, and a SHA-256 hash-chained audit log with tamper detection.
Secret Mission: Added a customer risk scoring system that tracks cross-payment behavior, adjusts recovery probabilities per customer, and displays risk profiles in a dedicated dashboard tab.

Ready to quiz yourself?