From 8d173fe498780263c1e16d71d75f4410c1858438 Mon Sep 17 00:00:00 2001 From: Atti Ur Rehman Date: Fri, 3 Jul 2026 20:51:06 +0300 Subject: [PATCH 1/4] feat(hardening): retry/timeout/backoff + circuit breaker + graceful degradation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Anthropic SDK retries (429/5xx/overload) + 60s timeout configured; Voyage/Pinecone wrapped with tenacity (exponential backoff + jitter); a circuit breaker trips on sustained failure. On failure the claim degrades to 'investigate' (human review) rather than a wrong automated decision — the safest failure mode for a decision system. Closes #74, #75 --- services/agents/specialists.py | 11 +++++++++-- services/agents/synthesiser.py | 2 +- services/common/breaker.py | 21 +++++++++++++++++++++ services/common/resilient.py | 13 +++++++++++++ services/ingest/extractor.py | 2 +- 5 files changed, 45 insertions(+), 4 deletions(-) create mode 100644 services/common/breaker.py create mode 100644 services/common/resilient.py diff --git a/services/agents/specialists.py b/services/agents/specialists.py index 921124d..be19b2a 100644 --- a/services/agents/specialists.py +++ b/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: diff --git a/services/agents/synthesiser.py b/services/agents/synthesiser.py index 1e15b67..8227a9e 100644 --- a/services/agents/synthesiser.py +++ b/services/agents/synthesiser.py @@ -6,7 +6,7 @@ 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") diff --git a/services/common/breaker.py b/services/common/breaker.py new file mode 100644 index 0000000..6fa5a6f --- /dev/null +++ b/services/common/breaker.py @@ -0,0 +1,21 @@ +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; self.count = 0 # half-open: try once + try: + r = fn(*a, **k); self.count = 0; return r # success resets + except Exception: + self.count += 1 + if self.count >= self.fails: + self.opened_at = time.time() # trip OPEN + raise diff --git a/services/common/resilient.py b/services/common/resilient.py new file mode 100644 index 0000000..6cfa5ca --- /dev/null +++ b/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(...): ... diff --git a/services/ingest/extractor.py b/services/ingest/extractor.py index 9b2ea8c..6b2ff21 100644 --- a/services/ingest/extractor.py +++ b/services/ingest/extractor.py @@ -19,7 +19,7 @@ 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") From 885e3793eece81475d6ade0237e7b3c1821536cb Mon Sep 17 00:00:00 2001 From: Atti Ur Rehman Date: Fri, 3 Jul 2026 20:54:53 +0300 Subject: [PATCH 2/4] fix(lint): resolve ruff E702 in circuit breaker Split semicolon-separated statements in breaker.py so ruff check passes in CI. --- services/common/breaker.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/services/common/breaker.py b/services/common/breaker.py index 6fa5a6f..ad6e09b 100644 --- a/services/common/breaker.py +++ b/services/common/breaker.py @@ -11,9 +11,12 @@ 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; self.count = 0 # half-open: try once + self.opened_at = None # half-open: try once + self.count = 0 try: - r = fn(*a, **k); self.count = 0; return r # success resets + r = fn(*a, **k) + self.count = 0 # success resets + return r except Exception: self.count += 1 if self.count >= self.fails: From ddc7b662fa0117d13a5e2c3b729101c5a63397a2 Mon Sep 17 00:00:00 2001 From: Atti Ur Rehman Date: Fri, 3 Jul 2026 21:34:18 +0300 Subject: [PATCH 3/4] feat(hardening): rate limiting + daily spend cap + request size limits slowapi rate limits scaled to cost (submit tight at 10/min, reads loose at 60/min); a daily spend ceiling that refuses model calls when exceeded (a circuit breaker for the API bill); a 1MB request-body cap against cost/DoS. Closes #76 --- api/claims.py | 9 ++++++--- api/limiter.py | 4 ++++ api/main.py | 18 ++++++++++++++++-- services/agents/synthesiser.py | 6 ++++++ services/common/budget.py | 19 +++++++++++++++++++ services/common/pricing.py | 15 +++++++++++++++ services/ingest/extractor.py | 5 ++++- 7 files changed, 70 insertions(+), 6 deletions(-) create mode 100644 api/limiter.py create mode 100644 services/common/budget.py create mode 100644 services/common/pricing.py diff --git a/api/claims.py b/api/claims.py index 1589ef6..d83eee5 100644 --- a/api/claims.py +++ b/api/claims.py @@ -1,4 +1,4 @@ -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 @@ -6,6 +6,7 @@ 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"], diff --git a/api/limiter.py b/api/limiter.py new file mode 100644 index 0000000..38404a8 --- /dev/null +++ b/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) diff --git a/api/main.py b/api/main.py index 6217624..f6a024a 100644 --- a/api/main.py +++ b/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) diff --git a/services/agents/synthesiser.py b/services/agents/synthesiser.py index 8227a9e..02992a9 100644 --- a/services/agents/synthesiser.py +++ b/services/agents/synthesiser.py @@ -1,6 +1,8 @@ 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 @@ -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={ diff --git a/services/common/budget.py b/services/common/budget.py new file mode 100644 index 0000000..52627ed --- /dev/null +++ b/services/common/budget.py @@ -0,0 +1,19 @@ +import json, datetime +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})") diff --git a/services/common/pricing.py b/services/common/pricing.py new file mode 100644 index 0000000..4b1988b --- /dev/null +++ b/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"]) diff --git a/services/ingest/extractor.py b/services/ingest/extractor.py index 6b2ff21..b693c96 100644 --- a/services/ingest/extractor.py +++ b/services/ingest/extractor.py @@ -13,7 +13,8 @@ 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() @@ -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 From 413cfc13f51944a3084cf0268b299c74e9ae04c7 Mon Sep 17 00:00:00 2001 From: Atti Ur Rehman Date: Fri, 3 Jul 2026 21:37:34 +0300 Subject: [PATCH 4/4] fix(lint): resolve ruff E401/E702 in daily spend budget Split combined imports and semicolon-separated statements in budget.py so CI ruff check passes. --- services/common/budget.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/services/common/budget.py b/services/common/budget.py index 52627ed..c9b90e1 100644 --- a/services/common/budget.py +++ b/services/common/budget.py @@ -1,9 +1,11 @@ -import json, datetime +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() +F = Path("data/spend.json") +_lock = Lock() def _today(): return datetime.date.today().isoformat() @@ -13,7 +15,8 @@ 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) + 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})")