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/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..02992a9 100644 --- a/services/agents/synthesiser.py +++ b/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={ diff --git a/services/common/breaker.py b/services/common/breaker.py new file mode 100644 index 0000000..ad6e09b --- /dev/null +++ b/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 diff --git a/services/common/budget.py b/services/common/budget.py new file mode 100644 index 0000000..c9b90e1 --- /dev/null +++ b/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})") 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/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..b693c96 100644 --- a/services/ingest/extractor.py +++ b/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