# Project export: SelfAudit

This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.

## Project metadata

- Hackathon: UC Berkeley AI Hackathon 2026
- Tagline: Catches AI agents that spend without progress
- Devpost: https://devpost.com/software/selfaudit
- GitHub: https://github.com/apoorva-khandelwal/selfaudit
- Video: https://www.youtube.com/embed/A98pr6kXMXU?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 4 GitHub contributor(s) — nandiniippili (38 commits), Claude Sonnet 4.6 (37 commits), apoorva (12 commits), apoorva-khandelwal (4 commits)

## Devpost submission (written by the team)

### Inspiration

We kept hearing the same story: an AI agent gets stuck not crashed, just looping on a failing step and nobody notices because each API call looks normal. The FinOps community now has a name for it: "agentic resource exhaustion." The problem: cost dashboards tell you how much an agent spent, not whether it bought anything an agent can stay under budget and still be a total waste. What It Does SelfAudit tracks cost and progress in parallel, measuring progress only by objective signals completed steps, retries, elapsed time vs. baseline never by letting the agent self-report. When cost climbs but progress stalls, it flags the issue with recommendations (e.g. cheaper models) and waits for human approval in this human in the loop process, rather than auto-killing the agent; ambiguous cases go to a separate review queue instead of triggering false alarms. Past alerts are stored in Redis as embeddings so new flags get compared against similar prior cases, and dashboard state syncs across machines via Redis. What We Built The core is a small Python SDK a Watcher class you wrap around an existing agent loop with a context manager (with watcher.trace(agent_id, action) as t:), so dropping it into a real project takes about three lines. Behind that sits a Flask dashboard streaming live updates over Server-Sent Events, five simulated agent behaviors (healthy, stuck, slow, fast-failing, and an "ambiguous" agent that makes some progress before stalling) to demo every code path, and two Redis integrations: one for cross-session alert memory using simple vector similarity, and one for cross-machine state sync via pub/sub. Challenges One of our challenges was the dashboard's timer freezing it looked like agents had stalled, but really the screen just stopped updating because our live updates only fired when something happened, not on a steady clock. We fixed it by sending a regular "still alive" signal even when nothing new was happening. What We Learned The most useful design decision we made was the one that removed complexity: refusing to let the system grade its own value, and instead anchoring everything to numbers that don't require judgment completed steps, retries, elapsed time. It made the whole system more defensible, easier to demo, and harder to argue with.

## README (from the GitHub repository)

# selfaudit

**Catches AI agents that spend without progress.**

A monitoring layer for AI agents that tracks cost against actual progress — not just total spend — and flags the moment an agent is burning money without producing anything. Built around a real, widely-cited industry incident: four agents stuck in a retry loop for 11 days, $47,000 in API charges before anyone noticed.

## The problem

Every existing cost-monitoring tool answers one question: *how much is this agent spending?* None of them answer the more important one *is that spend buying anything?* An agent can stay well under budget every single time and still be a complete waste, because the failure mode isn't "too expensive," it's "spending with nothing to show for it." That's invisible to a tool that only watches dollars.

## What it does

- Tracks **cost vs. progress** for any agent, using only objective signals: distinct successful steps, retry counts on failed actions, elapsed time against an expected baseline. The agent never grades its own work — that's an intentional design choice (see [Design decisions](#design-decisions)).
- When cost climbs while progress stays flat, it **flags the agent and surfaces a recommendation** — including cheaper-model alternatives with published pricing — instead of auto-killing anything. A human decides whether to pause, re-run, or escalate.
- Ambiguous cases (some progress, but elevated retries or cost) go to a **review queue** instead of firing a false alarm, so the system distinguishes "definitely stuck" from "worth a second look."
- Every alert is **embedded and stored in Redis**; future alerts are checked against similar past cases instead of being judged cold each time.
- Dashboard state is **pushed through Redis pub/sub**, so a second dashboard instance on a different machine can mirror the same live data with no direct connection to the process running the agents.

## Demo

```bash
python main.py
```
Opens a live dashboard at `http://localhost:5050` and runs five simulated agents (healthy, stuck, slow, fast-failing, and an ambiguous one that makes partial progress before stalling) to exercise every code path.

To run a second dashboard reading purely from Redis (useful to demo the cross-machine sync):
```bash
python dashboard.py 5051
```

## Setup

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

Create a `.env` file (never committed — see `.env.example` for the required keys):
```
REDIS_HOST=...
REDIS_PORT=...
REDIS_USERNAME=...
REDIS_PASSWORD=...
ANTHROPIC_API_KEY=...
```

## Architecture

```
sdk.py        — Watcher class: drop-in SDK (`with watcher.trace(agent_id, action) as t:`)
dashboard.py  — Flask app, SSE stream, live UI
memory.py     — Redis-backed alert memory (embed → store → retrieve similar past alerts)
redis_store.py— Cross-machine dashboard state via Redis snapshot + pub/sub
models.py     — Static model cost/spec lookup table, used for "cheaper alternative" recommendations
main.py       — Demo runner: simulates 5 agent behaviors against the Watcher
```

## Design decisions

**Why doesn't the agent judge its own work?** We considered it and rejected it. An agent self-reporting "I'm doing fine" has the same structural problem as a student grading their own exam — no real incentive to flag its own failure. Instead, every signal SelfAudit uses is something you can count, not something you have to ask an LLM to judge: completed steps, retry counts, elapsed time.

**Why human approval instead of auto-retry?** A stuck agent re-run blindly just repeats the same failure and burns more money — the exact problem we're trying to prevent. SelfAudit surfaces a recommendation; a human (or your own approval logic) decides.

## Built with

Python · Flask · Server-Sent Events · Redis (alert memory + cross-machine pub/sub) · Anthropic API · vanilla JavaScript

## Known limitations

Being upfront about scope, since a hackathon judge will appreciate honesty over a polished overclaim:
- Embeddings used for Redis similarity search are a lightweight character-trigram hash, not a trained semantic model — "similar" means textually similar, not deeply semantic.
- An earlier LLM-based peer-review prototype exists in the repo but isn't wired into the live path; the current design uses a human review queue instead (see [Design decisions](#design-decisions)).
- Agent behaviors are simulated for demo purposes, not connected to a live production agent — integration with a real agent loop is a 3-line drop-in (see `examples.py`).


## Detected evidence (automated analysis)

Indexed codebase: 9 recognized source files, 106 KB.
- Python (language) — detected in the code
- Flask (technology) — claimed on Devpost, not found in the code
- Redis (technology) — claimed on Devpost, not found in the code
- AI coding agent: Claude Code — evidence: commit authorship or trailers

## Codebase structure (from repository index)

### Files (10 of 10)

```
__init__.py
.gitignore
dashboard.py
examples.py
main.py
memory.py
models.py
README.md
redis_store.py
sdk.py
```

### Dependencies

No dependency index available.

### Recent commits (newest first)

- Update README.md
- Update README.md
- Add model tradeoffs, FLAG/ALERT tuning, Redis session clearing
- Add Claude-backed watcher, proper Redis schema, fast-fail fix
- fixed the timing and added logo
- changing the files on my side to match
- Merge branch 'main' of https://github.com/apoorva-khandelwal/selfaudit
- Redesign UI: Inter font, softer color palette, less verbose copy
- updated the dashboard with more Redis implementation
- Update README.md
- Remove old/unused files: agents.py, watcher.py, test_redis.py
- Remove peer_judge.py and tracing.py — not part of core design
- updated ui features oft he dashboard
- Fix global thresholds 500 — use _retry_threshold not retry_threshold
- Resolve dashboard.py merge conflicts — take teammate changes
- Remove EST. SAVED stat from header
- re-added the peer judge
- changed it so when i switch tabs, notif is created
- Remove Phoenix tracing (not needed for demo)
- Fix undo for escalate and clear_flag actions

## Key source files (fetched from GitHub, selected and truncated for size)

### main.py

```python
"""
SelfAudit — example runner.

Usage:
    python3 main.py               # 4 agents (default)
    python3 main.py --agents 12   # stress test with 12 agents
    python3 main.py --no-phoenix  # skip Phoenix, just the dashboard
"""

import time
import random
import threading
import argparse
from sdk import Watcher


# ── agent behaviors ────────────────────────────────────────────────────────────

def run_healthy(watcher: Watcher, agent_id: str):
    actions = ["fetch_data", "parse_response", "summarize", "write_output"]
    for i, action in enumerate(actions):
        time.sleep(random.uniform(1.0, 2.0))
        with watcher.trace(agent_id, action, model="claude-haiku-4-5") as t:
            t.success(cost_usd=round(random.uniform(0.002, 0.008), 4),
                      completed=(i == len(actions) - 1))


def run_stuck(watcher: Watcher, agent_id: str):
    for _ in range(20):
        time.sleep(random.uniform(0.8, 1.2))
        with watcher.trace(agent_id, "call_external_api", model="claude-opus-4-8") as t:
            t.fail(cost_usd=round(random.uniform(0.015, 0.025), 4))


def run_slow(watcher: Watcher, agent_id: str):
    actions = ["load_dataset", "preprocess", "run_inference", "validate", "format_output"]
    for i, action in enumerate(actions):
        time.sleep(random.uniform(1.8, 2.5))
        with watcher.trace(agent_id, action, model="claude-sonnet-4-6") as t:
            t.success(cost_usd=round(random.uniform(0.005, 0.012), 4),
                      completed=(i == len(actions) - 1))


def run_fast_fail(watcher: Watcher, agent_id: str):
    for _ in range(5):
        time.sleep(0.3)
        with watcher.trace(agent_id, "authenticate", model="claude-sonnet-4-6") as t:
            t.fail(cost_usd=round(random.uniform(0.001, 0.003), 4))


def run_intermittent(watcher: Watcher, agent_id: str):
    """Makes some progress then gets stuck — tests the ambiguous zone."""
    for action in ["connect", "fetch"]:
        time.sleep(1.0)
        with watcher.trace(agent_id, action, model="claude-opus-4-8") as t:
            t.success(cost_usd=round(random.uniform(0.003, 0.007), 4))
    for _ in range(15):
        time.sleep(1.0)
        with watcher.trace(agent_id, "process_chunk", model="claude-opus-4-8") as t:
            t.fail(cost_usd=round(random.uniform(0.010, 0.020), 4))


# ── orchestrator ───────────────────────────────────────────────────────────────

BEHAVIORS = [run_healthy, run_stuck, run_slow, run_fast_fail, run_intermittent]
BEHAVIOR_NAMES = ["healthy", "stuck", "slow", "fast-fail", "intermittent"]


def build_agent_plan(n: int):
    """
    For n agents, assign behaviors so there's always at least one stuck agent
    and a mix of others.
    """
    plan = []
    # first 4 are always the canonical set
    fixed = [
        (run_healthy,      "healthy"),
        (run_stuck,        "stuck"),
        (run_slow,         "slow"),
        (run_fast_fail,    "fast-fail"),
        (run_intermittent, "intermittent"),
    ]
    for fn, name in fixed[:min(n, 5)]:
        idx = len(plan) + 1
        plan.append((fn, f"agent-{idx} ({name})"))

    # extra agents get random behaviors
    for i in range(len(plan), n):
        fn = random.choice(BEHAVIORS)
        name = BEHAVIOR_NAMES[BEHAVIORS.index(fn)]
        plan.append((fn, f"agent-{i+1} ({name})"))

    return plan


def main():
    parser = argparse.ArgumentParser(description="SelfAudit demo runner")
    parser.add_argument("--agents",     type=int, default=5,
                        help="number of agents to simulate (default: 5)")
    parser.add_argument("--no-phoenix", action="store_true",
                        help="skip Phoenix trace UI")
    args = parser.parse_args()

    watcher = Watcher(
        task_description="Process a dataset and produce a structured summary report.",
    )
    watcher.start_dashboard(port=5050)

    plan = build_agent_plan(args.agents)
    print(f"SelfAudit — watching {len(plan)} agents\n")

    threads = [
        threading.Thread(target=fn, args=(watcher, agent_id), daemon=True)
        for fn, agent_id in plan
    ]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    watcher.summary()
    print("Dashboard → http://localhost:5050  |  Ctrl+C to exit")
    try:
        threading.Event().wait()
    except KeyboardInterrupt:
        pass


if __name__ == "__main__":
    main()
```

### __init__.py

```python
from sdk import Watcher, Alert

```

### examples.py

```python
"""
Drop SelfAudit into your existing agent in 3 lines.

Copy the pattern that matches your setup.
"""

# ─────────────────────────────────────────────────────────────
# PATTERN 1 — Anthropic SDK (cost extracted automatically)
# ─────────────────────────────────────────────────────────────
"""
import anthropic
from selfaudit.sdk import Watcher

client  = anthropic.Anthropic()
watcher = Watcher()
watcher.start_dashboard()   # http://localhost:5050

for step in my_agent_loop():
    with watcher.trace("my-agent", action=step.name) as t:
        response = client.messages.create(
            model="claude-opus-4-8",
            max_tokens=1024,
            messages=[{"role": "user", "content": step.prompt}],
        )
        t.success_from_anthropic(response)          # cost extracted from response.usage
        # or mark the final step as done:
        t.success_from_anthropic(response, completed=True)
"""


# ─────────────────────────────────────────────────────────────
# PATTERN 2 — Any LLM / manual cost
# ─────────────────────────────────────────────────────────────
"""
from selfaudit.sdk import Watcher

watcher = Watcher()
watcher.start_dashboard()

with watcher.trace("my-agent", action="summarize") as t:
    result = my_llm.call(prompt)
    if result.ok:
        t.success(cost_usd=result.cost)
    else:
        t.fail(cost_usd=result.cost)   # watcher starts counting retries
"""


# ─────────────────────────────────────────────────────────────
# PATTERN 3 — Wrap an existing retry loop
# ─────────────────────────────────────────────────────────────
"""
from selfaudit.sdk import Watcher

watcher = Watcher(
    retry_threshold=3,       # alert after 3 failed retries
    cost_threshold=0.10,     # alert if $0.10 spent with no progress
    time_threshold=60.0,     # alert if 60s elapsed with no progress
)
watcher.start_dashboard()

for attempt in range(MAX_RETRIES):
    with watcher.trace("my-agent", action="call_api") as t:
        try:
            result = call_external_api()
            t.success(cost_usd=0.002)
            break
        except Exception:
            t.fail(cost_usd=0.002)
            # SelfAudit fires an alert and shows cheaper model alternatives
            # after retry_threshold failures — you decide whether to stop
"""


# ─────────────────────────────────────────────────────────────
# PATTERN 4 — Custom alert handler (e.g. Slack, PagerDuty)
# ─────────────────────────────────────────────────────────────
"""
from selfaudit.sdk import Watcher

def my_alert_handler(alert):
    slack.send(f"ALERT: {alert.agent_id} — {alert.reason} (${alert.cost_usd:.2f} spent)")
    # alert.recommendation includes model tradeoff table

watcher = Watcher(on_alert=my_alert_handler)
watcher.start_dashboard()
"""
```

### memory.py

```python
"""
Alert memory layer — stores past alerts in Redis and retrieves similar ones
when a new alert fires. Gives devs context: "this happened before, here's what occurred."

Falls back silently if Redis is unavailable — memory is a nice-to-have, not required.

Requires REDIS_HOST / REDIS_PORT / REDIS_USERNAME / REDIS_PASSWORD in environment (or .env).
"""

import os
import time
import hashlib
import numpy as np

_EMBED_DIM  = 128
_KEY_PREFIX = "selfaudit:alert:"
_INDEX_KEY  = "selfaudit:alert:index"


def _embed(text: str) -> np.ndarray:
    """Character-trigram hash embedding — fast, deterministic, no ML model needed."""
    vec = np.zeros(_EMBED_DIM, dtype=np.float32)
    text = text.lower()
    for i in range(len(text) - 2):
        h = int(hashlib.md5(text[i:i+3].encode()).hexdigest(), 16)
        vec[h % _EMBED_DIM] += 1.0
    norm = np.linalg.norm(vec)
    return vec / norm if norm > 0 else vec


def _client():
    try:
        import redis as redis_lib
        try:
            from dotenv import load_dotenv
            load_dotenv(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env"))
        except ImportError:
            pass
        r = redis_lib.Redis(
            host=os.getenv("REDIS_HOST", "localhost"),
            port=int(os.getenv("REDIS_PORT", 6379)),
            username=os.getenv("REDIS_USERNAME"),
            password=os.getenv("REDIS_PASSWORD"),
            decode_responses=False,
            ssl=False,
            socket_connect_timeout=3,
            socket_timeout=3,
        )
        r.ping()
        return r
    except Exception:
        return None


def store_alert(agent_id: str, reason: str, model: str, cost: float,
                retries: int, progress: int, outcome: str = "open") -> None:
    """Store an alert in Redis for future similarity lookup."""
    r = _client()
    if not r:
        return
    try:
        vec = _embed(f"{agent_id} {reason} {model or ''}")
        suffix = hashlib.md5(f"{agent_id}{time.time()}".encode()).hexdigest()[:12]
        key = f"{_KEY_PREFIX}{suffix}"
        r.hset(key, mapping={
            "agent_id": agent_id,
            "reason":   reason,
            "model":    model or "",
            "cost":     str(round(cost, 4)),
            "retries":  str(retries),
            "progress": str(progress),
            "outcome":  outcome,
            "ts":       str(time.time()),
            "embedding": vec.tobytes(),
        })
        r.sadd(_INDEX_KEY, key)
    except Exception:
        pass


def get_similar(agent_id: str, reason: str, model: str, top_k: int = 3,
                min_age_seconds: float = 3600) -> list:
    """Return top_k past alerts most similar to this one, from previous sessions only."""
    r = _client()
    if not r:
        return []
    try:
        import datetime
        query_vec = _embed(f"{agent_id} {reason} {model or ''}")
        keys = r.smembers(_INDEX_KEY)
        cutoff = time.time() - min_age_seconds
        scored = []
        for key in keys:
            raw = r.hgetall(key)
            if not raw or b"embedding" not in raw:
                continue
            ts = float(raw.get(b"ts", b"0").decode())
            if ts > cutoff:
                continue  # skip alerts from this session
            stored = np.frombuffer(raw[b"embedding"], dtype=np.float32)
            score = float(np.dot(query_vec, stored))
            scored.append((score, {
                "agent_id": raw.get(b"agent_id", b"").decode(),
                "reason":   raw.get(b"reason", b"").decode(),
                "model":    raw.get(b"model", b"").decode(),
                "cost":     raw.get(b"cost", b"0").decode(),
                "outcome":  raw.get(b"outcome", b"open").decode(),
                "time":     datetime.datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M") if ts else "?",
            }))
        scored.sort(key=lambda x: x[0], reverse=True)
        return [d for _, d in scored[:top_k]]
    except Exception:
        return []
```

### models.py

```python
"""
Static model tradeoff lookup table.
Specs are factual and published — no claims about which model will succeed at a specific task.
"""

MODELS = [
    # ── Anthropic ──────────────────────────────────────────────────────────────
    {
        "id": "claude-opus-4-8",
        "provider": "Anthropic",
        "input_cost_per_1m": 5.00,
        "output_cost_per_1m": 25.00,
        "context_window_k": 1000,
        "notes": "Most capable. Best for complex reasoning.",
        "tradeoff": "Highest cost per token of any model here.",
    },
    {
        "id": "claude-sonnet-4-6",
        "provider": "Anthropic",
        "input_cost_per_1m": 3.00,
        "output_cost_per_1m": 15.00,
        "context_window_k": 1000,
        "notes": "Strong balance of quality and cost.",
        "tradeoff": "Less capable than Opus on the hardest reasoning tasks.",
    },
    {
        "id": "claude-haiku-4-5",
        "provider": "Anthropic",
        "input_cost_per_1m": 1.00,
        "output_cost_per_1m": 5.00,
        "context_window_k": 200,
        "notes": "Fastest and cheapest Anthropic model. Good for simple, repetitive tasks.",
        "tradeoff": "Smaller 200K context; weaker on multi-step reasoning.",
    },
    # ── OpenAI ─────────────────────────────────────────────────────────────────
    {
        "id": "gpt-4o",
        "provider": "OpenAI",
        "input_cost_per_1m": 2.50,
        "output_cost_per_1m": 10.00,
        "context_window_k": 128,
        "notes": "OpenAI flagship. Strong reasoning and tool use.",
        "tradeoff": "Only 128K context; different provider means a new SDK and API key.",
    },
    {
        "id": "gpt-4o-mini",
        "provider": "OpenAI",
        "input_cost_per_1m": 0.15,
        "output_cost_per_1m": 0.60,
        "context_window_k": 128,
        "notes": "Very cheap OpenAI model. Good for high-volume, low-complexity steps.",
        "tradeoff": "Noticeably weaker reasoning; struggles on complex multi-step work.",
    },
    {
        "id": "gpt-4.1",
        "provider": "OpenAI",
        "input_cost_per_1m": 2.00,
        "output_cost_per_1m": 8.00,
        "context_window_k": 1000,
        "notes": "Latest GPT-4 generation. Large context, strong coding.",
        "tradeoff": "Different provider; quality varies by task vs Claude/Gemini.",
    },
    {
        "id": "gpt-4.1-mini",
        "provider": "OpenAI",
        "input_cost_per_1m": 0.40,
        "output_cost_per_1m": 1.60,
        "context_window_k": 1000,
        "notes": "Cheap GPT-4.1 variant. Good balance for agentic loops.",
        "tradeoff": "Mid-tier quality; not for the hardest reasoning steps.",
    },
    {
        "id": "gpt-4.1-nano",
        "provider": "OpenAI",
        "input_cost_per_1m": 0.10,
        "output_cost_per_1m": 0.40,
        "context_window_k": 1000,
        "notes": "Cheapest OpenAI model. Best for classification and simple extraction.",
        "tradeoff": "Lowest quality; only safe for trivial, well-scoped tasks.",
    },
    {
        "id": "o4-mini",
        "provider": "OpenAI",
        "input_cost_per_1m": 1.10,
        "output_cost_per_1m": 4.40,
        "context_window_k": 200,
        "notes": "Reasoning model. Cheaper than o3 for math and code tasks.",
        "tradeoff": "Higher latency from extra reasoning; overkill for simple steps.",
    },
    # ── Google ─────────────────────────────────────────────────────────────────
    {
        "id": "gemini-2.5-pro",
        "provider": "Google",
        "input_cost_per_1m": 1.25,
        "output_cost_per_1m": 10.00,
        "context_window_k": 1000,
        "notes": "Google flagship. Strong reasoning, very large context.",
        "tradeoff": "Different provider/SDK; tool-use behavior differs from Claude.",
    },
    {
        "id": "gemini-2.5-flash",
        "provider": "Google",
        "input_cost_per_1m": 0.30,
        "output_cost_per_1m": 2.50,
        "context_window_k": 1000,
        "notes": "Fast and cheap Google model. Good for high-throughput pipelines.",
        "tradeoff": "Weaker on complex reasoning than the Pro tier.",
    },
    {
        "id": "gemini-2.0-flash",
        "provider": "Google",
        "input_cost_per_1m": 0.10,
        "output_cost_per_1m": 0.40,
        "context_window_k": 1000,
        "notes": "Cheapest with large context. Fast throughput.",
        "tradeoff": "Older generation; lower quality than 2.5 models.",
    },
    # ── Meta (via API providers) ───────────────────────────────────────────────
    {
        "id": "llama-3.3-70b",
        "provider": "Meta/Groq",
        "input_cost_per_1m": 0.59,
        "output_cost_per_1m": 0.79,
        "context_window_k": 128,
        "notes": "Open-weight. Very cheap on Groq. Good for structured extraction.",
        "tradeoff": "Only 128K context; weaker tool use and instruction-following.",
    },
    {
        "id": "llama-3.1-8b",
        "provider": "Meta/Groq",
        "input_cost_per_1m": 0.05,
        "output_cost_per_1m": 0.08,
        "context_window_k": 128,
        "notes": "Smallest useful open model. Near-zero cost for simple tasks.",
        "tradeoff": "Low capability; unreliable on anything beyond simple tasks.",
    },
    # ── Mistral ────────────────────────────────────────────────────────────────
    {
        "id": "mistral-small-3.1",
        "provider": "Mistral",
        "input_cost_per_1m": 0.10,
        "output_cost_per_1m": 0.30,
        "context_window_k": 128,
        "notes": "Cheap Mistral model. Good for classification and simple generation.",
        "tradeoff": "Limited reasoning depth; 128K context only.",
    },
]


def get_cheaper_alternatives(current_model_id: str, budget_per_1m: float) -> list:
    """Return models cheaper than budget_per_1m input cost, excluding the current model."""
    return [
        m for m in MODELS
        if m["id"] != current_model_id and m["input_cost_per_1m"] < budget_per_1m
    ]


def format_tradeoffs(alternatives: list) -> str:
    if not alternatives:
    
[truncated — 565 more characters]
```

### redis_store.py

```python
"""
Redis-backed shared state for SelfAudit.

Data model:
  selfaudit:agents          SET   — all agent IDs active this session
  selfaudit:agent:{id}      HASH  — per-agent state (cost, retries, status, …)
  selfaudit:alerts          ZSET  — alert log, scored by insertion order
  selfaudit:flagged         ZSET  — flagged-for-review entries, scored by insertion order
  selfaudit:stats           HASH  — session-level totals (cost, counts)
  selfaudit:updates         channel — pub/sub for live SSE notifications

Falls back silently if Redis is unavailable — local-only mode still works.
"""

import json
import os
import time

_NS      = "selfaudit"
_CHANNEL = f"{_NS}:updates"
_TTL     = 86400  # 24 hours

_KEY_AGENTS  = f"{_NS}:agents"
_KEY_AGENT   = f"{_NS}:agent:"   # + agent_id
_KEY_ALERTS  = f"{_NS}:alerts"
_KEY_FLAGGED = f"{_NS}:flagged"
_KEY_STATS   = f"{_NS}:stats"


def _client():
    try:
        import redis as redis_lib
        try:
            from dotenv import load_dotenv
            load_dotenv(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env"))
        except ImportError:
            pass
        r = redis_lib.Redis(
            host=os.getenv("REDIS_HOST", "localhost"),
            port=int(os.getenv("REDIS_PORT", 6379)),
            username=os.getenv("REDIS_USERNAME"),
            password=os.getenv("REDIS_PASSWORD"),
            ssl=False,
            socket_connect_timeout=2,
            socket_timeout=2,
        )
        r.ping()
        return r
    except Exception:
        return None


def clear_session() -> None:
    """Delete all keys from the previous session so stale state doesn't bleed through."""
    r = _client()
    if not r:
        return
    try:
        pipe = r.pipeline()
        for aid_bytes in r.smembers(_KEY_AGENTS):
            pipe.delete(_KEY_AGENT + aid_bytes.decode())
        pipe.delete(_KEY_AGENTS)
        pipe.delete(_KEY_ALERTS)
        pipe.delete(_KEY_FLAGGED)
        pipe.delete(_KEY_STATS)
        pipe.execute()
    except Exception:
        pass


def push_snapshot(snapshot: dict) -> None:
    """
    Write dashboard state to Redis using typed data structures.
    Each agent gets its own HASH; alerts and flagged entries live in ZSETs.
    """
    r = _client()
    if not r:
        return
    try:
        pipe = r.pipeline()

        # session-level counters
        pipe.hset(_KEY_STATS, mapping={
            "total_cost":    snapshot["total_cost"],
            "alert_count":   snapshot["alert_count"],
            "done_count":    snapshot["done_count"],
            "running_count": snapshot["running_count"],
            "paused_count":  snapshot["paused_count"],
        })
        pipe.expire(_KEY_STATS, _TTL)

        # one HASH per agent
        for agent in snapshot["agents"]:
            aid = agent["agent_id"]
            pipe.sadd(_KEY_AGENTS, aid)
            pipe.hset(_KEY_AGENT + aid, mapping={
                "agent_id":       aid,
                "status":         agent["status"],
                "cost":           agent["cost"],
                "progress":       agent["progress"],
                "retries":        agent["retries"],
                "elapsed":        agent["elapsed"],
                "alerted":        int(agent["alerted"]),
                "paused":         int(agent["paused"]),
                "flagged":        int(agent["flagged"]),
                "alert_reason":   agent["alert_reason"] or "",
                "proj_1h":        agent["proj_1h"],
                "budget":         agent["budget"] or "",
                "model":          agent["model"] or "",
                "progress_mode":  agent["progress_mode"],
                "t_retry":        agent["t_retry"] if agent["t_retry"] is not None else "",
                "t_cost":         agent["t_cost"] if agent["t_cost"] is not None else "",
                "t_time":         agent["t_time"] if agent["t_time"] is not None else "",
                "notes":          json.dumps(agent["notes"]),
                "recent_actions": json.dumps(agent["recent_actions"]),
            })
            pipe.expire(_KEY_AGENT + aid, _TTL)
        pipe.expire(_KEY_AGENTS, _TTL)

        # alert log as a ZSET (score = insertion index, preserves order)
        pipe.delete(_KEY_ALERTS)
        for i, alert in enumerate(snapshot["alerts"]):
            pipe.zadd(_KEY_ALERTS, {json.dumps(alert): i})
        if snapshot["alerts"]:
            pipe.expire(_KEY_ALERTS, _TTL)

        # flagged-for-review as a ZSET
        pipe.delete(_KEY_FLAGGED)
        for i, entry in enumerate(snapshot["flagged"]):
            pipe.zadd(_KEY_FLAGGED, {json.dumps(entry): i})
        if snapshot["flagged"]:
            pipe.expire(_KEY_FLAGGED, _TTL)

        pipe.execute()
        r.publish(_CHANNEL, str(time.time()))
    except Exception:
        pass


def get_snapshot():
    """
    Reconstruct the dashboard snapshot from individual Redis keys.
    Returns None if Redis is unavailable or no data exists yet.
    """
    r = _client()
    if not r:
        return None
    try:
        stats_raw = r.hgetall(_KEY_STATS)
        if not stats_raw:
            return None

        def s(v):
            return v.decode() if isinstance(v, bytes) else v

        stats = {s(k): s(v) for k, v in stats_raw.items()}

        # rebuild agent list from individual hashes
        agents = []
        for aid_bytes in r.smembers(_KEY_AGENTS):
            aid = aid_bytes.decode()
            raw = r.hgetall(_KEY_AGENT + aid)
            if not raw:
                continue
            h = {k.decode(): v.decode() for k, v in raw.items()}
            agents.append({
                "agent_id":       h["agent_id"],
                "status":         h["status"],
                "cost":           h["cost"],
                "progress":       int(h["progress"]),
                "retries":        int(h["retries"]),
                "elapsed":        h["elapsed"],
                "alerted":        bool(int(h
[truncated — 2527 more characters]
```

### sdk.py

```python
"""
SelfAudit SDK — drop-in monitoring for any Python AI agent.

Quickstart:
    from selfaudit.sdk import Watcher

    watcher = Watcher()
    watcher.start_dashboard()          # http://localhost:5050

    with watcher.trace("my-agent", action="summarize") as t:
        response = client.messages.create(...)
        t.success_from_anthropic(response)

    with watcher.trace("my-agent", action="call_api") as t:
        if result.ok:
            t.success(cost_usd=0.001)
        else:
            t.fail()
"""

import os
import time
import threading
import datetime
from contextlib import contextmanager
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Set

try:
    from dotenv import load_dotenv
    load_dotenv(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env"))
except ImportError:
    pass

from models import MODELS, get_cheaper_alternatives, format_tradeoffs

_ANTHROPIC_PRICES = {
    "claude-opus-4-8":   (5.00 / 1_000_000, 25.00 / 1_000_000),
    "claude-sonnet-4-6": (3.00 / 1_000_000, 15.00 / 1_000_000),
    "claude-haiku-4-5":  (1.00 / 1_000_000,  5.00 / 1_000_000),
}

def _model_rec(model: Optional[str]) -> str:
    m = next((x for x in MODELS if x["id"] == model), None)
    price = m["input_cost_per_1m"] if m else 5.00
    alts = get_cheaper_alternatives(model or "claude-opus-4-8", price)
    return format_tradeoffs(alts)


def _build_recommendation(agent_id: str, model: Optional[str],
                           cost: float, retries: int, progress: int,
                           situation: str) -> dict:
    """Build context-aware recommendation. Returns a dict for structured rendering."""
    m = next((x for x in MODELS if x["id"] == model), None)
    price = m["input_cost_per_1m"] if m else 5.00
    alts = get_cheaper_alternatives(model or "claude-opus-4-8", price)
    model_label = model or "unknown model"

    if situation == "zero_progress":
        headline = f"{retries} retries, 0 steps completed. ${cost:.4f} on {model_label}"
        steps = [
            "Something's broken upstream. Retrying won't help. Check the error first.",
            f"Diagnose on a cheaper model before burning more {model_label} calls.",
        ]
    elif situation == "stuck_subtask":
        headline = f"Stalled after {progress} step(s), {retries} retries on the same call"
        steps = [
            f"The first {progress} step(s) are fine. Only the failing sub-task needs attention.",
            f"Consider a cheaper model just for the stuck call; {model_label} isn't needed to retry a broken step.",
        ]
    elif situation == "high_cost_ratio":
        headline = f"${cost:.4f} for {progress} step(s) on {model_label}. Cost is high relative to output."
        steps = [
            "Find the expensive call. One step usually accounts for most of it.",
            f"Route simpler steps to a cheaper model; save {model_label} for what actually needs it.",
        ]
    else:
        headline = f"{agent_id}: ${cost:.4f}, {progress} steps, {retries} retries on {model_label}"
        steps = []

    return {"headline": headline, "steps": steps, "alternatives": alts}

RETRY_ALERT_THRESHOLD    = 3
COST_STALL_THRESHOLD_USD = 0.05
EXPECTED_TASK_SECONDS    = 30.0


# ── data model ─────────────────────────────────────────────────────────────────

@dataclass
class _Event:
    agent_id: str
    action: str
    cost_usd: float
    success: bool
    cumulative_cost_usd: float
    retry_count: int
    timestamp: float = field(default_factory=time.time)
    completed: bool = False


@dataclass
class _State:
    agent_id: str
    events: List[_Event] = field(default_factory=list)
    unique_successes: set = field(default_factory=set)
    total_successes: int = 0
    retry_counts: Dict[str, int] = field(default_factory=dict)
    completed: bool = False
    alerted: bool = False
    paused: bool = False
    flagged: bool = False
    notes: List[str] = field(default_factory=list)
    peer_verdict: Optional[dict] = None
    model: Optional[str] = None          # last model used by this agent
    budget_usd: Optional[float] = None   # hard cap; None = no cap
    progress_mode: str = "unique"        # "unique" or "total"
    retry_threshold: Optional[int] = None    # per-agent override; None = use global
    cost_threshold: Optional[float] = None
    time_threshold: Optional[float] = None
    start_time: float = field(default_factory=time.time)

    @property
    def cumulative_cost(self) -> float:
        return self.events[-1].cumulative_cost_usd if self.events else 0.0

    @property
    def progress_score(self) -> int:
        return self.total_successes if self.progress_mode == "total" else len(self.unique_successes)

    @property
    def max_retry_count(self) -> int:
        return max(self.retry_counts.values(), default=0)

    @property
    def cost_rate_per_min(self) -> float:
        """Average spend per minute since start."""
        elapsed_min = self.elapsed / 60.0
        return (self.cumulative_cost / elapsed_min) if elapsed_min > 0.01 else 0.0

    @property
    def projected_cost_1h(self) -> float:
        return self.cost_rate_per_min * 60.0

    @property
    def elapsed(self) -> float:
        return time.time() - self.start_time


@dataclass
class Alert:
    agent_id: str
    reason: str
    cost_usd: float
    retry_count: int
    progress_score: int
    recommendation: dict
    id: str = field(default_factory=lambda: str(time.time()))
    timestamp: float = field(default_factory=time.time)
    dismissed: bool = False


# ── trace handle ───────────────────────────────────────────────────────────────

class TraceHandle:
    def __init__(self, watcher: "Watcher", agent_id: str, action: str, model: str = None):
        self._watcher  = watcher
        self._agent_id = agent_id
        self._action   = action
        self._model    = model
        self._resolved = False

    def success(self, cost_usd: float = 0.0, completed: bool = False, output: str
[truncated — 21353 more characters]
```

[1 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]