Build an AWS Cost Guardrail Engine

Introduction

30 Second Summary

Cloud bills rarely fail all at once. A small daily change can grow quietly until an automated response creates a second problem.

In this project, you will build a local Python FinOps guardrail for deterministic AWS spending analysis. Its reports prove that cost controls remain safe when incidents repeat.

What You'll Build

A single local demo replays a runaway-bill incident while its generated report proves that critical production stays protected.

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

• A repeatable anomaly analysis that identifies the two labelled spending spikes without escalating tiny increases.
• A calendar-correct month-end forecast that shows which environments are heading beyond budget.
• An evidence bundle that makes every policy decision inspectable through generated reports with computed safety metrics.
• Secret Mission: Prove production stays protected when a runaway-bill incident is replayed under a hostile policy.

Are there any prerequisites?

No AWS account is required. The setup guide covers Python 3.11 or newer on macOS and Windows.

Before We Start

Before any hands-on work, commit to building a deterministic local FinOps safety simulator that uses explicit fixtures without claiming to reproduce AWS billing behavior. It detects material spending anomalies and forecast overruns while forbidding automatic disruption of critical production workloads.

Set Up the Workspace and Generate Billing Data

Reliable cost controls begin with reproducible inputs. This local FinOps simulator needs an isolated runtime plus explicit billing fixtures before any result can be trusted.

This step creates the Python workspace that generates deterministic AWS service costs. The generator never contacts AWS or changes a cloud resource.

In this step, get ready to:

• Create an isolated Python workspace with the pinned test dependency.
• Build the deterministic billing fixture generator.
• Generate 225 cost records plus safe and unsafe policy fixtures.

Create an isolated Python workspace

A project-local virtual environment keeps this engine's packages separate from the rest of your computer. Python 3.11 or newer is required before you create it.

Start by checking which Python version your terminal can reach.

macOS

• Check your installed Python version by running this command:

python3 --version

The command prints the Python version connected to python3.

Windows

• Check your installed Python version by running this command:

py -3 --version

The command asks the Windows Python launcher for its Python 3 version.

✔️ I see version 3.11 or higher

Your Python runtime meets the project requirement. You can use it to create the isolated environment.

ⓧ I see an older version

The installed runtime is below the project's minimum version. Upgrade to Python 3.11 or newer before continuing.

• Visit the official Python downloads page.
• Install a current Python release for your operating system.
• Return to the platform check above after the installation completes.

ⓧ Command not found

Your terminal cannot find a Python installation. Install Python before creating the workspace.

• Visit the official Python downloads page.
• Install a current Python release for your operating system.
• Restart your terminal after the installation completes.
• Return to the platform check above.

With the runtime confirmed, create a named workspace on your Desktop so every later path stays predictable.

• Press Cmd+Space on macOS or the Windows key on Windows to open system search.
• Type Visual Studio Code and press Enter to open Visual Studio Code.
• Click File in the top menu.
• Click Open Folder.
• Click New Folder in the Desktop file picker.
• Enter aws-cost-guardrail-engine as the folder name.
• Select the aws-cost-guardrail-engine folder.
• Click Open.

You will see the empty aws-cost-guardrail-engine folder in the Explorer sidebar.

• Click the New File icon in the Explorer sidebar.
• Enter requirements.txt as the file name.
• Pin the project's only external dependency by copying this line into requirements.txt:

pytest==9.2.0

Why Pin pytest?

The exact pytest version makes future test results reproducible. Recreating the environment from requirements.txt installs the same release.

• Save requirements.txt with Cmd+S on macOS or Ctrl+S on Windows.

The Explorer sidebar now lists requirements.txt inside aws-cost-guardrail-engine.

The environment command differs by operating system. Each path creates .venv inside the workspace before activating it.

macOS

• Open the integrated terminal by pressing Ctrl+`.
• Create the project-local environment by running this command:

python3 -m venv .venv

Python creates the .venv directory with an isolated interpreter plus package installation area.

• Activate the new environment by running this command:

source .venv/bin/activate

You will see (.venv) at the start of your terminal prompt.

Windows

• Open the integrated terminal by pressing Ctrl+`.
• Create the project-local environment by running this command:

py -3 -m venv .venv

Python creates the .venv directory with an isolated interpreter plus package installation area.

• Activate the new environment by running this command:

.venv\Scripts\Activate.ps1

You will see (.venv) at the start of your PowerShell prompt.

PowerShell Blocking Activation?

PowerShell can block local activation scripts through its execution policy. Apply the documented current-user setting with this command:

Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser

Run the activation command again after PowerShell accepts the change.

Need another set of eyes? Help me activate this virtual environment safely.

• Install the pinned packages into the activated environment by running:

pip install -r requirements.txt

The first installation may take a moment while pip downloads the package. The output finishes with a successful installation of pytest 9.2.0.

Package Installation Failed?

Confirm that your terminal prompt starts with (.venv). Check that requirements.txt contains the exact pinned line shown above.

If the download still fails, help me troubleshoot this pytest installation.

• Verify the installed test runner by running:

pytest --version

You will see pytest 9.2.0 in the terminal. Your isolated runtime now has the exact test dependency the project expects.

• Click the Extensions icon in the left sidebar.
• Search for Python.
• Select the Python extension published by Microsoft.
• Click Install if the extension is not already installed.

Extension installation can take several seconds. The button changes to an installed state when Visual Studio Code is ready to discover the virtual environment.

• Press Cmd+Shift+P on macOS or Ctrl+Shift+P on Windows.
• Enter Python: Select Interpreter in the Command Palette.
• Select the interpreter inside .venv.

The Visual Studio Code status bar now identifies the interpreter from .venv.

• Create .gitignore inside aws-cost-guardrail-engine with the New File icon.
• Copy these local-only paths into .gitignore:

.venv/
__pycache__/
.pytest_cache/
*.pyc
data/guardrail.db

What Does This Ignore?

These entries keep the disposable environment plus generated caches out of version control. The future local database also stays outside commits.

• Save .gitignore.

You will see .gitignore beside requirements.txt in the workspace.

• Create a .vscode folder inside aws-cost-guardrail-engine with the New Folder icon.
• Create settings.json inside the .vscode folder.
• Configure the interpreter plus test discovery by copying this JSON:

{
  "python.defaultInterpreterPath": "${workspaceFolder}/.venv",
  "python.testing.pytestArgs": ["tests"],
  "python.testing.pytestEnabled": true,
  "python.testing.unittestEnabled": false
}

What Do These Settings Control?

• The interpreter path keeps Visual Studio Code attached to the workspace's .venv directory.
• The testing settings enable pytest discovery inside the future tests folder.
• The final setting prevents the editor from mixing unittest discovery into pytest results.

• Save .vscode/settings.json.

The workspace now shows .venv, .vscode, .gitignore, and requirements.txt in the Explorer sidebar.

Build the deterministic fixture generator

Deterministic fixtures give every run the same dates, service costs, anomaly labels, and policies. The generator stores money as strings in JSON so the loader can convert each value into exact decimal arithmetic.

• Create a cost_guardrail folder inside aws-cost-guardrail-engine with the New Folder icon.
• Create __init__.py inside the cost_guardrail folder.
• Identify the package by copying this module description into __init__.py:

"""Deterministic local cost guardrails."""

What Does This File Do?

The module docstring describes the package's local safety boundary. The file also makes the package location explicit in the workspace.

• Save cost_guardrail/__init__.py.

The cost_guardrail folder now contains __init__.py in the Explorer sidebar.

• Create core.py inside the cost_guardrail folder.
• Add the imports, deterministic constants, policy vocabulary, and money helper by copying this first chunk:

import calendar
import datetime
import decimal
import json
import pathlib

D = decimal.Decimal
CENT = D("0.01")
ZERO = D("0.00")
EVALUATION_DATE = datetime.date(2026, 10, 15)
SERVICES = {"Amazon EC2": "10", "Amazon S3": "3", "AWS Lambda": "2", "Amazon RDS": "30", "Amazon CloudWatch": "4"}
MULTIPLIERS = {"development": "1", "staging": "1.5", "production": "3"}
ACTIONS = {"NOTIFY", "REQUIRE_APPROVAL", "SIMULATE_REMEDIATION"}
REMEDIATIONS = {"NONE", "RIGHTSIZE_NONCRITICAL", "STOP_RESOURCE"}


class PolicyValidationError(ValueError):
    pass


def money(value):
    return D(value).quantize(CENT, rounding=decimal.ROUND_HALF_UP)

How Does the Foundation Work?

• The Decimal aliases keep monetary calculations exact to two places.
• The evaluation date fixes every generated run at 2026-10-15.
• The service prices plus environment multipliers create repeatable daily costs.
• The action and remediation sets define the vocabulary written into policy fixtures.

• Save cost_guardrail/core.py.

Visual Studio Code should show the constants plus PolicyValidationError and money() without syntax markers.

• Add the complete generate() function directly below money() by copying this chunk:

def generate(workspace):
    data = pathlib.Path(workspace) / "data"
    data.mkdir(parents=True, exist_ok=True)
    records = []
    for day in range(1, 16):
        day_value = EVALUATION_DATE.replace(day=day)
        for environment, multiplier in MULTIPLIERS.items():
            for service, base in SERVICES.items():
                amount = money(D(base) * D(multiplier))
                expected = False
                if day == 15 and environment == "development" and service == "Amazon EC2":
                    amount, expected = money("45"), True
                if day == 15 and environment == "production" and service == "Amazon RDS":
                    amount, expected = money("240"), True
                records.append({
                    "date": day_value.isoformat(), "environment": environment, "service": service,
                    "amount": f"{amount:.2f}", "critical": environment == "production" and service == "Amazon RDS",
                    "expected_anomaly": expected,
                })
    dataset = {"evaluation_date": EVALUATION_DATE.isoformat(), "budgets": {"development": "1500", "staging": "2500", "production": "4500"}, "records": records}
    rules = [
        {"name": "protect-production", "priority": 10, "match": {"environment": "production", "service": "Amazon RDS", "critical": True}, "action": "REQUIRE_APPROVAL", "remediation": "STOP_RESOURCE", "disruptive": True},
        {"name": "simulate-development", "priority": 20, "match": {"environment": "development", "service": "Amazon EC2", "critical": False}, "action": "SIMULATE_REMEDIATION", "remediation": "RIGHTSIZE_NONCRITICAL", "disruptive": False},
        {"name": "notify-fallback", "priority": 100, "match": {"environment": "*", "service": "*", "critical": "*"}, "action": "NOTIFY", "remediation": "NONE", "disruptive": False},
    ]
    unsafe = [{"name": "unsafe-stop", "priority": 1, "match": {"environment": "production", "service": "Amazon RDS", "critical": True}, "action": "SIMULATE_REMEDIATION", "remediation": "STOP_RESOURCE", "disruptive": True}]
    for name, value in (("costs.json", dataset), ("policy.json", {"version": 1, "rules": rules}), ("unsafe-policy.json", {"version": 1, "rules": unsafe})):
        (data / name).write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return len(records)

What Does the Generator Create?

• The nested loops produce 15 days across three environments plus five services.
• The final day labels the development Amazon EC2 spike plus the critical production Amazon RDS spike as expected anomalies.
• The safe policy separates approval, simulation, and notification responses.
• The unsafe policy deliberately requests a disruptive simulated production action for later validation.

• Save cost_guardrail/core.py.

The generate() function now ends by returning the number of records written.

• Add the fixture loader directly below generate() by copying this final chunk:

def load_data(path):
    payload = json.loads(path.read_text(encoding="utf-8"))
    rows = [{**row, "date": datetime.date.fromisoformat(row["date"]), "amount": money(row["amount"])} for row in payload["records"]]
    return rows, {key: money(value) for key, value in payload["budgets"].items()}

What Does the Loader Preserve?

The loader converts date strings into Python dates. It also converts every amount plus budget into the shared two-place money representation.

• Save cost_guardrail/core.py.

The completed module now exposes generate() plus load_data() without syntax markers.

• Create main.py inside aws-cost-guardrail-engine with the New File icon.
• Add the initial command-line entry point by copying this code:

import argparse
import pathlib

from cost_guardrail.core import generate

parser = argparse.ArgumentParser(description="Local AWS cost guardrail simulator")
parser.add_argument("command", choices=["generate", "run", "demo"])
args = parser.parse_args()
workspace = pathlib.Path.cwd()
if args.command == "generate":
    print(f"Generated {generate(workspace)} deterministic cost records and 2 policy files.")

How Does the CLI Connect?

The parser reads the requested command from the terminal. The initial generate path passes the current workspace into generate() and prints the resulting record count.

• Save main.py.

Seeing Python Syntax Markers?

Check that each block sits in the named file. Confirm that the indented lines inside generate() remain aligned with four spaces per level.

For targeted help, help me compare these Python files with the expected structure.

Generate and inspect the billing fixtures

The CLI now has enough code to create the complete input dataset. This is the step's proof that the isolated runtime plus deterministic generator work together.

Before you run the generator, how many records do you expect from 15 days across three environments and five services?

• Generate the fixture files by running:

python main.py generate

You will see Generated 225 deterministic cost records and 2 policy files. in the terminal.

That is the first deterministic run complete. Your local engine can now reproduce the same billing evidence without making an AWS API call.

Generator Did Not Complete?

Confirm that the terminal is inside aws-cost-guardrail-engine with (.venv) at the prompt. Check that main.py sits beside the cost_guardrail folder.

If Python reports a line number, help me trace the generator failure.

• Expand the generated data folder in the Explorer sidebar.
• Select costs.json.

You will see an evaluation date of 2026-10-15 plus development, staging, and production budgets. The records span 2026-10-01 through 2026-10-15.

• Search costs.json for "expected_anomaly": true with Cmd+F on macOS or Ctrl+F on Windows.

You will find two labeled records. Development Amazon EC2 has an amount of 45.00 while critical production Amazon RDS has an amount of 240.00 on the evaluation date.

• Select policy.json in the Explorer sidebar.

You will see the protect-production, simulate-development, and notify-fallback rules. Critical production uses REQUIRE_APPROVAL while development uses SIMULATE_REMEDIATION.

• Select unsafe-policy.json in the Explorer sidebar.

You will see the unsafe-stop rule request STOP_RESOURCE for critical production Amazon RDS with disruptive set to true.

Why Keep an Unsafe Fixture?

The unsafe policy is hostile test input. A later safety boundary must reject it before any decision can be treated as valid.

Use this reference to compare every authored file before the checkpoint.

✔️ Awesome, I've got everything!

Your environment is active. Save every file before continuing.

ⓧ I'd like to double check the full code

pytest==9.2.0

.venv/
__pycache__/
.pytest_cache/
*.pyc
data/guardrail.db

{
  "python.defaultInterpreterPath": "${workspaceFolder}/.venv",
  "python.testing.pytestArgs": ["tests"],
  "python.testing.pytestEnabled": true,
  "python.testing.unittestEnabled": false
}

"""Deterministic local cost guardrails."""

import calendar
import datetime
import decimal
import json
import pathlib

D = decimal.Decimal
CENT = D("0.01")
ZERO = D("0.00")
EVALUATION_DATE = datetime.date(2026, 10, 15)
SERVICES = {"Amazon EC2": "10", "Amazon S3": "3", "AWS Lambda": "2", "Amazon RDS": "30", "Amazon CloudWatch": "4"}
MULTIPLIERS = {"development": "1", "staging": "1.5", "production": "3"}
ACTIONS = {"NOTIFY", "REQUIRE_APPROVAL", "SIMULATE_REMEDIATION"}
REMEDIATIONS = {"NONE", "RIGHTSIZE_NONCRITICAL", "STOP_RESOURCE"}


class PolicyValidationError(ValueError):
    pass


def money(value):
    return D(value).quantize(CENT, rounding=decimal.ROUND_HALF_UP)


def generate(workspace):
    data = pathlib.Path(workspace) / "data"
    data.mkdir(parents=True, exist_ok=True)
    records = []
    for day in range(1, 16):
        day_value = EVALUATION_DATE.replace(day=day)
        for environment, multiplier in MULTIPLIERS.items():
            for service, base in SERVICES.items():
                amount = money(D(base) * D(multiplier))
                expected = False
                if day == 15 and environment == "development" and service == "Amazon EC2":
                    amount, expected = money("45"), True
                if day == 15 and environment == "production" and service == "Amazon RDS":
                    amount, expected = money("240"), True
                records.append({
                    "date": day_value.isoformat(), "environment": environment, "service": service,
                    "amount": f"{amount:.2f}", "critical": environment == "production" and service == "Amazon RDS",
                    "expected_anomaly": expected,
                })
    dataset = {"evaluation_date": EVALUATION_DATE.isoformat(), "budgets": {"development": "1500", "staging": "2500", "production": "4500"}, "records": records}
    rules = [
        {"name": "protect-production", "priority": 10, "match": {"environment": "production", "service": "Amazon RDS", "critical": True}, "action": "REQUIRE_APPROVAL", "remediation": "STOP_RESOURCE", "disruptive": True},
        {"name": "simulate-development", "priority": 20, "match": {"environment": "development", "service": "Amazon EC2", "critical": False}, "action": "SIMULATE_REMEDIATION", "remediation": "RIGHTSIZE_NONCRITICAL", "disruptive": False},
        {"name": "notify-fallback", "priority": 100, "match": {"environment": "*", "service": "*", "critical": "*"}, "action": "NOTIFY", "remediation": "NONE", "disruptive": False},
    ]
    unsafe = [{"name": "unsafe-stop", "priority": 1, "match": {"environment": "production", "service": "Amazon RDS", "critical": True}, "action": "SIMULATE_REMEDIATION", "remediation": "STOP_RESOURCE", "disruptive": True}]
    for name, value in (("costs.json", dataset), ("policy.json", {"version": 1, "rules": rules}), ("unsafe-policy.json", {"version": 1, "rules": unsafe})):
        (data / name).write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return len(records)


def load_data(path):
    payload = json.loads(path.read_text(encoding="utf-8"))
    rows = [{**row, "date": datetime.date.fromisoformat(row["date"]), "amount": money(row["amount"])} for row in payload["records"]]
    return rows, {key: money(value) for key, value in payload["budgets"].items()}

import argparse
import pathlib

from cost_guardrail.core import generate

parser = argparse.ArgumentParser(description="Local AWS cost guardrail simulator")
parser.add_argument("command", choices=["generate", "run", "demo"])
args = parser.parse_args()
workspace = pathlib.Path.cwd()
if args.command == "generate":
    print(f"Generated {generate(workspace)} deterministic cost records and 2 policy files.")

The generated data folder should also contain costs.json, policy.json, and unsafe-policy.json.

Your reproducible billing workspace is ready. Next, you will calculate historical baselines and expose the false positive hidden inside a percentage-only anomaly rule.

Detect Spikes and Expose the False Positive

Your deterministic billing fixtures now give the engine something trustworthy to analyze. The next goal is to calculate Decimal baselines for each environment and service.

A ratio-only rule catches obvious spending spikes. It also escalates tiny increases that have little financial impact. You will use pytest to make that weakness visible before changing the rule.

In this step, get ready to:

• Calculate historical spending baselines from prior fixture records.
• Detect ratio spikes while handling missing history and zero baselines.
• Test the detector against a tiny increase that exposes its false positive.

Calculate historical baselines

A baseline represents the average prior daily cost for one environment and service. Keeping those dimensions together prevents one workload from distorting another workload's expected spend.

• Return to Visual Studio Code from the previous step.
• Select cost_guardrail/core.py from your project files.
• Find the end of load_data(path).
• Add the ratio-only detector below load_data(path) by pasting this code:

def anomalies(rows, evaluation_date):
    history, current = {}, {}
    for row in rows:
        key = (row["environment"], row["service"])
        if row["date"] < evaluation_date:
            history.setdefault(key, []).append(row["amount"])
        elif row["date"] == evaluation_date:
            current[key] = row
    found = []
    for key in sorted(current):
        prior = history.get(key, [])
        if not prior:
            continue
        baseline = money(sum(prior, ZERO) / D(len(prior)))
        actual = current[key]["amount"]
        increase = money(actual - baseline)
        if actual > baseline * D("1.50"):
            row = current[key]
            found.append({
                "event_id": f"{evaluation_date}:{row['environment']}:{row['service']}", "evaluation_date": evaluation_date,
                "environment": row["environment"], "service": row["service"], "critical": row["critical"],
                "baseline": baseline, "actual": actual, "increase": increase,
                "ratio": None if baseline == ZERO else (actual / baseline).quantize(D("0.01")),
            })
    return found

What does this detector do?

• The history map groups prior amounts by environment and service.
• The current map stores each service record from the evaluation date.
• The baseline averages prior amounts with Decimal arithmetic through money().
• A missing history group is skipped because the engine has no evidence for a comparison.
• A zero baseline stores None as the ratio because division by zero has no valid result.
• The temporary condition flags an incident whenever its current amount exceeds 150% of its baseline.

• Save cost_guardrail/core.py.
• Check the editor diagnostics for Python syntax problems.

You should see no syntax warning for the new function.

What does it find in the fixture?

• Development Amazon EC2 rises from a 10.00 baseline to 45.00.
• Production Amazon RDS rises from a 90.00 baseline to 240.00.
• Both records exceed the temporary ratio threshold.

Seeing a syntax warning?

Check that anomalies() begins after load_data() has finished. Confirm that every line inside the function uses four spaces of indentation.

Help me check the indentation in my anomaly detector.

Add detector edge-case tests

The obvious spikes prove the rule can react to large changes. Edge cases reveal whether the same rule behaves safely when its evidence is incomplete or financially insignificant.

• Create a tests folder beside the existing cost_guardrail folder using Visual Studio Code's folder creation control.
• Create test_guardrail.py inside the new tests folder using the file creation control.

You should now see tests/test_guardrail.py in your project files.

• Add the intermediate detector tests to tests/test_guardrail.py by pasting this code:

import decimal

from cost_guardrail.core import EVALUATION_DATE, anomalies, money


def row(day, amount):
    return {"date": day, "environment": "development", "service": "Amazon EC2", "amount": money(amount), "critical": False, "expected_anomaly": False}


def test_anomaly_thresholds_and_missing_history():
    day = EVALUATION_DATE
    assert anomalies([row(day, "100")], day) == []
    assert anomalies([row(day.replace(day=14), "10"), row(day, "45")], day)[0]["increase"] == decimal.Decimal("35.00")
    assert anomalies([row(day.replace(day=14), "1"), row(day, "2")], day) == []


def test_zero_baseline_is_supported():
    found = anomalies([row(EVALUATION_DATE.replace(day=14), "0"), row(EVALUATION_DATE, "25")], EVALUATION_DATE)
    assert len(found) == 1 and found[0]["ratio"] is None

What do these tests cover?

• The row() helper creates focused cost records without changing the generated fixture files.
• The first assertion confirms that a current record without prior history produces no incident.
• The second assertion confirms that a jump from 10.00 to 45.00 produces an increase of 35.00.
• The third assertion defines the safety expectation for a tiny increase from 1.00 to 2.00.
• The zero-baseline test confirms that the detector records None instead of dividing by zero.

• Save tests/test_guardrail.py.

Run the planned failure

Before you run the tests, do you think a rule based only on percentage change will satisfy every assertion?

• Execute the detector tests by running this command:

pytest -q tests/test_guardrail.py

You should see one failing test. The zero-baseline test should pass.

This failure is intentional

The ratio-only detector treats a rise from 1.00 to 2.00 as an anomaly because the amount doubled. The absolute increase is only 1.00, so the alert has no meaningful cost impact.

This failed assertion proves that percentage change alone cannot represent material spending risk.

Seeing a different failure?

A collection or import failure means the tests did not reach the designed obstacle. Confirm that your virtual environment from the previous step is still active.

Help me diagnose my unexpected pytest failure.

• Keep the ratio-only implementation unchanged so the planned failing test remains visible at this checkpoint.

✔️ Awesome, I've got everything!

Your detector catches the two large fixture spikes. Your tests now preserve the tiny-spike false positive as a visible engineering problem.

ⓧ I'd like to double check the full code

Compare the two files you changed in this step with the intermediate versions below.

import calendar
import datetime
import decimal
import json
import pathlib

D = decimal.Decimal
CENT = D("0.01")
ZERO = D("0.00")
EVALUATION_DATE = datetime.date(2026, 10, 15)
SERVICES = {"Amazon EC2": "10", "Amazon S3": "3", "AWS Lambda": "2", "Amazon RDS": "30", "Amazon CloudWatch": "4"}
MULTIPLIERS = {"development": "1", "staging": "1.5", "production": "3"}
ACTIONS = {"NOTIFY", "REQUIRE_APPROVAL", "SIMULATE_REMEDIATION"}
REMEDIATIONS = {"NONE", "RIGHTSIZE_NONCRITICAL", "STOP_RESOURCE"}


class PolicyValidationError(ValueError):
    pass


def money(value):
    return D(value).quantize(CENT, rounding=decimal.ROUND_HALF_UP)


def generate(workspace):
    data = pathlib.Path(workspace) / "data"
    data.mkdir(parents=True, exist_ok=True)
    records = []
    for day in range(1, 16):
        day_value = EVALUATION_DATE.replace(day=day)
        for environment, multiplier in MULTIPLIERS.items():
            for service, base in SERVICES.items():
                amount = money(D(base) * D(multiplier))
                expected = False
                if day == 15 and environment == "development" and service == "Amazon EC2":
                    amount, expected = money("45"), True
                if day == 15 and environment == "production" and service == "Amazon RDS":
                    amount, expected = money("240"), True
                records.append({
                    "date": day_value.isoformat(), "environment": environment, "service": service,
                    "amount": f"{amount:.2f}", "critical": environment == "production" and service == "Amazon RDS",
                    "expected_anomaly": expected,
                })
    dataset = {"evaluation_date": EVALUATION_DATE.isoformat(), "budgets": {"development": "1500", "staging": "2500", "production": "4500"}, "records": records}
    rules = [
        {"name": "protect-production", "priority": 10, "match": {"environment": "production", "service": "Amazon RDS", "critical": True}, "action": "REQUIRE_APPROVAL", "remediation": "STOP_RESOURCE", "disruptive": True},
        {"name": "simulate-development", "priority": 20, "match": {"environment": "development", "service": "Amazon EC2", "critical": False}, "action": "SIMULATE_REMEDIATION", "remediation": "RIGHTSIZE_NONCRITICAL", "disruptive": False},
        {"name": "notify-fallback", "priority": 100, "match": {"environment": "*", "service": "*", "critical": "*"}, "action": "NOTIFY", "remediation": "NONE", "disruptive": False},
    ]
    unsafe = [{"name": "unsafe-stop", "priority": 1, "match": {"environment": "production", "service": "Amazon RDS", "critical": True}, "action": "SIMULATE_REMEDIATION", "remediation": "STOP_RESOURCE", "disruptive": True}]
    for name, value in (("costs.json", dataset), ("policy.json", {"version": 1, "rules": rules}), ("unsafe-policy.json", {"version": 1, "rules": unsafe})):
        (data / name).write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return len(records)


def load_data(path):
    payload = json.loads(path.read_text(encoding="utf-8"))
    rows = [{**row, "date": datetime.date.fromisoformat(row["date"]), "amount": money(row["amount"])} for row in payload["records"]]
    return rows, {key: money(value) for key, value in payload["budgets"].items()}


def anomalies(rows, evaluation_date):
    history, current = {}, {}
    for row in rows:
        key = (row["environment"], row["service"])
        if row["date"] < evaluation_date:
            history.setdefault(key, []).append(row["amount"])
        elif row["date"] == evaluation_date:
            current[key] = row
    found = []
    for key in sorted(current):
        prior = history.get(key, [])
        if not prior:
            continue
        baseline = money(sum(prior, ZERO) / D(len(prior)))
        actual = current[key]["amount"]
        increase = money(actual - baseline)
        if actual > baseline * D("1.50"):
            row = current[key]
            found.append({
                "event_id": f"{evaluation_date}:{row['environment']}:{row['service']}", "evaluation_date": evaluation_date,
                "environment": row["environment"], "service": row["service"], "critical": row["critical"],
                "baseline": baseline, "actual": actual, "increase": increase,
                "ratio": None if baseline == ZERO else (actual / baseline).quantize(D("0.01")),
            })
    return found

import decimal

from cost_guardrail.core import EVALUATION_DATE, anomalies, money


def row(day, amount):
    return {"date": day, "environment": "development", "service": "Amazon EC2", "amount": money(amount), "critical": False, "expected_anomaly": False}


def test_anomaly_thresholds_and_missing_history():
    day = EVALUATION_DATE
    assert anomalies([row(day, "100")], day) == []
    assert anomalies([row(day.replace(day=14), "10"), row(day, "45")], day)[0]["increase"] == decimal.Decimal("35.00")
    assert anomalies([row(day.replace(day=14), "1"), row(day, "2")], day) == []


def test_zero_baseline_is_supported():
    found = anomalies([row(EVALUATION_DATE.replace(day=14), "0"), row(EVALUATION_DATE, "25")], EVALUATION_DATE)
    assert len(found) == 1 and found[0]["ratio"] is None

You have turned a vague detector weakness into reproducible test evidence. Next, you will add a dollar materiality gate and prove that calendar-correct forecasts handle budget edge cases.

Apply Materiality Gates and Forecast Spend

Your ratio-only detector now catches both intended spikes. The tiny increase from 1.00 to 2.00 also proves that a high percentage can represent trivial spend.

A materiality gate combines relative growth with a minimum dollar increase. Month-end forecasting also needs Decimal arithmetic plus the real number of days in the evaluation month.

In this step, get ready to:

• Require percentage growth plus a minimum 20.00 increase.
• Project month-end spend with Decimal arithmetic.
• Test leap-year month length plus zero-budget utilization.

Make small spikes harmless

The ratio threshold measures relative change. The minimum dollar threshold filters out changes that have little financial impact.

• Switch back to cost_guardrail/core.py in Visual Studio Code.
• Find the complete anomalies() function.
• Replace that function with this version:

def anomalies(rows, evaluation_date):
    history, current = {}, {}
    for row in rows:
        key = (row["environment"], row["service"])
        if row["date"] < evaluation_date:
            history.setdefault(key, []).append(row["amount"])
        elif row["date"] == evaluation_date:
            current[key] = row
    found = []
    for key in sorted(current):
        prior = history.get(key, [])
        if not prior:
            continue
        baseline = money(sum(prior, ZERO) / D(len(prior)))
        actual = current[key]["amount"]
        increase = money(actual - baseline)
        if actual > baseline * D("1.50") and increase >= money("20"):
            row = current[key]
            found.append({
                "event_id": f"{evaluation_date}:{row['environment']}:{row['service']}", "evaluation_date": evaluation_date,
                "environment": row["environment"], "service": row["service"], "critical": row["critical"],
                "baseline": baseline, "actual": actual, "increase": increase,
                "ratio": None if baseline == ZERO else (actual / baseline).quantize(D("0.01")),
            })
    return found

What does this code do?

• The history map keeps earlier amounts grouped by environment plus service.
• The current map keeps each observation from the evaluation date.
• The first comparison requires the current amount to exceed 150% of its baseline.
• The second comparison requires an increase of at least 20.00.
• A zero baseline still produces an anomaly when the increase is material. Its ratio remains None because division by zero has no valid percentage result.

• Save cost_guardrail/core.py.
• Confirm the final condition contains both comparisons on the same line.

Seeing a syntax warning?

Replace the entire anomalies() function so the indentation stays consistent. Check that the condition ends with a colon.

Ask for help with the materiality condition.

Forecast month-end spend

An elapsed-day forecast extends the average spend so far across the full month. calendar.monthrange supplies the correct month length for dates such as February 2024.

• Place the following function directly below anomalies() in cost_guardrail/core.py.

def forecasts(rows, budgets, evaluation_date):
    month_days = calendar.monthrange(evaluation_date.year, evaluation_date.month)[1]
    result = []
    for environment in sorted(budgets):
        mtd = money(sum((row["amount"] for row in rows if row["environment"] == environment and row["date"].year == evaluation_date.year and row["date"].month == evaluation_date.month and row["date"] <= evaluation_date), ZERO))
        daily = money(mtd / D(evaluation_date.day))
        projected, budget = money(daily * D(month_days)), budgets[environment]
        result.append({"environment": environment, "month_to_date": mtd, "daily_average": daily, "projected_monthly": projected, "budget": budget, "overrun": projected > budget, "overrun_amount": money(max(projected - budget, ZERO)), "utilization_percent": None if budget == ZERO else (projected / budget * D(100)).quantize(D("0.01"))})
    return result

How does the forecast work?

• The month-to-date total includes records from the evaluation month through the evaluation date.
• The daily average divides that total by the number of elapsed days.
• The projection multiplies the daily average by the calendar month length.
• The overrun amount never falls below 0.00.
• A zero budget produces None utilization because a percentage would require division by zero.

• Save cost_guardrail/core.py.

Before you run the tests, do you expect the insignificant-spike assertion to pass now?

• Test the corrected detector by running:

pytest -q tests/test_guardrail.py

What does this check?

This command runs the analytics tests in tests/test_guardrail.py with quiet output. The existing tiny-spike case now exercises the dollar gate.

Good. You should see the current tests pass with no failure for the 1.00-to-2.00 increase.

Is the tiny-spike test still failing?

Check that the condition uses increase >= money("20"). Confirm that increase is calculated before the condition.

Help me diagnose the failing threshold test.

Prove the calendar edge cases

Boundary tests make the policy measurable. The anomaly test covers missing history plus both materiality gates.

• Switch back to tests/test_guardrail.py.
• Replace the existing anomaly threshold test with this final version:

def test_anomaly_thresholds_and_missing_history():
    day = EVALUATION_DATE
    assert anomalies([row(day, "100")], day) == []
    assert anomalies([row(day.replace(day=14), "10"), row(day, "45")], day)[0]["increase"] == decimal.Decimal("35.00")
    assert anomalies([row(day.replace(day=14), "1"), row(day, "2")], day) == []

What does this test prove?

• A current observation without history produces no anomaly.
• A baseline of 10.00 plus an actual amount of 45.00 produces an increase of 35.00.
• A 100% increase worth only 1.00 stays out of the anomaly list.

• Save tests/test_guardrail.py.

Before you run the updated test, which assertion now depends on both gates?

• Verify the final anomaly behavior by running:

pytest -q tests/test_guardrail.py

What does this run confirm?

The test runner checks missing history plus the material increase. It also confirms that the percentage-only false positive is gone.

That test run confirms the detector now treats financial impact as part of anomaly severity.

Seeing an anomaly assertion failure?

Compare each amount with the test exactly. Check that the second assertion expects decimal.Decimal("35.00").

Help me trace the failing anomaly assertion.

Forecasting has two useful edge cases. Leap-year February checks the calendar calculation. A zero-dollar budget checks safe division behavior.

• Place the following test directly below test_zero_baseline_is_supported() in tests/test_guardrail.py.

def test_leap_year_forecast_and_zero_budget():
    day = datetime.date(2024, 2, 10)
    rows = [row(day.replace(day=value), "10") for value in range(1, 11)]
    result = forecasts(rows, {"development": money("0")}, day)[0]
    assert result["projected_monthly"] == decimal.Decimal("290.00")
    assert result["overrun"] and result["utilization_percent"] is None

What does this test prove?

• Ten days at 10.00 per day create a month-to-date total of 100.00.
• February 2024 has 29 days. The projected total is therefore 290.00.
• A projection above a zero budget is an overrun.
• Zero-budget utilization remains None.

• Save tests/test_guardrail.py.

Before the final run, do you expect the leap-year projection to equal 280.00 or 290.00?

• Verify the anomaly plus forecast suite by running:

pytest -q tests/test_guardrail.py

What does the final run verify?

This run checks the corrected materiality gate plus the calendar-aware forecast. It also checks zero-baseline anomaly handling plus zero-budget utilization.

You've locked down the analytics boundary. You should see every test pass with no failures.

Is the forecast test failing?

Check that calendar.monthrange receives the evaluation year plus month. Confirm that the daily average divides by evaluation_date.day.

Help me debug the forecast calculation.

✔️ Awesome, I've got everything!

Your materiality gate plus forecast tests are passing. Double check that both edited files are saved.

ⓧ I'd like to double check the full code

Compare cost_guardrail/core.py plus tests/test_guardrail.py with these cumulative versions.

import calendar
import datetime
import decimal
import json
import pathlib

D = decimal.Decimal
CENT = D("0.01")
ZERO = D("0.00")
EVALUATION_DATE = datetime.date(2026, 10, 15)
SERVICES = {"Amazon EC2": "10", "Amazon S3": "3", "AWS Lambda": "2", "Amazon RDS": "30", "Amazon CloudWatch": "4"}
MULTIPLIERS = {"development": "1", "staging": "1.5", "production": "3"}
ACTIONS = {"NOTIFY", "REQUIRE_APPROVAL", "SIMULATE_REMEDIATION"}
REMEDIATIONS = {"NONE", "RIGHTSIZE_NONCRITICAL", "STOP_RESOURCE"}


class PolicyValidationError(ValueError):
    pass


def money(value):
    return D(value).quantize(CENT, rounding=decimal.ROUND_HALF_UP)


def generate(workspace):
    data = pathlib.Path(workspace) / "data"
    data.mkdir(parents=True, exist_ok=True)
    records = []
    for day in range(1, 16):
        day_value = EVALUATION_DATE.replace(day=day)
        for environment, multiplier in MULTIPLIERS.items():
            for service, base in SERVICES.items():
                amount = money(D(base) * D(multiplier))
                expected = False
                if day == 15 and environment == "development" and service == "Amazon EC2":
                    amount, expected = money("45"), True
                if day == 15 and environment == "production" and service == "Amazon RDS":
                    amount, expected = money("240"), True
                records.append({
                    "date": day_value.isoformat(), "environment": environment, "service": service,
                    "amount": f"{amount:.2f}", "critical": environment == "production" and service == "Amazon RDS",
                    "expected_anomaly": expected,
                })
    dataset = {"evaluation_date": EVALUATION_DATE.isoformat(), "budgets": {"development": "1500", "staging": "2500", "production": "4500"}, "records": records}
    rules = [
        {"name": "protect-production", "priority": 10, "match": {"environment": "production", "service": "Amazon RDS", "critical": True}, "action": "REQUIRE_APPROVAL", "remediation": "STOP_RESOURCE", "disruptive": True},
        {"name": "simulate-development", "priority": 20, "match": {"environment": "development", "service": "Amazon EC2", "critical": False}, "action": "SIMULATE_REMEDIATION", "remediation": "RIGHTSIZE_NONCRITICAL", "disruptive": False},
        {"name": "notify-fallback", "priority": 100, "match": {"environment": "*", "service": "*", "critical": "*"}, "action": "NOTIFY", "remediation": "NONE", "disruptive": False},
    ]
    unsafe = [{"name": "unsafe-stop", "priority": 1, "match": {"environment": "production", "service": "Amazon RDS", "critical": True}, "action": "SIMULATE_REMEDIATION", "remediation": "STOP_RESOURCE", "disruptive": True}]
    for name, value in (("costs.json", dataset), ("policy.json", {"version": 1, "rules": rules}), ("unsafe-policy.json", {"version": 1, "rules": unsafe})):
        (data / name).write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return len(records)


def load_data(path):
    payload = json.loads(path.read_text(encoding="utf-8"))
    rows = [{**row, "date": datetime.date.fromisoformat(row["date"]), "amount": money(row["amount"])} for row in payload["records"]]
    return rows, {key: money(value) for key, value in payload["budgets"].items()}


def anomalies(rows, evaluation_date):
    history, current = {}, {}
    for row in rows:
        key = (row["environment"], row["service"])
        if row["date"] < evaluation_date:
            history.setdefault(key, []).append(row["amount"])
        elif row["date"] == evaluation_date:
            current[key] = row
    found = []
    for key in sorted(current):
        prior = history.get(key, [])
        if not prior:
            continue
        baseline = money(sum(prior, ZERO) / D(len(prior)))
        actual = current[key]["amount"]
        increase = money(actual - baseline)
        if actual > baseline * D("1.50") and increase >= money("20"):
            row = current[key]
            found.append({
                "event_id": f"{evaluation_date}:{row['environment']}:{row['service']}", "evaluation_date": evaluation_date,
                "environment": row["environment"], "service": row["service"], "critical": row["critical"],
                "baseline": baseline, "actual": actual, "increase": increase,
                "ratio": None if baseline == ZERO else (actual / baseline).quantize(D("0.01")),
            })
    return found


def forecasts(rows, budgets, evaluation_date):
    month_days = calendar.monthrange(evaluation_date.year, evaluation_date.month)[1]
    result = []
    for environment in sorted(budgets):
        mtd = money(sum((row["amount"] for row in rows if row["environment"] == environment and row["date"].year == evaluation_date.year and row["date"].month == evaluation_date.month and row["date"] <= evaluation_date), ZERO))
        daily = money(mtd / D(evaluation_date.day))
        projected, budget = money(daily * D(month_days)), budgets[environment]
        result.append({"environment": environment, "month_to_date": mtd, "daily_average": daily, "projected_monthly": projected, "budget": budget, "overrun": projected > budget, "overrun_amount": money(max(projected - budget, ZERO)), "utilization_percent": None if budget == ZERO else (projected / budget * D(100)).quantize(D("0.01"))})
    return result

import datetime
import decimal
import json

import pytest

from cost_guardrail.core import EVALUATION_DATE, PolicyValidationError, anomalies, forecasts, generate, load_policy, money, validate_policy
from cost_guardrail.engine import demo, run


def row(day, amount):
    return {"date": day, "environment": "development", "service": "Amazon EC2", "amount": money(amount), "critical": False, "expected_anomaly": False}


def test_anomaly_thresholds_and_missing_history():
    day = EVALUATION_DATE
    assert anomalies([row(day, "100")], day) == []
    assert anomalies([row(day.replace(day=14), "10"), row(day, "45")], day)[0]["increase"] == decimal.Decimal("35.00")
    assert anomalies([row(day.replace(day=14), "1"), row(day, "2")], day) == []


def test_zero_baseline_is_supported():
    found = anomalies([row(EVALUATION_DATE.replace(day=14), "0"), row(EVALUATION_DATE, "25")], EVALUATION_DATE)
    assert len(found) == 1 and found[0]["ratio"] is None


def test_leap_year_forecast_and_zero_budget():
    day = datetime.date(2024, 2, 10)
    rows = [row(day.replace(day=value), "10") for value in range(1, 11)]
    result = forecasts(rows, {"development": money("0")}, day)[0]
    assert result["projected_monthly"] == decimal.Decimal("290.00")
    assert result["overrun"] and result["utilization_percent"] is None

Your engine now filters immaterial spikes. It also produces calendar-correct month-end projections. Next, you'll place safe policy boundaries around every decision.

Enforce Safe Persistent Decisions

Your analytics now detect material anomalies and forecast budget overruns with calendar-correct arithmetic. Those signals give the engine enough evidence to consider a response.

A cost response can create an outage when its policy is malformed or its history disappears after a restart. This step adds strict JSON policy validation plus SQLite persistence for durable idempotency.

In this step, get ready to:

• Reject malformed policies and unsafe remediation rules.
• Persist decisions, simulations, and audit events in SQLite.
• Block critical production remediation and suppress replayed events.

Validate policy files before selection

A policy controls what the engine does after detecting an anomaly. Exact fields and allowed values keep incomplete rules from reaching the decision path.

• Return to the cost_guardrail/core.py file in your VS Code workspace.
• Add the policy validator below forecasts() by pasting this code:

def validate_policy(policy):
    if not isinstance(policy, dict) or policy.get("version") != 1 or not isinstance(policy.get("rules"), list) or not policy["rules"]:
        raise PolicyValidationError("policy requires version 1 and a non-empty rules list")
    required = {"name", "priority", "match", "action", "remediation", "disruptive"}
    for rule in policy["rules"]:
        if not isinstance(rule, dict) or set(rule) != required or set(rule.get("match", {})) != {"environment", "service", "critical"}:
            raise PolicyValidationError("malformed rule")
        if rule["action"] not in ACTIONS:
            raise PolicyValidationError(f"unknown action: {rule['action']}")
        if rule["remediation"] not in REMEDIATIONS:
            raise PolicyValidationError(f"unknown remediation: {rule['remediation']}")
        if rule["action"] == "SIMULATE_REMEDIATION" and (rule["disruptive"] or rule["remediation"] == "NONE"):
            raise PolicyValidationError("simulated remediation must be non-disruptive")
        if rule["action"] == "NOTIFY" and (rule["disruptive"] or rule["remediation"] != "NONE"):
            raise PolicyValidationError("NOTIFY cannot remediate")
    return sorted(policy["rules"], key=lambda rule: rule["priority"])

What does this validation enforce?

• The top-level object must use policy version 1 with at least one rule.
• Each rule must contain the exact required fields and match keys.
• A simulated remediation must name a non-disruptive control.
• A notification rule cannot carry a remediation or disruption flag.

• Save cost_guardrail/core.py.
• Confirm the existing analytics remain green by running:

pytest -q tests/test_guardrail.py

Your existing anomaly and forecast tests should still pass. The new validator has not changed either analytics path.

Seeing a policy name error?

Confirm that ACTIONS, REMEDIATIONS, and PolicyValidationError remain above the new function in cost_guardrail/core.py.

Use the traceback line to locate the missing identifier. Help me diagnose the policy validator failure.

Validated rules still need deterministic loading and selection. Priority sorting makes the first matching rule authoritative while the final fallback covers unmatched incidents.

• Add the policy loader and selector below validate_policy() by pasting this code:

def load_policy(path):
    policy = json.loads(path.read_text(encoding="utf-8"))
    policy["rules"] = validate_policy(policy)
    return policy


def select_rule(policy, incident):
    for rule in policy["rules"]:
        match = rule["match"]
        if match["environment"] in {"*", incident["environment"]} and match["service"] in {"*", incident["service"]} and (match["critical"] == "*" or match["critical"] == incident["critical"]):
            return rule
    raise PolicyValidationError("no fallback rule matched")

How does rule selection work?

The loader parses the policy before sorting validated rules by priority. The selector returns the first rule whose environment, service, and criticality values match the incident.

A wildcard can match any value. A missing fallback raises a validation error instead of allowing an unexplained decision.

• Save cost_guardrail/core.py.
• Replace the import section at the top of tests/test_guardrail.py with this code:

import datetime
import decimal

import pytest

from cost_guardrail.core import EVALUATION_DATE, PolicyValidationError, anomalies, forecasts, generate, load_policy, money, validate_policy

Why add these imports?

The tests use pytest.raises() to prove invalid policies stop execution. The fixture generator provides both safe and unsafe policy files inside an isolated temporary directory.

• Add these policy tests below the existing forecast test:

def test_malformed_and_unknown_policies_are_rejected():
    with pytest.raises(PolicyValidationError): validate_policy({"version": 1, "rules": []})
    policy = {"version": 1, "rules": [{"name": "bad", "priority": 1, "match": {"environment": "*", "service": "*", "critical": "*"}, "action": "DELETE", "remediation": "NONE", "disruptive": False}]}
    with pytest.raises(PolicyValidationError, match="unknown action"): validate_policy(policy)


def test_unsafe_policy_is_rejected(tmp_path):
    generate(tmp_path)
    with pytest.raises(PolicyValidationError, match="non-disruptive"): load_policy(tmp_path / "data" / "unsafe-policy.json")

What do these tests prove?

• An empty rules list fails before the engine can select a response.
• An unknown action fails even when the surrounding rule has the expected shape.
• A disruptive simulated remediation fails when loaded from the generated unsafe policy.

• Save tests/test_guardrail.py.
• Verify the policy boundary by running:

pytest -q tests/test_guardrail.py

All five tests should pass. The unsafe policy passes the test because the loader rejects it with the expected validation error.

Does an unsafe policy test fail?

Check that the simulated-remediation condition rejects either a disruptive flag or a NONE remediation. Confirm that load_policy() replaces the parsed rules with the validated priority-sorted rules.

Share the failing assertion without changing the generated policy fixture. Help me debug the unsafe policy test.

Persist decisions and audit events

An in-memory set forgets processed events when the process exits. SQLite keeps primary-key claims on disk so a new connection can recognize a replay.

Why SQLite instead of an in-memory set?

SQLite provides a disk-backed database without a separate server. Primary keys give the simulator a transferable model for atomic duplicate suppression.

A distributed production engine would use a durable conditional-write store. This local database proves the restart behavior without adding cloud infrastructure.

• Create cost_guardrail/engine.py inside the existing cost_guardrail folder from the VS Code Explorer sidebar.
• Add the imports plus the database initialization and audit writer by pasting this code:

import datetime
import decimal
import json
import pathlib
import sqlite3

from cost_guardrail.core import EVALUATION_DATE, PolicyValidationError, anomalies, forecasts, generate, load_data, load_policy, select_rule


class Store:
    def __init__(self, path):
        self.path = pathlib.Path(path)
        with sqlite3.connect(str(self.path)) as connection:
            connection.executescript("""
            CREATE TABLE IF NOT EXISTS processed(event_id TEXT PRIMARY KEY, decision TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS simulations(event_id TEXT PRIMARY KEY, remediation TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS audit(sequence INTEGER PRIMARY KEY AUTOINCREMENT, evaluation_date TEXT NOT NULL, event_type TEXT NOT NULL, event_id TEXT NOT NULL, payload TEXT NOT NULL);
            """)

    def add_audit(self, day, kind, event_id, payload):
        with sqlite3.connect(str(self.path)) as connection:
            connection.execute("INSERT INTO audit(evaluation_date,event_type,event_id,payload) VALUES(?,?,?,?)", (day, kind, event_id, json.dumps(payload, sort_keys=True)))

What does the store initialize?

• The processed table gives each decision event one primary-key claim.
• The simulations table gives each simulated remediation one primary-key claim.
• The audit table assigns an increasing sequence number to every recorded event.
• The add_audit() method serializes each payload into a stable JSON string.

• Save cost_guardrail/engine.py.
• Confirm the existing test suite still passes by running:

pytest -q tests/test_guardrail.py

All five tests should remain green. This check confirms the new module has not changed the analytics or policy behavior.

Seeing a SQLite import problem?

Confirm that the import is spelled sqlite3 and that engine.py sits inside the cost_guardrail folder. SQLite comes through Python's standard library.

Use the traceback to identify the failing import or indentation level. Help me debug the Store setup.

The primary-key tables become useful when claims use insert-if-absent behavior. Each duplicate still creates an audit event so suppression remains visible.

• Add the remaining methods inside Store below add_audit() by pasting this code:

    def claim(self, day, decision):
        with sqlite3.connect(str(self.path)) as connection:
            created = connection.execute("INSERT OR IGNORE INTO processed VALUES(?,?)", (decision["event_id"], json.dumps(decision, sort_keys=True))).rowcount == 1
        self.add_audit(day, "DECISION_RECORDED" if created else "DUPLICATE_SUPPRESSED", decision["event_id"], decision)
        return created

    def simulate(self, day, event_id, remediation):
        with sqlite3.connect(str(self.path)) as connection:
            created = connection.execute("INSERT OR IGNORE INTO simulations VALUES(?,?)", (event_id, remediation)).rowcount == 1
        if created:
            self.add_audit(day, "REMEDIATION_SIMULATED", event_id, {"remediation": remediation})

    def list_json(self, table, column):
        with sqlite3.connect(str(self.path)) as connection:
            return [json.loads(row[0]) for row in connection.execute(f"SELECT {column} FROM {table} ORDER BY 1")]

    def audits(self):
        with sqlite3.connect(str(self.path)) as connection:
            rows = connection.execute("SELECT sequence,evaluation_date,event_type,event_id,payload FROM audit ORDER BY sequence").fetchall()
        return [{"sequence": row[0], "evaluation_date": row[1], "event_type": row[2], "event_id": row[3], "payload": json.loads(row[4])} for row in rows]

    def count(self, kind):
        with sqlite3.connect(str(self.path)) as connection:
            return connection.execute("SELECT COUNT(*) FROM audit WHERE event_type=?", (kind,)).fetchone()[0]

    def duplicate_remediations(self):
        with sqlite3.connect(str(self.path)) as connection:
            return connection.execute("SELECT COUNT(*) FROM (SELECT event_id FROM simulations GROUP BY event_id HAVING COUNT(*)>1)").fetchone()[0]

How does persistence suppress replays?

• The claim() method records the first decision and audits later attempts as duplicate suppressions.
• The simulate() method writes one remediation per event ID.
• The read methods expose persisted decisions and append-only audit rows for reporting.
• The duplicate query proves that no event received more than one simulated remediation.

• Save cost_guardrail/engine.py.
• Run the existing tests once more to catch indentation or syntax errors:

pytest -q tests/test_guardrail.py

All five tests should pass again. The policy and analytics behavior remains stable while the persistence layer is assembled.

Seeing an indentation failure?

Confirm that every new method aligns with add_audit() inside the Store class. Each method body needs one additional indentation level.

Compare the first failing line with the method above it. Help me repair the Store method indentation.

Run safe decisions and verify restart behavior

The engine can now join analytics, policy selection, and persistence. A second production guard converts any simulated critical production remediation into an approval requirement.

• Add the serialization helper below the Store class by pasting this code:

def serial(value):
    if isinstance(value, decimal.Decimal): return format(value, "f")
    if isinstance(value, datetime.date): return value.isoformat()
    if isinstance(value, list): return [serial(item) for item in value]
    if isinstance(value, dict): return {key: serial(item) for key, item in value.items()}
    return value

Why serialize the report?

Analytics use Decimal values and date objects for deterministic calculations. The helper converts those values into strings that a later report can encode as JSON.

• Save cost_guardrail/engine.py.
• Confirm the existing tests still pass by running:

pytest -q tests/test_guardrail.py

All five tests should remain green. The serializer is now available to the decision runner.

• Add the first runnable decision orchestrator below serial() by pasting this code:

def run(workspace, day=EVALUATION_DATE, reset=False):
    workspace = pathlib.Path(workspace)
    if not (workspace / "data" / "costs.json").exists(): generate(workspace)
    database = workspace / "data" / "guardrail.db"
    if reset: database.unlink(missing_ok=True)
    rows, budgets = load_data(workspace / "data" / "costs.json")
    policy = load_policy(workspace / "data" / "policy.json")
    found, projected, store = anomalies(rows, day), forecasts(rows, budgets, day), Store(database)
    for incident in found:
        rule = select_rule(policy, incident)
        action, remediation = rule["action"], rule["remediation"]
        explanation = f"Rule {rule['name']} selected {action}."
        if incident["critical"] and incident["environment"] == "production" and action == "SIMULATE_REMEDIATION":
            action, explanation = "REQUIRE_APPROVAL", "Defense in depth blocked critical production remediation."
        status = {"NOTIFY": "NOTIFIED", "REQUIRE_APPROVAL": "BLOCKED", "SIMULATE_REMEDIATION": "SIMULATED"}[action]
        explanation += " No real AWS action exists in this simulator."
        decision = {"event_id": incident["event_id"], "environment": incident["environment"], "service": incident["service"], "action": action, "remediation": remediation, "status": status, "explanation": explanation}
        if store.claim(day.isoformat(), decision) and action == "SIMULATE_REMEDIATION": store.simulate(day.isoformat(), incident["event_id"], remediation)
    truth = {(row["environment"], row["service"]) for row in rows if row["date"] == day and row["expected_anomaly"]}
    detected = {(item["environment"], item["service"]) for item in found}
    tp, fp, fn = len(truth & detected), len(detected - truth), len(truth - detected)
    decisions = store.list_json("processed", "decision")
    metrics = {
        "unsafe_actions_executed": store.count("UNSAFE_ACTION_EXECUTED"),
        "duplicate_remediations": store.duplicate_remediations(),
        "duplicate_suppressions": store.count("DUPLICATE_SUPPRESSED"),
        "blocked_unsafe_actions": sum(item["status"] == "BLOCKED" for item in decisions),
    }
    report = serial({"evaluation_date": day, "anomalies": found, "forecasts": projected, "decisions": decisions, "measurements": metrics})
    return report

What safety boundaries run here?

• The runner loads validated fixtures before analyzing anomalies and forecasts.
• The selected policy determines the proposed action and remediation label.
• The defense-in-depth condition forces critical production simulation into REQUIRE_APPROVAL with a BLOCKED status.
• The store claims each event before recording a simulated development remediation.

• Save cost_guardrail/engine.py.
• Replace main.py with this CLI code:

import argparse
import pathlib

from cost_guardrail.core import generate
from cost_guardrail.engine import run

parser = argparse.ArgumentParser(description="Local AWS cost guardrail simulator")
parser.add_argument("command", choices=["generate", "run"])
parser.add_argument("--reset", action="store_true")
args = parser.parse_args()
workspace = pathlib.Path.cwd()
if args.command == "generate":
    print(f"Generated {generate(workspace)} deterministic cost records and 2 policy files.")
else:
    report = run(workspace, reset=args.reset)
    print(f"Anomalies: {len(report['anomalies'])}")
    for key, value in report["measurements"].items(): print(f"{key}: {value}")

What does the run command expose?

The CLI keeps fixture generation available and adds a local decision run. The reset flag removes the existing database before processing so the first execution starts with empty idempotency state.

• Save main.py.

Before you run the engine, which incident do you expect the policy to simulate and which incident do you expect it to block?

• Create fresh persistent state by running:

python main.py run --reset

You should see two anomalies with one blocked unsafe action and zero duplicate remediations. The command creates data/guardrail.db without contacting AWS or executing a remediation.

Does the run command fail?

Confirm that engine.py is inside the cost_guardrail folder. Check that data/policy.json and data/costs.json remain in the existing data folder.

Use the first traceback frame inside your project to locate the failing function. Help me debug the local decision run.

The first run established persisted decisions. The final metrics complete the evidence by measuring detector quality and policy rejection counts.

• Find the current metrics dictionary near the bottom of run().
• Confirm that it currently looks like this:

    metrics = {
        "unsafe_actions_executed": store.count("UNSAFE_ACTION_EXECUTED"),
        "duplicate_remediations": store.duplicate_remediations(),
        "duplicate_suppressions": store.count("DUPLICATE_SUPPRESSED"),
        "blocked_unsafe_actions": sum(item["status"] == "BLOCKED" for item in decisions),
    }

• Replace that dictionary with the complete measurement set:

    metrics = {
        "precision": decimal.Decimal(tp / (tp + fp) if tp + fp else 0).quantize(decimal.Decimal("0.01")),
        "recall": decimal.Decimal(tp / (tp + fn) if tp + fn else 0).quantize(decimal.Decimal("0.01")),
        "unsafe_actions_executed": store.count("UNSAFE_ACTION_EXECUTED"),
        "duplicate_remediations": store.duplicate_remediations(),
        "duplicate_suppressions": store.count("DUPLICATE_SUPPRESSED"),
        "blocked_unsafe_actions": sum(item["status"] == "BLOCKED" for item in decisions),
        "policy_rejections": store.count("POLICY_REJECTED"),
    }

What do the completed measurements show?

• Precision measures how many detected incidents match the fixture truth labels.
• Recall measures how many truth-labeled incidents the detector found.
• Safety counters expose unsafe executions, blocked actions, and duplicate remediation attempts.
• The policy rejection count remains available for later hostile-policy evidence.

• Save cost_guardrail/engine.py.

Before the replay, do you expect the event count to grow or the existing event IDs to be suppressed?

• Replay the same incidents through a new process by running:

python main.py run

You should see two duplicate suppressions and zero duplicate remediations. SQLite recognized both persisted event IDs after the first process had exited.

• Add the engine import below the core import in tests/test_guardrail.py by pasting this line:

from cost_guardrail.engine import run

Why import the runner?

The integration test calls run() twice against one temporary workspace. The second call creates fresh SQLite connections while retaining the database file.

• Add the persistent safety test below the policy tests:

def test_production_safety_and_persistent_idempotency(tmp_path):
    generate(tmp_path)
    first = run(tmp_path, reset=True)
    assert {item["service"]: item["status"] for item in first["decisions"]} == {"Amazon EC2": "SIMULATED", "Amazon RDS": "BLOCKED"}
    replay = run(tmp_path)
    assert replay["measurements"]["unsafe_actions_executed"] == 0
    assert replay["measurements"]["duplicate_remediations"] == 0
    assert replay["measurements"]["duplicate_suppressions"] == 2

What does the restart test prove?

The development Amazon EC2 incident receives a simulated noncritical response. The critical production Amazon RDS incident remains blocked behind approval.

The replay uses the same database through new connections. Both event IDs are suppressed without creating a second remediation.

• Save tests/test_guardrail.py.

Before the final check, do you expect the unsafe policy or replay assertions to expose a gap?

• Verify policy safety and persistent idempotency by running:

pytest -q tests/test_guardrail.py

All six tests should pass. The result proves malformed policy rejection, unsafe simulation rejection, production protection, and restart-safe duplicate suppression.

Does the replay assertion fail?

Confirm that both processed.event_id and simulations.event_id are primary keys. Check that claim() audits a duplicate whenever the insert creates no row.

Keep the second run(tmp_path) call free of the reset flag so it reuses the first database. Help me debug the persistent replay test.

✔️ Awesome, I've got everything!

Your policy engine rejects unsafe inputs. Your SQLite state now protects decisions and simulations across process restarts.

ⓧ I'd like to double check the full code

Compare these cumulative files with your workspace. Keep the generated fixtures unchanged.

import calendar
import datetime
import decimal
import json
import pathlib

D = decimal.Decimal
CENT = D("0.01")
ZERO = D("0.00")
EVALUATION_DATE = datetime.date(2026, 10, 15)
SERVICES = {"Amazon EC2": "10", "Amazon S3": "3", "AWS Lambda": "2", "Amazon RDS": "30", "Amazon CloudWatch": "4"}
MULTIPLIERS = {"development": "1", "staging": "1.5", "production": "3"}
ACTIONS = {"NOTIFY", "REQUIRE_APPROVAL", "SIMULATE_REMEDIATION"}
REMEDIATIONS = {"NONE", "RIGHTSIZE_NONCRITICAL", "STOP_RESOURCE"}


class PolicyValidationError(ValueError):
    pass


def money(value):
    return D(value).quantize(CENT, rounding=decimal.ROUND_HALF_UP)


def generate(workspace):
    data = pathlib.Path(workspace) / "data"
    data.mkdir(parents=True, exist_ok=True)
    records = []
    for day in range(1, 16):
        day_value = EVALUATION_DATE.replace(day=day)
        for environment, multiplier in MULTIPLIERS.items():
            for service, base in SERVICES.items():
                amount = money(D(base) * D(multiplier))
                expected = False
                if day == 15 and environment == "development" and service == "Amazon EC2":
                    amount, expected = money("45"), True
                if day == 15 and environment == "production" and service == "Amazon RDS":
                    amount, expected = money("240"), True
                records.append({
                    "date": day_value.isoformat(), "environment": environment, "service": service,
                    "amount": f"{amount:.2f}", "critical": environment == "production" and service == "Amazon RDS",
                    "expected_anomaly": expected,
                })
    dataset = {"evaluation_date": EVALUATION_DATE.isoformat(), "budgets": {"development": "1500", "staging": "2500", "production": "4500"}, "records": records}
    rules = [
        {"name": "protect-production", "priority": 10, "match": {"environment": "production", "service": "Amazon RDS", "critical": True}, "action": "REQUIRE_APPROVAL", "remediation": "STOP_RESOURCE", "disruptive": True},
        {"name": "simulate-development", "priority": 20, "match": {"environment": "development", "service": "Amazon EC2", "critical": False}, "action": "SIMULATE_REMEDIATION", "remediation": "RIGHTSIZE_NONCRITICAL", "disruptive": False},
        {"name": "notify-fallback", "priority": 100, "match": {"environment": "*", "service": "*", "critical": "*"}, "action": "NOTIFY", "remediation": "NONE", "disruptive": False},
    ]
    unsafe = [{"name": "unsafe-stop", "priority": 1, "match": {"environment": "production", "service": "Amazon RDS", "critical": True}, "action": "SIMULATE_REMEDIATION", "remediation": "STOP_RESOURCE", "disruptive": True}]
    for name, value in (("costs.json", dataset), ("policy.json", {"version": 1, "rules": rules}), ("unsafe-policy.json", {"version": 1, "rules": unsafe})):
        (data / name).write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return len(records)


def load_data(path):
    payload = json.loads(path.read_text(encoding="utf-8"))
    rows = [{**row, "date": datetime.date.fromisoformat(row["date"]), "amount": money(row["amount"])} for row in payload["records"]]
    return rows, {key: money(value) for key, value in payload["budgets"].items()}


def anomalies(rows, evaluation_date):
    history, current = {}, {}
    for row in rows:
        key = (row["environment"], row["service"])
        if row["date"] < evaluation_date:
            history.setdefault(key, []).append(row["amount"])
        elif row["date"] == evaluation_date:
            current[key] = row
    found = []
    for key in sorted(current):
        prior = history.get(key, [])
        if not prior:
            continue
        baseline = money(sum(prior, ZERO) / D(len(prior)))
        actual = current[key]["amount"]
        increase = money(actual - baseline)
        if actual > baseline * D("1.50") and increase >= money("20"):
            row = current[key]
            found.append({
                "event_id": f"{evaluation_date}:{row['environment']}:{row['service']}", "evaluation_date": evaluation_date,
                "environment": row["environment"], "service": row["service"], "critical": row["critical"],
                "baseline": baseline, "actual": actual, "increase": increase,
                "ratio": None if baseline == ZERO else (actual / baseline).quantize(D("0.01")),
            })
    return found


def forecasts(rows, budgets, evaluation_date):
    month_days = calendar.monthrange(evaluation_date.year, evaluation_date.month)[1]
    result = []
    for environment in sorted(budgets):
        mtd = money(sum((row["amount"] for row in rows if row["environment"] == environment and row["date"].year == evaluation_date.year and row["date"].month == evaluation_date.month and row["date"] <= evaluation_date), ZERO))
        daily = money(mtd / D(evaluation_date.day))
        projected, budget = money(daily * D(month_days)), budgets[environment]
        result.append({"environment": environment, "month_to_date": mtd, "daily_average": daily, "projected_monthly": projected, "budget": budget, "overrun": projected > budget, "overrun_amount": money(max(projected - budget, ZERO)), "utilization_percent": None if budget == ZERO else (projected / budget * D(100)).quantize(D("0.01"))})
    return result


def validate_policy(policy):
    if not isinstance(policy, dict) or policy.get("version") != 1 or not isinstance(policy.get("rules"), list) or not policy["rules"]:
        raise PolicyValidationError("policy requires version 1 and a non-empty rules list")
    required = {"name", "priority", "match", "action", "remediation", "disruptive"}
    for rule in policy["rules"]:
        if not isinstance(rule, dict) or set(rule) != required or set(rule.get("match", {})) != {"environment", "service", "critical"}:
            raise PolicyValidationError("malformed rule")
        if rule["action"] not in ACTIONS:
            raise PolicyValidationError(f"unknown action: {rule['action']}")
        if rule["remediation"] not in REMEDIATIONS:
            raise PolicyValidationError(f"unknown remediation: {rule['remediation']}")
        if rule["action"] == "SIMULATE_REMEDIATION" and (rule["disruptive"] or rule["remediation"] == "NONE"):
            raise PolicyValidationError("simulated remediation must be non-disruptive")
        if rule["action"] == "NOTIFY" and (rule["disruptive"] or rule["remediation"] != "NONE"):
            raise PolicyValidationError("NOTIFY cannot remediate")
    return sorted(policy["rules"], key=lambda rule: rule["priority"])


def load_policy(path):
    policy = json.loads(path.read_text(encoding="utf-8"))
    policy["rules"] = validate_policy(policy)
    return policy


def select_rule(policy, incident):
    for rule in policy["rules"]:
        match = rule["match"]
        if match["environment"] in {"*", incident["environment"]} and match["service"] in {"*", incident["service"]} and (match["critical"] == "*" or match["critical"] == incident["critical"]):
            return rule
    raise PolicyValidationError("no fallback rule matched")

import datetime
import decimal
import json
import pathlib
import sqlite3

from cost_guardrail.core import EVALUATION_DATE, PolicyValidationError, anomalies, forecasts, generate, load_data, load_policy, select_rule


class Store:
    def __init__(self, path):
        self.path = pathlib.Path(path)
        with sqlite3.connect(str(self.path)) as connection:
            connection.executescript("""
            CREATE TABLE IF NOT EXISTS processed(event_id TEXT PRIMARY KEY, decision TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS simulations(event_id TEXT PRIMARY KEY, remediation TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS audit(sequence INTEGER PRIMARY KEY AUTOINCREMENT, evaluation_date TEXT NOT NULL, event_type TEXT NOT NULL, event_id TEXT NOT NULL, payload TEXT NOT NULL);
            """)

    def add_audit(self, day, kind, event_id, payload):
        with sqlite3.connect(str(self.path)) as connection:
            connection.execute("INSERT INTO audit(evaluation_date,event_type,event_id,payload) VALUES(?,?,?,?)", (day, kind, event_id, json.dumps(payload, sort_keys=True)))

    def claim(self, day, decision):
        with sqlite3.connect(str(self.path)) as connection:
            created = connection.execute("INSERT OR IGNORE INTO processed VALUES(?,?)", (decision["event_id"], json.dumps(decision, sort_keys=True))).rowcount == 1
        self.add_audit(day, "DECISION_RECORDED" if created else "DUPLICATE_SUPPRESSED", decision["event_id"], decision)
        return created

    def simulate(self, day, event_id, remediation):
        with sqlite3.connect(str(self.path)) as connection:
            created = connection.execute("INSERT OR IGNORE INTO simulations VALUES(?,?)", (event_id, remediation)).rowcount == 1
        if created:
            self.add_audit(day, "REMEDIATION_SIMULATED", event_id, {"remediation": remediation})

    def list_json(self, table, column):
        with sqlite3.connect(str(self.path)) as connection:
            return [json.loads(row[0]) for row in connection.execute(f"SELECT {column} FROM {table} ORDER BY 1")]

    def audits(self):
        with sqlite3.connect(str(self.path)) as connection:
            rows = connection.execute("SELECT sequence,evaluation_date,event_type,event_id,payload FROM audit ORDER BY sequence").fetchall()
        return [{"sequence": row[0], "evaluation_date": row[1], "event_type": row[2], "event_id": row[3], "payload": json.loads(row[4])} for row in rows]

    def count(self, kind):
        with sqlite3.connect(str(self.path)) as connection:
            return connection.execute("SELECT COUNT(*) FROM audit WHERE event_type=?", (kind,)).fetchone()[0]

    def duplicate_remediations(self):
        with sqlite3.connect(str(self.path)) as connection:
            return connection.execute("SELECT COUNT(*) FROM (SELECT event_id FROM simulations GROUP BY event_id HAVING COUNT(*)>1)").fetchone()[0]


def serial(value):
    if isinstance(value, decimal.Decimal): return format(value, "f")
    if isinstance(value, datetime.date): return value.isoformat()
    if isinstance(value, list): return [serial(item) for item in value]
    if isinstance(value, dict): return {key: serial(item) for key, item in value.items()}
    return value


def run(workspace, day=EVALUATION_DATE, reset=False):
    workspace = pathlib.Path(workspace)
    if not (workspace / "data" / "costs.json").exists(): generate(workspace)
    database = workspace / "data" / "guardrail.db"
    if reset: database.unlink(missing_ok=True)
    rows, budgets = load_data(workspace / "data" / "costs.json")
    policy = load_policy(workspace / "data" / "policy.json")
    found, projected, store = anomalies(rows, day), forecasts(rows, budgets, day), Store(database)
    for incident in found:
        rule = select_rule(policy, incident)
        action, remediation = rule["action"], rule["remediation"]
        explanation = f"Rule {rule['name']} selected {action}."
        if incident["critical"] and incident["environment"] == "production" and action == "SIMULATE_REMEDIATION":
            action, explanation = "REQUIRE_APPROVAL", "Defense in depth blocked critical production remediation."
        status = {"NOTIFY": "NOTIFIED", "REQUIRE_APPROVAL": "BLOCKED", "SIMULATE_REMEDIATION": "SIMULATED"}[action]
        explanation += " No real AWS action exists in this simulator."
        decision = {"event_id": incident["event_id"], "environment": incident["environment"], "service": incident["service"], "action": action, "remediation": remediation, "status": status, "explanation": explanation}
        if store.claim(day.isoformat(), decision) and action == "SIMULATE_REMEDIATION": store.simulate(day.isoformat(), incident["event_id"], remediation)
    truth = {(row["environment"], row["service"]) for row in rows if row["date"] == day and row["expected_anomaly"]}
    detected = {(item["environment"], item["service"]) for item in found}
    tp, fp, fn = len(truth & detected), len(detected - truth), len(truth - detected)
    decisions = store.list_json("processed", "decision")
    metrics = {
        "precision": decimal.Decimal(tp / (tp + fp) if tp + fp else 0).quantize(decimal.Decimal("0.01")),
        "recall": decimal.Decimal(tp / (tp + fn) if tp + fn else 0).quantize(decimal.Decimal("0.01")),
        "unsafe_actions_executed": store.count("UNSAFE_ACTION_EXECUTED"),
        "duplicate_remediations": store.duplicate_remediations(),
        "duplicate_suppressions": store.count("DUPLICATE_SUPPRESSED"),
        "blocked_unsafe_actions": sum(item["status"] == "BLOCKED" for item in decisions),
        "policy_rejections": store.count("POLICY_REJECTED"),
    }
    report = serial({"evaluation_date": day, "anomalies": found, "forecasts": projected, "decisions": decisions, "measurements": metrics})
    return report

import argparse
import pathlib

from cost_guardrail.core import generate
from cost_guardrail.engine import run

parser = argparse.ArgumentParser(description="Local AWS cost guardrail simulator")
parser.add_argument("command", choices=["generate", "run"])
parser.add_argument("--reset", action="store_true")
args = parser.parse_args()
workspace = pathlib.Path.cwd()
if args.command == "generate":
    print(f"Generated {generate(workspace)} deterministic cost records and 2 policy files.")
else:
    report = run(workspace, reset=args.reset)
    print(f"Anomalies: {len(report['anomalies'])}")
    for key, value in report["measurements"].items(): print(f"{key}: {value}")

import datetime
import decimal

import pytest

from cost_guardrail.core import EVALUATION_DATE, PolicyValidationError, anomalies, forecasts, generate, load_policy, money, validate_policy
from cost_guardrail.engine import run


def row(day, amount):
    return {"date": day, "environment": "development", "service": "Amazon EC2", "amount": money(amount), "critical": False, "expected_anomaly": False}


def test_anomaly_thresholds_and_missing_history():
    day = EVALUATION_DATE
    assert anomalies([row(day, "100")], day) == []
    assert anomalies([row(day.replace(day=14), "10"), row(day, "45")], day)[0]["increase"] == decimal.Decimal("35.00")
    assert anomalies([row(day.replace(day=14), "1"), row(day, "2")], day) == []


def test_zero_baseline_is_supported():
    found = anomalies([row(EVALUATION_DATE.replace(day=14), "0"), row(EVALUATION_DATE, "25")], EVALUATION_DATE)
    assert len(found) == 1 and found[0]["ratio"] is None


def test_leap_year_forecast_and_zero_budget():
    day = datetime.date(2024, 2, 10)
    rows = [row(day.replace(day=value), "10") for value in range(1, 11)]
    result = forecasts(rows, {"development": money("0")}, day)[0]
    assert result["projected_monthly"] == decimal.Decimal("290.00")
    assert result["overrun"] and result["utilization_percent"] is None


def test_malformed_and_unknown_policies_are_rejected():
    with pytest.raises(PolicyValidationError): validate_policy({"version": 1, "rules": []})
    policy = {"version": 1, "rules": [{"name": "bad", "priority": 1, "match": {"environment": "*", "service": "*", "critical": "*"}, "action": "DELETE", "remediation": "NONE", "disruptive": False}]}
    with pytest.raises(PolicyValidationError, match="unknown action"): validate_policy(policy)


def test_unsafe_policy_is_rejected(tmp_path):
    generate(tmp_path)
    with pytest.raises(PolicyValidationError, match="non-disruptive"): load_policy(tmp_path / "data" / "unsafe-policy.json")


def test_production_safety_and_persistent_idempotency(tmp_path):
    generate(tmp_path)
    first = run(tmp_path, reset=True)
    assert {item["service"]: item["status"] for item in first["decisions"]} == {"Amazon EC2": "SIMULATED", "Amazon RDS": "BLOCKED"}
    replay = run(tmp_path)
    assert replay["measurements"]["unsafe_actions_executed"] == 0
    assert replay["measurements"]["duplicate_remediations"] == 0
    assert replay["measurements"]["duplicate_suppressions"] == 2

That is the safety boundary working: development control stays simulated while critical production remains protected across replays. Next, you will turn these persisted decisions into inspectable audit evidence and FinOps reports.

Generate Audit Evidence and Reports

The guardrail now blocks critical production control. SQLite also suppresses replayed events across process restarts.

Those outcomes remain claims until the runtime exports evidence. You will turn them into JSON, Markdown, and JSON Lines reports with computed FinOps metrics.

In this step, get ready to:

• Generate three report formats from persisted engine state.
• Add a replay path that records unsafe-policy rejection.
• Prove the final system through tests and architecture documentation.

Write the evidence bundle

The engine already assembles anomalies, forecasts, decisions, audits, and measurements in memory. A report writer turns that runtime state into files that another engineer can inspect without rerunning the analysis.

• In cost_guardrail/engine.py, add this function below serial() by copying the code:

def write_reports(folder, report, audit):
    folder.mkdir(exist_ok=True)
    (folder / "finops-report.json").write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    lines = ["# FinOps Guardrail Report", "", f"Evaluation date: {report['evaluation_date']}", "", "## Measurements"]
    lines += [f"- {key}: {value}" for key, value in report["measurements"].items()]
    lines += ["", "## Anomalies"] + [f"- {item['environment']} / {item['service']}: ${item['actual']} vs ${item['baseline']}" for item in report["anomalies"]]
    lines += ["", "## Decisions"] + [f"- {item['event_id']}: {item['action']} / {item['status']}. {item['explanation']}" for item in report["decisions"]]
    lines += ["", "No AWS API was called and no resource action was executed.", ""]
    (folder / "finops-report.md").write_text("\n".join(lines), encoding="utf-8")
    (folder / "audit-events.jsonl").write_text("".join(json.dumps(item, sort_keys=True) + "\n" for item in audit), encoding="utf-8")

What does this report writer do?

• The JSON report preserves the complete structured result for automated inspection.
• The Markdown report turns measurements and decisions into a readable evidence summary.
• The JSON Lines export writes one append-only audit event per line.
• The final Markdown statement confirms that no AWS API or resource action exists in this simulator.

• In run(), find the final return line:

return report

Why change this return path?

The serialized report currently leaves the engine without being written to disk. The writer must run immediately before this return so the returned object matches the exported evidence.

• Replace that line with this report-writing call and return:

write_reports(workspace / "reports", report, store.audits())
    return report

What changes in run?

The engine now exports the report and its audit history before returning the same serialized object. This keeps the CLI output and saved evidence tied to one runtime result.

• Save cost_guardrail/engine.py.

Resetting the database is safe here because it only rebuilds local simulator state from deterministic fixtures.

Before you run the engine, which measurements do you expect to remain at zero?

• Generate a fresh evidence bundle by running:

python main.py run --reset

What should you see?

• Your terminal reports two anomalies.
• Precision and recall both show 1.00.
• Unsafe executions and duplicate remediations both remain 0.
• The blocked unsafe action count shows 1.
• The final line confirms that reports were written to reports/.

• Expand reports in the VS Code Explorer sidebar.
• Confirm that finops-report.json, finops-report.md, and audit-events.jsonl are listed.

That is the evidence boundary working. Your runtime state now becomes an inspectable artifact.

Reports missing?

Confirm that write_reports() appears before return report inside run().

Check that the first argument is workspace / "reports" so the files land beside your existing project folders.

Help me debug why my FinOps report files were not generated.

Add the evidence replay path

A clean run proves the safe policy works under normal input. The final engine also needs a controlled path that records hostile-policy rejection before replaying the same incidents through persistent state.

• In cost_guardrail/engine.py, add this function below run() by copying the code:

def demo(workspace):
    workspace = pathlib.Path(workspace)
    generate(workspace)
    run(workspace, reset=True)
    store = Store(workspace / "data" / "guardrail.db")
    try: load_policy(workspace / "data" / "unsafe-policy.json")
    except PolicyValidationError as error: store.add_audit(EVALUATION_DATE.isoformat(), "POLICY_REJECTED", "unsafe-policy", {"reason": str(error)})
    return run(workspace)

How does the demo preserve safety?

• The first run resets the local database before processing the two deterministic incidents.
• The unsafe policy passes through the same validator used by normal policy loading.
• A rejected policy becomes a POLICY_REJECTED audit event.
• The second run reconnects to SQLite and proves that both processed event IDs are suppressed.

• Save cost_guardrail/engine.py.
• Check that the new function did not break the existing engine by running:

pytest -q tests/test_guardrail.py

What does this check prove?

The existing pytest suite still passes after the demo path is added. This catches syntax or import regressions before the CLI changes.

Tests failing after demo was added?

Check that demo() starts after the complete run() function. A misplaced indentation level can accidentally nest it inside the earlier function.

Help me fix the demo function without changing the existing run behavior.

The CLI currently exposes generation and normal processing. Replacing main.py adds the demo command while preserving both existing paths.

• Replace the contents of main.py with this final CLI:

import argparse
import pathlib

from cost_guardrail.core import generate
from cost_guardrail.engine import demo, run

parser = argparse.ArgumentParser(description="Local AWS cost guardrail simulator")
parser.add_argument("command", choices=["generate", "run", "demo"])
parser.add_argument("--reset", action="store_true")
args = parser.parse_args()
workspace = pathlib.Path.cwd()
if args.command == "generate":
    print(f"Generated {generate(workspace)} deterministic cost records and 2 policy files.")
else:
    report = demo(workspace) if args.command == "demo" else run(workspace, reset=args.reset)
    print(f"Anomalies: {len(report['anomalies'])}")
    for key, value in report["measurements"].items(): print(f"{key}: {value}")
    print("Reports written to reports/.")

What does the final CLI orchestrate?

• The command parser now accepts generate, run, or demo.
• The normal run still forwards the optional reset flag to the engine.
• The demo path delegates hostile-policy rejection and replay handling to demo().
• Both processing paths print the measurements returned by the engine.

• Save main.py.

This reset only rebuilds local simulator state. It does not contact a cloud service or execute a resource action.

• Confirm the existing run path still generates evidence by running:

python main.py run --reset

What should remain unchanged?

You should still see two anomalies with perfect precision and recall. The CLI should finish by identifying reports/ as the evidence location.

Prove and document the final system

Generated evidence becomes credible when tests verify its content. The repository also needs documentation that separates this local simulator from production AWS billing behavior.

• In tests/test_guardrail.py, find this import:

from cost_guardrail.engine import run

Why update this import?

The final report test calls the controlled demo path directly. Importing that function keeps the test focused on the engine API.

• Replace the import with this line:

from cost_guardrail.engine import demo, run

What changes in the test module?

The test module now has access to both normal processing and hostile-input replay processing. Existing tests continue using run() unchanged.

• Add this final test at the bottom of tests/test_guardrail.py by copying the code:

def test_demo_reports_computed_metrics(tmp_path):
    report = demo(tmp_path)
    assert report["measurements"]["precision"] == "1.00"
    assert report["measurements"]["recall"] == "1.00"
    assert report["measurements"]["policy_rejections"] == 1
    assert json.loads((tmp_path / "reports" / "finops-report.json").read_text(encoding="utf-8")) == report
    assert (tmp_path / "reports" / "finops-report.md").exists()
    events = [json.loads(line) for line in (tmp_path / "reports" / "audit-events.jsonl").read_text(encoding="utf-8").splitlines()]
    assert any(event["event_type"] == "POLICY_REJECTED" for event in events)

What does the final test prove?

• Precision and recall come from runtime truth labels.
• The hostile policy produces exactly one recorded rejection.
• The exported JSON matches the report returned by the engine.
• The Markdown and JSON Lines evidence files both exist.
• The audit export contains a persisted policy-rejection event.

• Save tests/test_guardrail.py.
• Run the complete guardrail test file:

pytest -q tests/test_guardrail.py

What should the suite report?

All seven tests should pass. The new test proves that the report content comes from runtime processing instead of handwritten claims.

Report test failing?

Confirm that write_reports() receives store.audits() inside run().

Check that demo() records the rejection before its final replay run.

Help me trace why test_demo_reports_computed_metrics is failing.

The last repository artifact explains the architecture and its production limits. It also gives reviewers the exact commands needed to reproduce the evidence.

• Create README.md beside main.py with the file creation control in VS Code's Explorer sidebar.
• Paste this opening and architecture section into README.md.

# AWS Cost Guardrail Engine

A deterministic local FinOps simulator. It never calls AWS and never executes remediation.

## Architecture

```mermaid
flowchart TD
A[Simulated billing data] --> B[Cost ingestion]
B --> C[Anomaly detection]
B --> D[Budget forecasting]
C --> E[Policy engine]
D --> E
E --> F[Notify]
E --> G[Require approval]
E --> H[Simulate remediation]
F --> I[SQLite idempotency and audit]
G --> I
H --> I
I --> J[FinOps reports]
```

What does the architecture show?

The diagram traces billing fixtures through analytics and policy selection. Every decision converges on SQLite before the report layer exports evidence.

• Save README.md.
• Confirm that the file begins with the project title and a Mermaid flowchart.

• Add a blank line below the Mermaid block.
• Append this setup section:

## Setup

macOS:

```bash
python3 --version
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```

Windows PowerShell:

```powershell
py -3 --version
py -3 -m venv .venv
.venv\Scripts\Activate.ps1
pip install -r requirements.txt
```

If PowerShell blocks activation:

```powershell
Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser
```

In VS Code, install the Microsoft Python extension, run `Python: Select Interpreter`, and choose `.venv`.

Why document both setup paths?

The repository uses the same project-local environment on macOS and Windows. The platform-specific activation commands make that environment reproducible for another developer.

• Save README.md.
• Confirm that the setup section includes separate macOS and Windows commands.

• Add a blank line below the interpreter instruction.
• Append the run commands and AWS mapping:

## Run

```text
python main.py generate
pytest -q
python main.py run --reset
python main.py demo
```

## AWS mapping and limitations

| Local component | Future AWS mapping |
| --- | --- |
| JSON cost loader | AWS Cost Explorer or a cost and usage export |
| Threshold detector | AWS Cost Anomaly Detection |
| Forecast and budgets | AWS Cost Explorer and AWS Budgets |
| CLI handler | AWS Lambda |
| Incident routing | Amazon EventBridge |
| Notification decision | Amazon Simple Notification Service (SNS) |
| SQLite idempotency | A production durable conditional-write store |

Why include a future AWS mapping?

The table connects each local responsibility to a possible production service. It preserves the educational simulator boundary while showing a credible integration path.

• Save README.md.
• Confirm that the run section lists all four CLI workflows.

• Add a blank line below the mapping table.
• Append the limitations and evidence sections:

AWS Cost Anomaly Detection uses machine learning, runs about three times daily after billing processing, and can inherit up to 24 hours of Cost Explorer delay. This simulator uses immediate fixtures and transparent thresholds. Its elapsed-day forecast has no prediction interval. It omits credits, refunds, taxes, discounts, currencies, account hierarchies, IAM, retries, throttling, concurrency, late data, and partial failures. EventBridge delivery behavior and SNS delivery are not simulated.

A production design could ingest Cost Explorer data on a controlled cadence, receive anomaly events through EventBridge, evaluate policy in Lambda, notify through SNS, and use a durable idempotency store. It would need least-privilege IAM, schema validation, retries, dead-letter handling, approvals, observability, and kill switches. Cost Explorer API calls are billed, so that integration would not inherit this project's guaranteed $0 cost.

## Evidence

The CLI generates `finops-report.json`, `finops-report.md`, `audit-events.jsonl`, and `guardrail.db`. Metrics are computed from truth labels, audit rows, and SQLite state. Tests cover missing history, zero baselines, insignificant spikes, leap-year forecasting, zero budgets, malformed policies, unknown actions, unsafe policies, production safety, replay, persistence, and reporting.

Resume description: Built a deterministic Python FinOps guardrail engine that detects material AWS spending anomalies, forecasts budget overruns with Decimal arithmetic, validates fail-safe JSON policies, persists idempotency and audit history in SQLite, and measures unsafe executions and duplicate remediations through pytest and generated reports.

What makes these claims credible?

The limitations separate transparent local thresholds from AWS billing systems. The evidence section points each claim back to generated files, persisted state, or automated tests.

• Save README.md.
• Confirm that the final section names all three reports and the SQLite database.

README structure look incomplete?

Check that each section starts on a new line after the previous paragraph, code block, or table. Missing blank lines can prevent Markdown renderers from recognizing headings and fences.

Help me compare my README structure with the expected architecture, setup, run, limitations, and evidence sections.

✔️ Awesome, I've got everything!

Your engine, CLI, tests, and README now form the complete local simulator. Double check that every changed file is saved.

ⓧ I'd like to double check the full code

Compare your changed files with these final references.

import datetime
import decimal
import json
import pathlib
import sqlite3

from cost_guardrail.core import EVALUATION_DATE, PolicyValidationError, anomalies, forecasts, generate, load_data, load_policy, select_rule


class Store:
    def __init__(self, path):
        self.path = pathlib.Path(path)
        with sqlite3.connect(str(self.path)) as connection:
            connection.executescript("""
            CREATE TABLE IF NOT EXISTS processed(event_id TEXT PRIMARY KEY, decision TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS simulations(event_id TEXT PRIMARY KEY, remediation TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS audit(sequence INTEGER PRIMARY KEY AUTOINCREMENT, evaluation_date TEXT NOT NULL, event_type TEXT NOT NULL, event_id TEXT NOT NULL, payload TEXT NOT NULL);
            """)

    def add_audit(self, day, kind, event_id, payload):
        with sqlite3.connect(str(self.path)) as connection:
            connection.execute("INSERT INTO audit(evaluation_date,event_type,event_id,payload) VALUES(?,?,?,?)", (day, kind, event_id, json.dumps(payload, sort_keys=True)))

    def claim(self, day, decision):
        with sqlite3.connect(str(self.path)) as connection:
            created = connection.execute("INSERT OR IGNORE INTO processed VALUES(?,?)", (decision["event_id"], json.dumps(decision, sort_keys=True))).rowcount == 1
        self.add_audit(day, "DECISION_RECORDED" if created else "DUPLICATE_SUPPRESSED", decision["event_id"], decision)
        return created

    def simulate(self, day, event_id, remediation):
        with sqlite3.connect(str(self.path)) as connection:
            created = connection.execute("INSERT OR IGNORE INTO simulations VALUES(?,?)", (event_id, remediation)).rowcount == 1
        if created:
            self.add_audit(day, "REMEDIATION_SIMULATED", event_id, {"remediation": remediation})

    def list_json(self, table, column):
        with sqlite3.connect(str(self.path)) as connection:
            return [json.loads(row[0]) for row in connection.execute(f"SELECT {column} FROM {table} ORDER BY 1")]

    def audits(self):
        with sqlite3.connect(str(self.path)) as connection:
            rows = connection.execute("SELECT sequence,evaluation_date,event_type,event_id,payload FROM audit ORDER BY sequence").fetchall()
        return [{"sequence": row[0], "evaluation_date": row[1], "event_type": row[2], "event_id": row[3], "payload": json.loads(row[4])} for row in rows]

    def count(self, kind):
        with sqlite3.connect(str(self.path)) as connection:
            return connection.execute("SELECT COUNT(*) FROM audit WHERE event_type=?", (kind,)).fetchone()[0]

    def duplicate_remediations(self):
        with sqlite3.connect(str(self.path)) as connection:
            return connection.execute("SELECT COUNT(*) FROM (SELECT event_id FROM simulations GROUP BY event_id HAVING COUNT(*)>1)").fetchone()[0]


def serial(value):
    if isinstance(value, decimal.Decimal): return format(value, "f")
    if isinstance(value, datetime.date): return value.isoformat()
    if isinstance(value, list): return [serial(item) for item in value]
    if isinstance(value, dict): return {key: serial(item) for key, item in value.items()}
    return value


def write_reports(folder, report, audit):
    folder.mkdir(exist_ok=True)
    (folder / "finops-report.json").write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    lines = ["# FinOps Guardrail Report", "", f"Evaluation date: {report['evaluation_date']}", "", "## Measurements"]
    lines += [f"- {key}: {value}" for key, value in report["measurements"].items()]
    lines += ["", "## Anomalies"] + [f"- {item['environment']} / {item['service']}: ${item['actual']} vs ${item['baseline']}" for item in report["anomalies"]]
    lines += ["", "## Decisions"] + [f"- {item['event_id']}: {item['action']} / {item['status']}. {item['explanation']}" for item in report["decisions"]]
    lines += ["", "No AWS API was called and no resource action was executed.", ""]
    (folder / "finops-report.md").write_text("\n".join(lines), encoding="utf-8")
    (folder / "audit-events.jsonl").write_text("".join(json.dumps(item, sort_keys=True) + "\n" for item in audit), encoding="utf-8")


def run(workspace, day=EVALUATION_DATE, reset=False):
    workspace = pathlib.Path(workspace)
    if not (workspace / "data" / "costs.json").exists(): generate(workspace)
    database = workspace / "data" / "guardrail.db"
    if reset: database.unlink(missing_ok=True)
    rows, budgets = load_data(workspace / "data" / "costs.json")
    policy = load_policy(workspace / "data" / "policy.json")
    found, projected, store = anomalies(rows, day), forecasts(rows, budgets, day), Store(database)
    for incident in found:
        rule = select_rule(policy, incident)
        action, remediation = rule["action"], rule["remediation"]
        explanation = f"Rule {rule['name']} selected {action}."
        if incident["critical"] and incident["environment"] == "production" and action == "SIMULATE_REMEDIATION":
            action, explanation = "REQUIRE_APPROVAL", "Defense in depth blocked critical production remediation."
        status = {"NOTIFY": "NOTIFIED", "REQUIRE_APPROVAL": "BLOCKED", "SIMULATE_REMEDIATION": "SIMULATED"}[action]
        explanation += " No real AWS action exists in this simulator."
        decision = {"event_id": incident["event_id"], "environment": incident["environment"], "service": incident["service"], "action": action, "remediation": remediation, "status": status, "explanation": explanation}
        if store.claim(day.isoformat(), decision) and action == "SIMULATE_REMEDIATION": store.simulate(day.isoformat(), incident["event_id"], remediation)
    truth = {(row["environment"], row["service"]) for row in rows if row["date"] == day and row["expected_anomaly"]}
    detected = {(item["environment"], item["service"]) for item in found}
    tp, fp, fn = len(truth & detected), len(detected - truth), len(truth - detected)
    decisions = store.list_json("processed", "decision")
    metrics = {
        "precision": decimal.Decimal(tp / (tp + fp) if tp + fp else 0).quantize(decimal.Decimal("0.01")),
        "recall": decimal.Decimal(tp / (tp + fn) if tp + fn else 0).quantize(decimal.Decimal("0.01")),
        "unsafe_actions_executed": store.count("UNSAFE_ACTION_EXECUTED"),
        "duplicate_remediations": store.duplicate_remediations(),
        "duplicate_suppressions": store.count("DUPLICATE_SUPPRESSED"),
        "blocked_unsafe_actions": sum(item["status"] == "BLOCKED" for item in decisions),
        "policy_rejections": store.count("POLICY_REJECTED"),
    }
    report = serial({"evaluation_date": day, "anomalies": found, "forecasts": projected, "decisions": decisions, "measurements": metrics})
    write_reports(workspace / "reports", report, store.audits())
    return report


def demo(workspace):
    workspace = pathlib.Path(workspace)
    generate(workspace)
    run(workspace, reset=True)
    store = Store(workspace / "data" / "guardrail.db")
    try: load_policy(workspace / "data" / "unsafe-policy.json")
    except PolicyValidationError as error: store.add_audit(EVALUATION_DATE.isoformat(), "POLICY_REJECTED", "unsafe-policy", {"reason": str(error)})
    return run(workspace)

import argparse
import pathlib

from cost_guardrail.core import generate
from cost_guardrail.engine import demo, run

parser = argparse.ArgumentParser(description="Local AWS cost guardrail simulator")
parser.add_argument("command", choices=["generate", "run", "demo"])
parser.add_argument("--reset", action="store_true")
args = parser.parse_args()
workspace = pathlib.Path.cwd()
if args.command == "generate":
    print(f"Generated {generate(workspace)} deterministic cost records and 2 policy files.")
else:
    report = demo(workspace) if args.command == "demo" else run(workspace, reset=args.reset)
    print(f"Anomalies: {len(report['anomalies'])}")
    for key, value in report["measurements"].items(): print(f"{key}: {value}")
    print("Reports written to reports/.")

import datetime
import decimal
import json

import pytest

from cost_guardrail.core import EVALUATION_DATE, PolicyValidationError, anomalies, forecasts, generate, load_policy, money, validate_policy
from cost_guardrail.engine import demo, run


def row(day, amount):
    return {"date": day, "environment": "development", "service": "Amazon EC2", "amount": money(amount), "critical": False, "expected_anomaly": False}


def test_anomaly_thresholds_and_missing_history():
    day = EVALUATION_DATE
    assert anomalies([row(day, "100")], day) == []
    assert anomalies([row(day.replace(day=14), "10"), row(day, "45")], day)[0]["increase"] == decimal.Decimal("35.00")
    assert anomalies([row(day.replace(day=14), "1"), row(day, "2")], day) == []


def test_zero_baseline_is_supported():
    found = anomalies([row(EVALUATION_DATE.replace(day=14), "0"), row(EVALUATION_DATE, "25")], EVALUATION_DATE)
    assert len(found) == 1 and found[0]["ratio"] is None


def test_leap_year_forecast_and_zero_budget():
    day = datetime.date(2024, 2, 10)
    rows = [row(day.replace(day=value), "10") for value in range(1, 11)]
    result = forecasts(rows, {"development": money("0")}, day)[0]
    assert result["projected_monthly"] == decimal.Decimal("290.00")
    assert result["overrun"] and result["utilization_percent"] is None


def test_malformed_and_unknown_policies_are_rejected():
    with pytest.raises(PolicyValidationError): validate_policy({"version": 1, "rules": []})
    policy = {"version": 1, "rules": [{"name": "bad", "priority": 1, "match": {"environment": "*", "service": "*", "critical": "*"}, "action": "DELETE", "remediation": "NONE", "disruptive": False}]}
    with pytest.raises(PolicyValidationError, match="unknown action"): validate_policy(policy)


def test_unsafe_policy_is_rejected(tmp_path):
    generate(tmp_path)
    with pytest.raises(PolicyValidationError, match="non-disruptive"): load_policy(tmp_path / "data" / "unsafe-policy.json")


def test_production_safety_and_persistent_idempotency(tmp_path):
    generate(tmp_path)
    first = run(tmp_path, reset=True)
    assert {item["service"]: item["status"] for item in first["decisions"]} == {"Amazon EC2": "SIMULATED", "Amazon RDS": "BLOCKED"}
    replay = run(tmp_path)
    assert replay["measurements"]["unsafe_actions_executed"] == 0
    assert replay["measurements"]["duplicate_remediations"] == 0
    assert replay["measurements"]["duplicate_suppressions"] == 2


def test_demo_reports_computed_metrics(tmp_path):
    report = demo(tmp_path)
    assert report["measurements"]["precision"] == "1.00"
    assert report["measurements"]["recall"] == "1.00"
    assert report["measurements"]["policy_rejections"] == 1
    assert json.loads((tmp_path / "reports" / "finops-report.json").read_text(encoding="utf-8")) == report
    assert (tmp_path / "reports" / "finops-report.md").exists()
    events = [json.loads(line) for line in (tmp_path / "reports" / "audit-events.jsonl").read_text(encoding="utf-8").splitlines()]
    assert any(event["event_type"] == "POLICY_REJECTED" for event in events)

# AWS Cost Guardrail Engine

A deterministic local FinOps simulator. It never calls AWS and never executes remediation.

## Architecture

```mermaid
flowchart TD
A[Simulated billing data] --> B[Cost ingestion]
B --> C[Anomaly detection]
B --> D[Budget forecasting]
C --> E[Policy engine]
D --> E
E --> F[Notify]
E --> G[Require approval]
E --> H[Simulate remediation]
F --> I[SQLite idempotency and audit]
G --> I
H --> I
I --> J[FinOps reports]
```

## Setup

macOS:

```bash
python3 --version
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```

Windows PowerShell:

```powershell
py -3 --version
py -3 -m venv .venv
.venv\Scripts\Activate.ps1
pip install -r requirements.txt
```

If PowerShell blocks activation:

```powershell
Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser
```

In VS Code, install the Microsoft Python extension, run `Python: Select Interpreter`, and choose `.venv`.

## Run

```text
python main.py generate
pytest -q
python main.py run --reset
python main.py demo
```

## AWS mapping and limitations

| Local component | Future AWS mapping |
| --- | --- |
| JSON cost loader | AWS Cost Explorer or a cost and usage export |
| Threshold detector | AWS Cost Anomaly Detection |
| Forecast and budgets | AWS Cost Explorer and AWS Budgets |
| CLI handler | AWS Lambda |
| Incident routing | Amazon EventBridge |
| Notification decision | Amazon Simple Notification Service (SNS) |
| SQLite idempotency | A production durable conditional-write store |

AWS Cost Anomaly Detection uses machine learning, runs about three times daily after billing processing, and can inherit up to 24 hours of Cost Explorer delay. This simulator uses immediate fixtures and transparent thresholds. Its elapsed-day forecast has no prediction interval. It omits credits, refunds, taxes, discounts, currencies, account hierarchies, IAM, retries, throttling, concurrency, late data, and partial failures. EventBridge delivery behavior and SNS delivery are not simulated.

A production design could ingest Cost Explorer data on a controlled cadence, receive anomaly events through EventBridge, evaluate policy in Lambda, notify through SNS, and use a durable idempotency store. It would need least-privilege IAM, schema validation, retries, dead-letter handling, approvals, observability, and kill switches. Cost Explorer API calls are billed, so that integration would not inherit this project's guaranteed $0 cost.

## Evidence

The CLI generates `finops-report.json`, `finops-report.md`, `audit-events.jsonl`, and `guardrail.db`. Metrics are computed from truth labels, audit rows, and SQLite state. Tests cover missing history, zero baselines, insignificant spikes, leap-year forecasting, zero budgets, malformed policies, unknown actions, unsafe policies, production safety, replay, persistence, and reporting.

Resume description: Built a deterministic Python FinOps guardrail engine that detects material AWS spending anomalies, forecasts budget overruns with Decimal arithmetic, validates fail-safe JSON policies, persists idempotency and audit history in SQLite, and measures unsafe executions and duplicate remediations through pytest and generated reports.

Before you run the final checks, do you expect any unsafe execution or duplicate remediation to appear?

• Regenerate the evidence and run the complete test suite with these commands:

python main.py run --reset
pytest -q

What does final success look like?

• The CLI reports two material anomalies.
• Precision and recall both remain 1.00.
• Unsafe executions and duplicate remediations both remain 0.
• All seven tests pass.
• The three report files remain available under reports/.

• Open reports/finops-report.md from the VS Code Explorer sidebar.
• Confirm that the report lists measurements, anomalies, decisions, and the no-action safety statement.
• Confirm that README.md contains the architecture and limitations.

You now have a complete evidence bundle that proves the engine detects cost risk without executing an unsafe action.

Secret mission

Stop a Runaway Bill Without Taking Down Production

Replay the runaway-bill incidents against persisted state while an unsafe production policy challenges the guardrail. Prove that development remediation stays simulated while critical production remains blocked across repeated processing.

Clean Up Your Resources

Clean Up Your Resources

No AWS API was called, so this project has no ongoing cloud costs. Choose whether to keep the evidence, pause your environment, or delete the generated runtime artifacts.

Resources you used:

• The .venv project environment with pytest 9.2.0.
• The data/guardrail.db SQLite database with durable decisions, simulation state, and audit history.
• The reports/finops-report.json, reports/finops-report.md, and reports/audit-events.jsonl evidence files.
• The .pytest_cache test cache.

Keep everything running

No action needed. Choose this if you want to preserve the replay evidence or rerun the simulator.

• Keep .venv so the pinned test environment stays available.
• Retain data/guardrail.db so the restart-safe duplicate suppression remains inspectable.
• Preserve the three files in reports so the computed safety metrics stay available.

Pause - I'll come back to this later

An activated virtual environment only changes the current shell session. Every project file remains on disk after you leave it.

• Return to the terminal session from earlier.
• Leave the active virtual environment by running this command:

deactivate

The deactivate command returns your shell to its normal Python environment. Your database, reports, fixtures, policies, source code, and tests remain unchanged.

Delete - I don't want to use this again

Removing generated evidence can feel final. The cleanup below preserves your source code and deterministic fixtures, so you can regenerate every deleted artifact.

• Choose the tab for your operating system to remove the local generated artifacts.
• Skip AWS cleanup because this project created no AWS resources.

macOS

• Return to the project terminal from earlier.
• If your prompt begins with (.venv), leave the virtual environment by running:

deactivate

This returns the terminal to your normal shell environment before the environment folder is removed.

• Delete the generated environment, database, test cache, and reports by running these commands:

rm -rf .venv
rm -f data/guardrail.db
rm -rf .pytest_cache
rm -rf reports

These commands remove only the generated environment, persistent state, test cache, and report files. Your source code, JSON fixtures, policies, tests, and README stay in place.

• In the VS Code Explorer sidebar, confirm that .venv, data/guardrail.db, .pytest_cache, and reports are gone.

Windows

• Return to the project terminal from earlier.
• If your prompt begins with (.venv), leave the virtual environment by running:

deactivate

This returns PowerShell to your normal environment before the environment folder is removed.

• Delete the generated environment, database, test cache, and reports by running these commands:

Remove-Item -Recurse -Force .venv, .pytest_cache, reports
Remove-Item -Force data\guardrail.db

These commands remove only the generated environment, persistent state, test cache, and report files. Your source code, JSON fixtures, policies, tests, and README stay in place.

• In the VS Code Explorer sidebar, confirm that .venv, data/guardrail.db, .pytest_cache, and reports are gone.

Nice Work!

Nice Work!

Outstanding work! Your local FinOps simulator now produces deterministic evidence that cost automation stays accurate across restarts. Its policy boundaries remain explainable under replayed incidents and hostile inputs.

Here's what you learned:

• Built deterministic anomaly detection that finds material spending spikes across labeled billing records. Added calendar-correct forecasting with Decimal arithmetic for inspectable month-end projections.
• Reproduced a ratio-only false positive with pytest. Corrected it with a $20 materiality gate while preserving missing-history and zero-baseline behavior.
• Enforced explainable production safety through validated JSON policies. Persisted event claims with SQLite primary keys. Generated computed reports showing precision 1.00. The reports show recall 1.00. They record zero unsafe executions. They record zero duplicate remediations.
• Completed the optional Secret Mission by replaying both anomaly events through a new SQLite connection. The development Amazon EC2 event retains exactly one simulated remediation. The critical production Amazon RDS action stays blocked. The unsafe policy is rejected. Duplicate event IDs are suppressed without repeating remediation.

Ready to quiz yourself?