diff --git a/app/routes/chat.py b/app/routes/chat.py
index 750ea85..4fba9f2 100644
--- a/app/routes/chat.py
+++ b/app/routes/chat.py
@@ -30,6 +30,7 @@
 from app.deps import get_db, get_key_context
 from app.quality_scores import resolve_model_metrics
 from app.schemas import ChatCompletionRequest
+from packages.auth.spend import budget_exceeded, get_lifetime_spend_microcents
 from packages.auth.types import KeyContext
 from packages.db.models.request_log import RequestLog
 from packages.litellm_adapter.catalog import CATALOG, CATALOG_BY_ID
@@ -267,6 +268,21 @@ async def chat_completions(
             detail=f"Model '{body.model}' is not allowed for this API key",
         )
 
+    # Budget enforcement: `budget_limit_cents` is a lifetime cap on this
+    # key's billable (status < 400) spend. Checked before any routing,
+    # resolution, or cache work so an exhausted key costs the operator
+    # nothing — no upstream attempt, no cache fill.
+    if kc.budget_limit_cents is not None:
+        spend = await get_lifetime_spend_microcents(db, str(kc.key_id))
+        if budget_exceeded(spend, kc.budget_limit_cents):
+            raise HTTPException(
+                status_code=429,
+                detail=(
+                    "API key budget exhausted "
+                    f"({spend} of {kc.budget_limit_cents * 10_000} microcents spent)."
+                ),
+            )
+
     client = await router_cache.get_router(db)
     raw_strategy = getattr(client, "strategy", None)
     strategy = raw_strategy if isinstance(raw_strategy, str) and raw_strategy else "balanced"
diff --git a/app/routes/keys.py b/app/routes/keys.py
index 2d1ca86..380b845 100644
--- a/app/routes/keys.py
+++ b/app/routes/keys.py
@@ -5,7 +5,7 @@
 from datetime import datetime, timezone
 
 from fastapi import APIRouter, Depends, HTTPException, Response
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -20,16 +20,46 @@
 
 class CreateKey(BaseModel):
     name: str
+    # Optional restrictions for child keys. Only reachable by unrestricted
+    # callers (require_unrestricted above), so a restricted key can never
+    # mint a sibling with looser limits than its own — it can't mint at all.
+    model_allowlist: list[str] | None = None
+    budget_limit_cents: int | None = Field(default=None, gt=0)
+
+
+def require_unrestricted(kc: KeyContext) -> None:
+    """Key management is reserved for unrestricted keys.
+
+    A key that carries any restriction (`model_allowlist` or
+    `budget_limit_cents`) must not be able to mint, list, or revoke other
+    keys — otherwise it could create a sibling with no restrictions and
+    trivially bypass its own allowlist/budget. Unrestricted keys already
+    hold the maximum privilege this single-workspace edition exposes
+    (same trust level as PUT /v1/providers/*), so denying restricted keys
+    here grants nothing to anyone; it only closes the escalation path.
+    """
+    if kc.model_allowlist is not None or kc.budget_limit_cents is not None:
+        raise HTTPException(
+            status_code=403,
+            detail=(
+                "Restricted API keys cannot manage keys. "
+                "Use an unrestricted key."
+            ),
+        )
 
 
 @router.get("")
 async def list_keys(
-    _kc: KeyContext = Depends(get_key_context),
+    kc: KeyContext = Depends(get_key_context),
     db: AsyncSession = Depends(get_db),
 ) -> dict:
+    require_unrestricted(kc)
     rows = (
         await db.execute(
-            select(ApiKey).where(ApiKey.is_deleted == 0).order_by(ApiKey.created_at)
+            select(ApiKey).where(
+                ApiKey.workspace_id == kc.workspace_id,
+                ApiKey.is_deleted == 0,
+            ).order_by(ApiKey.created_at)
         )
     ).scalars().all()
     return {
@@ -54,12 +84,15 @@ async def create_key(
     kc: KeyContext = Depends(get_key_context),
     db: AsyncSession = Depends(get_db),
 ) -> dict:
+    require_unrestricted(kc)
     full_key, key_hash, key_prefix = generate_api_key()
     row = ApiKey(
         workspace_id=kc.workspace_id,
         name=body.name,
         key_hash=key_hash,
         key_prefix=key_prefix,
+        model_allowlist=body.model_allowlist,
+        budget_limit_cents=body.budget_limit_cents,
     )
     db.add(row)
     await db.commit()
@@ -70,18 +103,28 @@ async def create_key(
         "name": row.name,
         "key_prefix": row.key_prefix,
         "api_key": full_key,  # plaintext shown ONCE
+        "model_allowlist": row.model_allowlist,
+        "budget_limit_cents": row.budget_limit_cents,
     }
 
 
 @router.delete("/{key_id}", status_code=204)
 async def revoke_key(
     key_id: str,
-    _kc: KeyContext = Depends(get_key_context),
+    kc: KeyContext = Depends(get_key_context),
     db: AsyncSession = Depends(get_db),
 ) -> Response:
+    require_unrestricted(kc)
     row = (
         await db.execute(
-            select(ApiKey).where(ApiKey.id == key_id, ApiKey.is_deleted == 0)
+            select(ApiKey).where(
+                ApiKey.id == key_id,
+                # Workspace scoping: without this, any key could revoke any
+                # other workspace's keys (the write path has always been
+                # scoped; the read/delete paths were not).
+                ApiKey.workspace_id == kc.workspace_id,
+                ApiKey.is_deleted == 0,
+            )
         )
     ).scalar_one_or_none()
     if row is None:
diff --git a/packages/auth/spend.py b/packages/auth/spend.py
new file mode 100644
index 0000000..ded59ed
--- /dev/null
+++ b/packages/auth/spend.py
@@ -0,0 +1,34 @@
+"""Per-key spend lookup used to enforce `ApiKey.budget_limit_cents`.
+
+Semantics: `budget_limit_cents` is a lifetime cap on the key's billable
+spend — request-log rows with `status_code < 400`. 1 cent = 10,000
+microcents (1 USD = 1,000,000 microcents, matching chat.py's cost math).
+
+Kept free of FastAPI imports so it stays unit-testable and reusable from
+non-HTTP contexts (background jobs, CLI minting tools).
+"""
+
+from __future__ import annotations
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from packages.db.models.request_log import RequestLog
+
+MICROCENTS_PER_CENT = 10_000
+
+
+async def get_lifetime_spend_microcents(
+    session: AsyncSession, api_key_id: str
+) -> int:
+    """Sum of billable (status < 400) spend ever recorded for this key."""
+    stmt = select(func.coalesce(func.sum(RequestLog.cost_microcents), 0)).where(
+        RequestLog.api_key_id == api_key_id,
+        RequestLog.is_deleted == 0,
+        RequestLog.status_code < 400,
+    )
+    return int((await session.execute(stmt)).scalar_one())
+
+
+def budget_exceeded(spend_microcents: int, budget_limit_cents: int) -> bool:
+    return spend_microcents >= budget_limit_cents * MICROCENTS_PER_CENT
diff --git a/tests/integration/test_budget_enforcement.py b/tests/integration/test_budget_enforcement.py
new file mode 100644
index 0000000..5eff090
--- /dev/null
+++ b/tests/integration/test_budget_enforcement.py
@@ -0,0 +1,235 @@
+"""Budget enforcement on /v1/chat/completions.
+
+`budget_limit_cents` was loaded into KeyContext but never enforced anywhere —
+a leaked key meant unbounded spend. These tests pin the new behavior: an
+exhausted key gets 429 before any routing / cache / upstream work,
+unbudgeted keys are unaffected, and the keys API can provision
+budgeted/allowlisted child keys.
+"""
+
+from __future__ import annotations
+
+import time
+from unittest.mock import AsyncMock
+
+import pytest
+
+
+@pytest.fixture
+async def budget_env(tmp_sqlite_url, monkeypatch):
+    """Full app + seeded root key, with the router client mocked out.
+
+    Yields (make_client, fake_client, session_factory, root_key).
+    """
+    monkeypatch.setenv("DATABASE_URL", tmp_sqlite_url)
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-test-openai")
+
+    from app import config as cfg
+    cfg.get_settings.cache_clear()
+
+    from packages.db.engine import build_engine
+    from packages.db.models.base import Base
+
+    engine = build_engine(tmp_sqlite_url)
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+    from sqlalchemy.ext.asyncio import async_sessionmaker
+
+    from packages.db import session as session_mod
+    factory = async_sessionmaker(engine, expire_on_commit=False)
+    session_mod._session_factory = factory
+
+    from app.seed import seed_initial_state
+    async with factory() as s:
+        seed = await seed_initial_state(s)
+
+    fake_client = AsyncMock()
+    fake_client.acompletion = AsyncMock(
+        return_value={
+            "id": "chatcmpl-budget-test",
+            "model": "gpt-4o-mini",
+            "object": "chat.completion",
+            "created": int(time.time()),
+            "choices": [{
+                "index": 0,
+                "message": {"role": "assistant", "content": "Hello!"},
+                "finish_reason": "stop",
+            }],
+            "usage": {"prompt_tokens": 5, "completion_tokens": 2, "total_tokens": 7},
+            "_orca_meta": {
+                "provider": "openai",
+                "litellm_model": "openai/gpt-4o-mini",
+                "latency_ms": 42,
+            },
+        }
+    )
+
+    from app import router_cache
+    router_cache.invalidate_router()
+
+    async def _fake_get_router(_session):
+        return fake_client
+
+    monkeypatch.setattr(router_cache, "get_router", _fake_get_router)
+
+    from httpx import ASGITransport, AsyncClient
+
+    from app.main import create_app
+    app = create_app()
+
+    async def make_client(api_key: str):
+        return AsyncClient(
+            transport=ASGITransport(app=app),
+            base_url="http://t",
+            headers={"Authorization": f"Bearer {api_key}"},
+        )
+
+    yield make_client, fake_client, factory, seed.api_key
+
+    await engine.dispose()
+    session_mod._session_factory = None
+
+
+async def _make_budgeted_key(
+    factory, *, budget_limit_cents: int | None
+) -> tuple[str, str]:
+    """Insert a budgeted child key; return (plaintext_key, key_id)."""
+    from packages.auth.hashing import generate_api_key
+    from packages.db.models.api_key import ApiKey
+
+    full_key, key_hash, key_prefix = generate_api_key()
+    async with factory() as s:
+        row = ApiKey(
+            workspace_id="default",
+            name="budgeted",
+            key_hash=key_hash,
+            key_prefix=key_prefix,
+            budget_limit_cents=budget_limit_cents,
+        )
+        s.add(row)
+        await s.commit()
+        await s.refresh(row)
+        return full_key, row.id
+
+
+async def _add_billable_spend(factory, key_id: str, microcents: int) -> None:
+    from packages.db.models.request_log import RequestLog
+
+    async with factory() as s:
+        s.add(RequestLog(
+            workspace_id="default",
+            api_key_id=key_id,
+            trace_id="budget-test-trace",
+            model_requested="gpt-4o-mini",
+            model_resolved="gpt-4o-mini",
+            provider="openai",
+            routing_strategy="balanced",
+            input_tokens=5,
+            output_tokens=2,
+            cost_microcents=microcents,
+            latency_ms=10,
+            status_code=200,
+        ))
+        await s.commit()
+
+
+async def test_exhausted_budget_returns_429_without_upstream_call(budget_env):
+    make_client, fake, factory, _root = budget_env
+    key, key_id = await _make_budgeted_key(factory, budget_limit_cents=1)
+    # Pre-load spend past the 1-cent cap (10_000 microcents).
+    await _add_billable_spend(factory, key_id, microcents=20_000)
+
+    async with await make_client(key) as c:
+        r = await c.post(
+            "/v1/chat/completions",
+            json={"model": "gpt-4o-mini",
+                  "messages": [{"role": "user", "content": "hi"}]},
+        )
+
+    assert r.status_code == 429, r.text
+    assert r.json()["error"]["type"] == "rate_limit_error"
+    fake.acompletion.assert_not_awaited()
+
+
+async def test_blocked_request_writes_no_log_row(budget_env):
+    make_client, _fake, factory, _root = budget_env
+    key, key_id = await _make_budgeted_key(factory, budget_limit_cents=1)
+    await _add_billable_spend(factory, key_id, microcents=99_999)
+
+    async with await make_client(key) as c:
+        await c.post(
+            "/v1/chat/completions",
+            json={"model": "gpt-4o-mini",
+                  "messages": [{"role": "user", "content": "hi"}]},
+        )
+
+    from sqlalchemy import func, select
+
+    from packages.db.models.request_log import RequestLog
+
+    async with factory() as s:
+        count = (
+            await s.execute(
+                select(func.count()).select_from(RequestLog).where(
+                    RequestLog.api_key_id == key_id
+                )
+            )
+        ).scalar_one()
+    assert count == 1  # only the pre-loaded history row
+
+
+async def test_under_budget_key_serves_normally(budget_env):
+    make_client, fake, factory, _root = budget_env
+    key, _key_id = await _make_budgeted_key(factory, budget_limit_cents=100)
+
+    async with await make_client(key) as c:
+        r = await c.post(
+            "/v1/chat/completions",
+            json={"model": "gpt-4o-mini",
+                  "messages": [{"role": "user", "content": "hi"}]},
+        )
+
+    assert r.status_code == 200, r.text
+    fake.acompletion.assert_awaited_once()
+
+
+async def test_unbudgeted_root_key_unaffected(budget_env):
+    make_client, fake, _factory, root = budget_env
+
+    async with await make_client(root) as c:
+        r = await c.post(
+            "/v1/chat/completions",
+            json={"model": "gpt-4o-mini",
+                  "messages": [{"role": "user", "content": "hi"}]},
+        )
+
+    assert r.status_code == 200, r.text
+    fake.acompletion.assert_awaited_once()
+
+
+async def test_create_key_accepts_restrictions(budget_env):
+    make_client, _fake, factory, root = budget_env
+
+    async with await make_client(root) as c:
+        r = await c.post("/v1/keys", json={
+            "name": "team-a",
+            "model_allowlist": ["gpt-4o-mini"],
+            "budget_limit_cents": 500,
+        })
+
+    assert r.status_code == 201, r.text
+    body = r.json()
+    assert body["model_allowlist"] == ["gpt-4o-mini"]
+    assert body["budget_limit_cents"] == 500
+
+    from sqlalchemy import select
+
+    from packages.db.models.api_key import ApiKey
+
+    async with factory() as s:
+        row = (
+            await s.execute(select(ApiKey).where(ApiKey.id == body["id"]))
+        ).scalar_one()
+    assert row.budget_limit_cents == 500
+    assert row.model_allowlist == ["gpt-4o-mini"]
diff --git a/tests/integration/test_keys_authz.py b/tests/integration/test_keys_authz.py
new file mode 100644
index 0000000..8df84eb
--- /dev/null
+++ b/tests/integration/test_keys_authz.py
@@ -0,0 +1,196 @@
+"""Key-management authorization tests.
+
+A restricted key (model_allowlist or budget_limit_cents set) must never be
+able to mint, list, or revoke API keys — otherwise it could mint an
+unrestricted sibling and bypass its own restrictions entirely.
+See issue: restricted-key privilege escalation via POST /v1/keys.
+"""
+
+import pytest
+
+
+@pytest.fixture
+async def seeded_keys(db_session):
+    """Seed the workspace root key plus one restricted and one budgeted key.
+
+    Returns (root_full_key, restricted_full_key, budgeted_full_key).
+    """
+    from app.seed import seed_initial_state
+    from packages.auth.hashing import generate_api_key
+    from packages.db.models.api_key import ApiKey
+
+    seed = await seed_initial_state(db_session)
+    assert seed.api_key is not None
+
+    def _make(**kwargs) -> str:
+        full_key, key_hash, key_prefix = generate_api_key()
+        row = ApiKey(
+            workspace_id="default",
+            name=kwargs.pop("name", "test"),
+            key_hash=key_hash,
+            key_prefix=key_prefix,
+            **kwargs,
+        )
+        db_session.add(row)
+        return full_key
+
+    # flush once so all rows land before any request reads them
+    restricted = _make(name="restricted", model_allowlist=["gpt-4o-mini"])
+    budgeted = _make(name="budgeted", budget_limit_cents=500)
+    await db_session.commit()
+    return seed.api_key, restricted, budgeted
+
+
+@pytest.fixture
+async def keys_app(db_session, monkeypatch):
+    """FastAPI app with auth middleware and only the /v1/keys routes mounted."""
+    monkeypatch.setenv("DATABASE_URL", str(db_session.bind.url))
+    from fastapi import FastAPI
+
+    from app.middleware.auth import AuthMiddleware
+    from packages.db import session as session_mod
+
+    class _PassthroughFactory:
+        async def __aenter__(self):
+            return db_session
+
+        async def __aexit__(self, *exc):
+            return False  # propagate, don't close — fixture owns the session
+
+    monkeypatch.setattr(session_mod, "_session_factory", lambda: _PassthroughFactory())
+
+    from app.routes.keys import router as keys_router
+
+    app = FastAPI()
+    app.add_middleware(AuthMiddleware)
+    app.include_router(keys_router)
+    return app
+
+
+async def _client(app):
+    from httpx import ASGITransport, AsyncClient
+
+    return AsyncClient(transport=ASGITransport(app=app), base_url="http://t")
+
+
+@pytest.mark.parametrize("which", [1, 2], ids=["allowlist-restricted", "budget-restricted"])
+async def test_restricted_key_cannot_create_keys(keys_app, seeded_keys, db_session, which):
+    keys, restricted, budgeted = seeded_keys
+    caller = (restricted, budgeted)[which - 1]
+    async with await _client(keys_app) as c:
+        r = await c.post(
+            "/v1/keys",
+            json={"name": "escalated"},
+            headers={"Authorization": f"Bearer {caller}"},
+        )
+    assert r.status_code == 403
+    # The escalation must not have persisted anything.
+    from sqlalchemy import func, select
+
+    from packages.db.models.api_key import ApiKey
+
+    count = (
+        await db_session.execute(select(func.count()).select_from(ApiKey))
+    ).scalar_one()
+    assert count == 3  # root + restricted + budgeted, nothing new
+
+
+@pytest.mark.parametrize("which", [1, 2], ids=["allowlist-restricted", "budget-restricted"])
+async def test_restricted_key_cannot_list_keys(keys_app, seeded_keys, which):
+    keys, restricted, budgeted = seeded_keys
+    caller = (restricted, budgeted)[which - 1]
+    async with await _client(keys_app) as c:
+        r = await c.get("/v1/keys", headers={"Authorization": f"Bearer {caller}"})
+    assert r.status_code == 403
+
+
+async def test_restricted_key_cannot_revoke_keys(keys_app, seeded_keys, db_session):
+    _, restricted, _budgeted = seeded_keys
+    from sqlalchemy import select
+
+    from packages.db.models.api_key import ApiKey
+
+    rows = (await db_session.execute(select(ApiKey))).scalars().all()
+    target_id = next(r.id for r in rows if r.name == "default")
+    async with await _client(keys_app) as c:
+        r = await c.delete(
+            f"/v1/keys/{target_id}",
+            headers={"Authorization": f"Bearer {restricted}"},
+        )
+    assert r.status_code == 403
+    target = next(r for r in rows if r.name == "default")
+    assert target.is_active  # untouched
+
+
+async def test_unrestricted_key_retains_full_management(keys_app, seeded_keys):
+    root, _restricted, _budgeted = seeded_keys
+    h = {"Authorization": f"Bearer {root}"}
+    async with await _client(keys_app) as c:
+        listed = await c.get("/v1/keys", headers=h)
+        assert listed.status_code == 200
+
+        created = await c.post("/v1/keys", json={"name": "child"}, headers=h)
+        assert created.status_code == 201
+        child_id = created.json()["id"]
+
+        revoked = await c.delete(f"/v1/keys/{child_id}", headers=h)
+        assert revoked.status_code == 204
+
+
+# ── Workspace scoping (IDOR regression tests) ────────────────────────────
+
+
+async def _make_foreign_workspace_key(db_session) -> tuple[str, str]:
+    """A key belonging to a different workspace; returns (id, name)."""
+    from packages.auth.hashing import generate_api_key
+    from packages.db.models.api_key import ApiKey
+    from packages.db.models.workspace import Workspace
+
+    db_session.add(Workspace(id="ws-other", name="Other", slug="other"))
+    await db_session.flush()
+
+    full_key, key_hash, key_prefix = generate_api_key()
+    row = ApiKey(
+        workspace_id="ws-other",
+        name="foreign-key",
+        key_hash=key_hash,
+        key_prefix=key_prefix,
+    )
+    db_session.add(row)
+    await db_session.commit()
+    return row.id, full_key
+
+
+async def test_list_keys_hides_other_workspaces(keys_app, seeded_keys, db_session):
+    root, *_ = seeded_keys
+    foreign_id, _foreign_key = await _make_foreign_workspace_key(db_session)
+
+    async with await _client(keys_app) as c:
+        r = await c.get(
+            "/v1/keys", headers={"Authorization": f"Bearer {root}"}
+        )
+
+    assert r.status_code == 200
+    listed_ids = {k["id"] for k in r.json()["keys"]}
+    assert foreign_id not in listed_ids
+
+
+async def test_revoke_rejects_other_workspaces_key(keys_app, seeded_keys, db_session):
+    from sqlalchemy import select
+
+    from packages.db.models.api_key import ApiKey
+
+    root, *_ = seeded_keys
+    foreign_id, _foreign_key = await _make_foreign_workspace_key(db_session)
+
+    async with await _client(keys_app) as c:
+        r = await c.delete(
+            f"/v1/keys/{foreign_id}",
+            headers={"Authorization": f"Bearer {root}"},
+        )
+
+    assert r.status_code == 404
+    row = (
+        await db_session.execute(select(ApiKey).where(ApiKey.id == foreign_id))
+    ).scalar_one()
+    assert row.is_active  # untouched
diff --git a/tests/unit/test_budget_spend.py b/tests/unit/test_budget_spend.py
new file mode 100644
index 0000000..21899d1
--- /dev/null
+++ b/tests/unit/test_budget_spend.py
@@ -0,0 +1,75 @@
+"""Unit tests for packages.auth.spend — lifetime spend aggregation."""
+
+import pytest
+
+from packages.auth.spend import (
+    MICROCENTS_PER_CENT,
+    budget_exceeded,
+    get_lifetime_spend_microcents,
+)
+
+
+@pytest.fixture
+async def seeded_log(db_session):
+    """Two keys with a mix of billable / failed / soft-deleted rows."""
+    from packages.db.models.api_key import ApiKey
+    from packages.db.models.request_log import RequestLog
+
+    k1 = ApiKey(workspace_id="default", name="a", key_hash="h-a", key_prefix="p-a")
+    k2 = ApiKey(workspace_id="default", name="b", key_hash="h-b", key_prefix="p-b")
+    db_session.add_all([k1, k2])
+    await db_session.flush()
+
+    rows = [
+        RequestLog(
+            workspace_id="default", api_key_id=k1.id, model_requested="m",
+            model_resolved="m", provider="openai", input_tokens=1, output_tokens=1,
+            cost_microcents=1000, status_code=200, routing_strategy="balanced", latency_ms=10, trace_id="t-1",
+        ),
+        RequestLog(
+            workspace_id="default", api_key_id=k1.id, model_requested="m",
+            model_resolved="m", provider="openai", input_tokens=1, output_tokens=1,
+            cost_microcents=500, status_code=200, routing_strategy="balanced", latency_ms=10, trace_id="t-1",
+        ),
+        # failed requests are not billable
+        RequestLog(
+            workspace_id="default", api_key_id=k1.id, model_requested="m",
+            model_resolved="m", provider="openai", input_tokens=9, output_tokens=9,
+            cost_microcents=999_999, status_code=503, routing_strategy="balanced", latency_ms=10, trace_id="t-3",
+        ),
+        # another key's spend must not leak in
+        RequestLog(
+            workspace_id="default", api_key_id=k2.id, model_requested="m",
+            model_resolved="m", provider="openai", input_tokens=2, output_tokens=2,
+            cost_microcents=777_777, status_code=200, routing_strategy="balanced", latency_ms=10, trace_id="t-4",
+        ),
+    ]
+    db_session.add_all(rows)
+    await db_session.commit()
+    return k1, k2
+
+
+async def test_spend_sums_only_billable_rows_for_the_key(db_session, seeded_log):
+    k1, _k2 = seeded_log
+    spend = await get_lifetime_spend_microcents(db_session, k1.id)
+    assert spend == 1500
+
+
+async def test_empty_history_is_zero(db_session, seeded_log):
+    _k1, k2 = seeded_log
+    from packages.db.models.api_key import ApiKey
+
+    fresh = ApiKey(workspace_id="default", name="c", key_hash="h-c", key_prefix="p-c")
+    db_session.add(fresh)
+    await db_session.commit()
+    assert await get_lifetime_spend_microcents(db_session, fresh.id) == 0
+
+
+def test_budget_exceeded_boundary():
+    assert budget_exceeded(10_000 - 1, 1) is False  # just under 1 cent
+    assert budget_exceeded(10_000, 1) is True       # exactly at the cap blocks
+    assert budget_exceeded(0, 1) is False
+
+
+def test_microcent_conversion_constant():
+    assert MICROCENTS_PER_CENT == 10_000