Develop

api/claims.py

@@ -1,11 +1,12 @@
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, HTTPException, Request
 from fastapi.concurrency import run_in_threadpool
 from pydantic import BaseModel
 from services.agents.run import start_claim, resume_claim
 from services.agents.audit import audit_node
 from services.ingest.extractor import extract_claim
 from fastapi import Depends
 from .auth import require_key
+from .limiter import limiter
 from . import store
 import json
 
@@ -18,7 +19,8 @@ class ClaimIn(BaseModel):
 
 
 @router.post("/claims", dependencies=[Depends(require_key)])
-async def submit(body: ClaimIn):
+@limiter.limit("10/minute")          # submit is paid — cap it hard
+async def submit(request: Request, body: ClaimIn):
     def _run():
         lbl = json.load(open(f"{body.claim_dir}/label.json"))
         pol = lbl["policy"]
@@ -42,7 +44,8 @@ def _run():
 
 
 @router.get("/claims")
-def list_claims():
+@limiter.limit("60/minute")          # reads are cheap — looser
+def list_claims(request: Request):
     # compact rows for the queue view
     return [{"claim_id": c["claim_id"], "status": c["status"],
              "recommendation": c["recommendation"]["decision"],

api/limiter.py

@@ -0,0 +1,4 @@
+from slowapi import Limiter
+from slowapi.util import get_remote_address
+
+limiter = Limiter(key_func=get_remote_address)

api/main.py

@@ -1,15 +1,19 @@
 import asyncio
 import os
 from contextlib import asynccontextmanager
-from fastapi import FastAPI
+from fastapi import FastAPI, Request, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from services.mcp.host import Host
 from services.agents.tools import set_shared_host
+from slowapi import _rate_limit_exceeded_handler
+from slowapi.errors import RateLimitExceeded
+from .limiter import limiter
 from .claims import router as claims_router
 
-
 ORIGINS = os.getenv("ALLOWED_ORIGINS", "http://localhost:5173").split(",")
 
+MAX_BODY = 1_000_000      # 1 MB — generous for a claim ref, caps abuse
+
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
@@ -23,6 +27,8 @@ async def lifespan(app: FastAPI):
     await app.state.host.close()
 
 app = FastAPI(title="Evincta API", lifespan=lifespan)
+app.state.limiter = limiter
+app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 
 # dev CORS — locked down to the real origin in 6b
 app.add_middleware(CORSMiddleware, allow_origins=ORIGINS,
@@ -34,3 +40,11 @@ async def lifespan(app: FastAPI):
 
 @app.get("/health")
 def health(): return {"status": "ok"}
+
+
+@app.middleware("http")
+async def limit_body(request: Request, call_next):
+    cl = request.headers.get("content-length")
+    if cl and int(cl) > MAX_BODY:
+        raise HTTPException(413, "payload too large")
+    return await call_next(request)

services/agents/specialists.py

@@ -1,5 +1,8 @@
 from .tools import call_tool
 from services.index.retrieve import find_precedents
+from services.common.breaker import CircuitBreaker
+
+_pre_breaker = CircuitBreaker(fails=5, reset_after=30)
 
 
 def coverage_node(state: dict) -> dict:
@@ -22,8 +25,12 @@ def cost_node(state: dict) -> dict:
 def precedent_node(state: dict) -> dict:
     r = state["record"]
     text = f"{r['incident_type']} {r['severity']} damage. {r['notes_summary']}"
-    hits = find_precedents(state["image_path"], text, k=5)
-    return {"precedents": hits}
+    try:
+        hits = _pre_breaker.call(find_precedents, state["image_path"], text, k=5)
+        return {"precedents": hits}
+    except Exception:
+        # degraded: no precedents → flag so the synthesiser escalates
+        return {"precedents": [], "degraded": True}
 
 
 def fraud_node(state: dict) -> dict:

services/agents/synthesiser.py

@@ -1,12 +1,14 @@
 import os
 import json
 import anthropic
+from services.common.pricing import est_cost
+from services.common.budget import add_and_check
 from dotenv import load_dotenv
 from langfuse import observe, get_client
 from .schema import Recommendation
 
 load_dotenv()
-client = anthropic.Anthropic()
+client = anthropic.Anthropic(max_retries=4, timeout=60.0)
 langfuse = get_client()
 MODEL = os.getenv("ANTHROPIC_MODEL", "claude-sonnet-4-5")
 
@@ -36,6 +38,10 @@ def synthesiser_node(state: dict) -> dict:
         tool_choice={"type": "tool", "name": "recommend"},
         messages=[{"role": "user",
                    "content": f"Findings:\n{json.dumps(evidence, indent=2)}"}])
+
+    cost = est_cost(msg.usage, model="sonnet")   # tokens → $
+    add_and_check(cost)
+
     rec = next(b.input for b in msg.content if b.type == "tool_use")
     Recommendation.model_validate(rec)          # validate at the boundary
     langfuse.update_current_span(metadata={

services/common/breaker.py

@@ -0,0 +1,24 @@
+import time
+
+
+class CircuitBreaker:
+    """Open after N failures; reject fast; half-open to test recovery."""
+    def __init__(self, fails=5, reset_after=30):
+        self.fails, self.reset_after = fails, reset_after
+        self.count, self.opened_at = 0, None
+
+    def call(self, fn, *a, **k):
+        if self.opened_at:                       # breaker is OPEN
+            if time.time() - self.opened_at < self.reset_after:
+                raise RuntimeError("circuit open — dependency unavailable")
+            self.opened_at = None  # half-open: try once
+            self.count = 0
+        try:
+            r = fn(*a, **k)
+            self.count = 0  # success resets
+            return r
+        except Exception:
+            self.count += 1
+            if self.count >= self.fails:
+                self.opened_at = time.time()         # trip OPEN
+            raise

services/common/budget.py

@@ -0,0 +1,22 @@
+import datetime
+import json
+from pathlib import Path
+from threading import Lock
+
+CAP_USD = 2.00                      # daily ceiling — tune as you like
+F = Path("data/spend.json")
+_lock = Lock()
+
+
+def _today(): return datetime.date.today().isoformat()
+
+
+def add_and_check(cost_usd: float):
+    """Record spend; raise if today's total exceeds the cap."""
+    with _lock:
+        d = json.loads(F.read_text()) if F.exists() else {}
+        day = _today()
+        d[day] = round(d.get(day, 0) + cost_usd, 4)
+        F.write_text(json.dumps(d))
+        if d[day] > CAP_USD:
+            raise RuntimeError(f"daily spend cap hit (${d[day]} > ${CAP_USD})")

services/common/pricing.py

@@ -0,0 +1,15 @@
+"""Token → USD cost for Claude calls. Prices are per million tokens."""
+
+# Claude Sonnet pricing (check current rates at anthropic.com/pricing)
+# stored per-token = per-million ÷ 1_000_000
+PRICES = {
+    "sonnet": {"input": 3.00 / 1_000_000, "output": 15.00 / 1_000_000},
+    # add other models if you use them, e.g. haiku
+}
+
+
+def est_cost(usage, model: str = "sonnet") -> float:
+    """Estimate the USD cost of one call from its token usage."""
+    p = PRICES[model]
+    return (usage.input_tokens * p["input"]
+            + usage.output_tokens * p["output"])

services/common/resilient.py

@@ -0,0 +1,13 @@
+from tenacity import (retry, stop_after_attempt, wait_exponential_jitter,
+                      retry_if_exception_type)
+
+# retry transient network/5xx; backoff 1s,2s,4s... with jitter; 4 tries
+external_retry = retry(
+    stop=stop_after_attempt(4),
+    wait=wait_exponential_jitter(initial=1, max=10),
+    retry=retry_if_exception_type((ConnectionError, TimeoutError, Exception)),
+    reraise=True)
+
+# usage: decorate the embed + pinecone calls
+#   @external_retry
+#   def embed_text(...): ...

services/ingest/extractor.py

@@ -13,13 +13,14 @@
 
 import anthropic
 from langfuse import observe, get_client
-
+from services.common.pricing import est_cost
+from services.common.budget import add_and_check
 from .schema import ClaimRecord
 
 load_dotenv()
 
 # ── clients & config ───────────────────────────────────────────────
-client = anthropic.Anthropic()              # reads ANTHROPIC_API_KEY from env
+client = anthropic.Anthropic(max_retries=4, timeout=60.0)  # reads ANTHROPIC_API_KEY from env
 langfuse = get_client()                       # reads LANGFUSE_* from env
 MODEL = os.getenv("ANTHROPIC_MODEL", "claude-sonnet-4-5")
 
@@ -94,6 +95,8 @@ def extract_claim(claim_dir: str | Path) -> ClaimRecord:
         tool_choice={"type": "tool", "name": "record_claim"},
         messages=[{"role": "user", "content": content}],
     )
+    cost = est_cost(msg.usage, model="sonnet")
+    add_and_check(cost)
 
     tool_use = next(b for b in msg.content if b.type == "tool_use")
     record = ClaimRecord.model_validate(tool_use.input)   # validates here