phase 1.7: rate-limit enforcement (in-memory sliding window, 429 + Retry-After)

publishers can now declare rate_limit in their manifest, and the hub
enforces it on /v1/chat/completions per api-key. exceeded quota returns
429 with Retry-After header + JSON body {error: {code, message,
retry_after}}. lost on hub restart, fine for v0.

components:
- zhub/ratelimit.py: parse_rate("60/min" -> (60, 60.0)), supports
  s/min/hour/day units, falls back to (60, 60.0) on None/empty/garbage.
  SlidingWindow class with per-key deque of timestamps, drops expired
  on check, returns (allowed, retry_after_or_None). injectable now_fn
  for deterministic testing without sleep.
- zhub/server.py:
    - Hub gains _rate_windows dict (ai_name -> SlidingWindow), lazily
      built from publisher.manifest.rate_limit on first check
    - Hub.check_rate_limit(ai_name, api_key) -> (bool, retry_after)
    - chat_completions handler calls check_rate_limit, returns 429 +
      Retry-After header before falling through to proxy_chat
- zhub/manifest.py: chat_only_manifest gains rate_limit kw (default
  "60/min"), passes through to Manifest()
- zhub/client.py: publish() gains rate_limit parameter, threads to
  manifest. legacy callers without rate_limit get the default.

tests:
- tests/test_rate_limit.py: 12 cases
    * 6 unit: parse_rate covers s/min/hour/day, default fallback,
      garbage fallback
    * 4 unit: SlidingWindow under-limit, at-limit, expiry, per-key
      isolation
    * 2 e2e: 429 + Retry-After after quota, per-publisher isolation
      (one publisher's 429 doesn't affect another)

result: 47/47 -> 49/49 pytest passing. existing flow unchanged for
publishers using the default 60/min.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Zawwarsami16

tests/test_rate_limit.py

@@ -0,0 +1,186 @@
+"""Rate-limit parsing + sliding-window counter + e2e enforcement."""
+
+import asyncio
+import socket
+import threading
+import time
+
+import pytest
+
+from zhub.ratelimit import parse_rate, SlidingWindow
+
+
+def test_parse_rate_per_second():
+    assert parse_rate("10/s") == (10, 1.0)
+
+
+def test_parse_rate_per_minute():
+    assert parse_rate("60/min") == (60, 60.0)
+
+
+def test_parse_rate_per_hour():
+    assert parse_rate("1000/hour") == (1000, 3600.0)
+
+
+def test_parse_rate_per_day():
+    assert parse_rate("100000/day") == (100000, 86400.0)
+
+
+def test_parse_rate_default_when_unset():
+    assert parse_rate(None) == (60, 60.0)
+    assert parse_rate("") == (60, 60.0)
+
+
+def test_parse_rate_garbage_falls_back_to_default():
+    assert parse_rate("not-a-rate") == (60, 60.0)
+    assert parse_rate("60") == (60, 60.0)
+    assert parse_rate("/min") == (60, 60.0)
+
+
+def test_sliding_window_under_limit_allows():
+    clock = [0.0]
+    w = SlidingWindow(limit=3, period_seconds=10.0, now_fn=lambda: clock[0])
+    assert w.check("k") == (True, None)
+    clock[0] = 1.0
+    assert w.check("k") == (True, None)
+    clock[0] = 2.0
+    assert w.check("k") == (True, None)
+
+
+def test_sliding_window_at_limit_denies():
+    clock = [0.0]
+    w = SlidingWindow(limit=3, period_seconds=10.0, now_fn=lambda: clock[0])
+    for _ in range(3):
+        ok, _ = w.check("k")
+        assert ok
+    ok, retry_after = w.check("k")
+    assert ok is False
+    assert retry_after is not None and retry_after > 0
+
+
+def test_sliding_window_expires_old_hits():
+    clock = [0.0]
+    w = SlidingWindow(limit=2, period_seconds=10.0, now_fn=lambda: clock[0])
+    assert w.check("k")[0] is True
+    assert w.check("k")[0] is True
+    assert w.check("k")[0] is False  # at limit
+    clock[0] = 11.0  # past the 10s window
+    assert w.check("k")[0] is True
+
+
+def test_sliding_window_per_key_isolation():
+    clock = [0.0]
+    w = SlidingWindow(limit=1, period_seconds=10.0, now_fn=lambda: clock[0])
+    assert w.check("a")[0] is True
+    assert w.check("b")[0] is True  # different key, different bucket
+    assert w.check("a")[0] is False  # a hit limit
+
+
+# ---- e2e enforcement ----
+
+try:
+    import fastapi  # noqa
+    import uvicorn  # noqa
+    SERVER_AVAILABLE = True
+except ImportError:
+    SERVER_AVAILABLE = False
+
+if SERVER_AVAILABLE:
+    from zhub.server import create_app
+from zhub import publish, Manifest, Capability
+
+
+def _free_port() -> int:
+    with socket.socket() as s:
+        s.bind(("", 0))
+        return s.getsockname()[1]
+
+
+@pytest.fixture(scope="module")
+def hub_port():
+    if not SERVER_AVAILABLE:
+        pytest.skip("fastapi/uvicorn not installed")
+    port = _free_port()
+
+    def run():
+        config = uvicorn.Config(create_app(), host="127.0.0.1", port=port, log_level="warning")
+        asyncio.run(uvicorn.Server(config).serve())
+
+    threading.Thread(target=run, daemon=True).start()
+    for _ in range(30):
+        try:
+            with socket.create_connection(("127.0.0.1", port), timeout=0.1):
+                break
+        except OSError:
+            time.sleep(0.1)
+    yield port
+
+
+@pytest.mark.asyncio
+async def test_rate_limit_429_after_quota(hub_port):
+    """Publisher with 3/s rate limit. Hit chat endpoint 4x rapidly. First 3
+    succeed; 4th returns 429 with Retry-After header + body."""
+    if not SERVER_AVAILABLE:
+        pytest.skip()
+
+    pub = publish(
+        name="rl",
+        description="rate limit test",
+        chat_handler=lambda m, o: "ok",
+        hub_url=f"ws://127.0.0.1:{hub_port}",
+        rate_limit="3/s",
+    )
+    for _ in range(50):
+        if pub.api_key:
+            break
+        await asyncio.sleep(0.05)
+    assert pub.api_key
+
+    import httpx
+    base = f"http://127.0.0.1:{hub_port}/rl/v1/chat/completions"
+    headers = {"Authorization": f"Bearer {pub.api_key}", "Content-Type": "application/json"}
+    body = {"messages": [{"role": "user", "content": "hi"}]}
+
+    async with httpx.AsyncClient() as c:
+        # First 3 should succeed
+        for i in range(3):
+            r = await c.post(base, json=body, headers=headers)
+            assert r.status_code == 200, f"call #{i+1}: {r.status_code} {r.text}"
+        # 4th should be rate-limited
+        r4 = await c.post(base, json=body, headers=headers)
+        assert r4.status_code == 429, f"expected 429, got {r4.status_code}: {r4.text}"
+        assert "Retry-After" in r4.headers, f"missing Retry-After header: {dict(r4.headers)}"
+        body_json = r4.json()
+        assert body_json["error"]["code"] == "rate_limited"
+
+
+@pytest.mark.asyncio
+async def test_rate_limit_per_publisher_isolation(hub_port):
+    """Two publishers with separate rate limits don't interfere with each other."""
+    if not SERVER_AVAILABLE:
+        pytest.skip()
+
+    pub_a = publish(name="rla", description="a", chat_handler=lambda m, o: "a",
+                    hub_url=f"ws://127.0.0.1:{hub_port}", rate_limit="2/s")
+    pub_b = publish(name="rlb", description="b", chat_handler=lambda m, o: "b",
+                    hub_url=f"ws://127.0.0.1:{hub_port}", rate_limit="2/s")
+    for _ in range(50):
+        if pub_a.api_key and pub_b.api_key:
+            break
+        await asyncio.sleep(0.05)
+
+    import httpx
+    body = {"messages": [{"role": "user", "content": "hi"}]}
+    async with httpx.AsyncClient() as c:
+        # Exhaust A's quota
+        for _ in range(2):
+            r = await c.post(f"http://127.0.0.1:{hub_port}/rla/v1/chat/completions",
+                             json=body, headers={"Authorization": f"Bearer {pub_a.api_key}"})
+            assert r.status_code == 200
+        r = await c.post(f"http://127.0.0.1:{hub_port}/rla/v1/chat/completions",
+                         json=body, headers={"Authorization": f"Bearer {pub_a.api_key}"})
+        assert r.status_code == 429
+        # B is unaffected
+        r = await c.post(f"http://127.0.0.1:{hub_port}/rlb/v1/chat/completions",
+                         json=body, headers={"Authorization": f"Bearer {pub_b.api_key}"})
+        assert r.status_code == 200

zhub/client.py

@@ -118,6 +118,7 @@ def publish(
     on_connection_event: Optional[ConnectionEventHandler] = None,
     api_key: Optional[str] = None,
     private_key: Optional[str] = None,
+    rate_limit: str = "60/min",
 ) -> ZhubPublication:
     """Create a ZhubPublication. Call .run_forever() to actually start serving.
 
@@ -134,6 +135,7 @@ def publish(
     manifest = chat_only_manifest(
         name=name, description=description,
         operator=operator, contact=contact, public=public,
+        rate_limit=rate_limit,
     )
     if capabilities:
         manifest.capabilities.extend(capabilities)

zhub/manifest.py

@@ -87,12 +87,14 @@ def chat_only_manifest(
     operator: str = "",
     contact: str = "",
     public: bool = False,
+    rate_limit: str = "60/min",
 ) -> Manifest:
     """The simplest possible manifest — an AI that only does chat."""
     return Manifest(
         name=name,
         description=description,
         accepts="openai-v1-chat-completions",
+        rate_limit=rate_limit,
         capabilities=[
             Capability(
                 name="chat",

zhub/ratelimit.py

@@ -0,0 +1,82 @@
+"""Rate-limit primitives — string parser + sliding-window counter.
+
+The hub uses these to enforce per-api-key request limits parsed from a
+publisher's manifest.rate_limit field. In-memory, lost on restart, fine
+for v0 — clients retry, no persistent damage.
+"""
+
+from __future__ import annotations
+
+import re
+import time
+from collections import deque
+from typing import Callable, Optional
+
+
+_RATE_RE = re.compile(r"^\s*(\d+)\s*/\s*(s|sec|second|m|min|minute|h|hour|d|day)\s*$", re.I)
+_UNIT_SECONDS = {
+    "s": 1.0, "sec": 1.0, "second": 1.0,
+    "m": 60.0, "min": 60.0, "minute": 60.0,
+    "h": 3600.0, "hour": 3600.0,
+    "d": 86400.0, "day": 86400.0,
+}
+
+DEFAULT_RATE = (60, 60.0)
+
+
+def parse_rate(text: Optional[str]) -> tuple[int, float]:
+    """Parse a rate string like '60/min' into (limit, period_seconds).
+
+    Falls back to (60, 60.0) for None, empty, or malformed input — the hub
+    needs a default rather than refusing service when a publisher omits the
+    field. Operators who want unmetered access publish with a deliberately
+    high rate (e.g., '1000000/hour').
+    """
+    if not text:
+        return DEFAULT_RATE
+    m = _RATE_RE.match(text)
+    if not m:
+        return DEFAULT_RATE
+    n = int(m.group(1))
+    unit = m.group(2).lower()
+    return n, _UNIT_SECONDS.get(unit, 60.0)
+
+
+class SlidingWindow:
+    """Per-key sliding-window counter.
+
+    Each key maps to a deque of timestamps. On check(), expired timestamps
+    are dropped first; if the live count is under the limit, the new hit is
+    recorded and check returns (True, None). Otherwise check returns
+    (False, retry_after_seconds_until_oldest_hit_expires).
+    """
+
+    def __init__(
+        self,
+        limit: int,
+        period_seconds: float,
+        now_fn: Callable[[], float] = time.monotonic,
+    ) -> None:
+        self.limit = limit
+        self.period = period_seconds
+        self._now = now_fn
+        self._buckets: dict[str, deque[float]] = {}
+
+    def check(self, key: str) -> tuple[bool, Optional[float]]:
+        now = self._now()
+        cutoff = now - self.period
+        bucket = self._buckets.get(key)
+        if bucket is None:
+            bucket = deque()
+            self._buckets[key] = bucket
+        while bucket and bucket[0] <= cutoff:
+            bucket.popleft()
+        if len(bucket) < self.limit:
+            bucket.append(now)
+            return True, None
+        oldest = bucket[0]
+        retry_after = max(0.0, oldest + self.period - now)
+        return False, retry_after
+
+    def clear(self, key: str) -> None:
+        self._buckets.pop(key, None)

zhub/server.py

@@ -39,6 +39,7 @@
     ) from e
 
 from .persistence import Storage, hash_key
+from .ratelimit import parse_rate, SlidingWindow
 from .protocol import (
     Envelope, registered, chat_request, chat_response,
     invoke_request, invoke_result, connection_event, error_envelope, new_request_id,
@@ -83,6 +84,22 @@ def __init__(self, storage: Optional[Storage] = None) -> None:
         # ws_connect-side chat-requests. Publisher emits chat-chunk back; we
         # route by request_id to the originating client.
         self.client_routes: dict[str, tuple[WebSocket, str]] = {}
+        # Rate-limit windows per AI. Each AI gets its own SlidingWindow
+        # configured from the publisher's manifest.rate_limit. Key into the
+        # window is the api_key string used by the caller.
+        self._rate_windows: dict[str, SlidingWindow] = {}
+
+    def check_rate_limit(self, ai_name: str, api_key: str) -> tuple[bool, Optional[float]]:
+        """Returns (allowed, retry_after_seconds_or_None)."""
+        window = self._rate_windows.get(ai_name)
+        if window is None:
+            publisher = self.publishers.get(ai_name)
+            if publisher is None:
+                return True, None  # AI offline; chat will 404 later
+            limit, period = parse_rate(publisher.manifest.get("rate_limit"))
+            window = SlidingWindow(limit=limit, period_seconds=period)
+            self._rate_windows[ai_name] = window
+        return window.check(api_key)
 
     # publishers --------------------------------------------------------
 
@@ -379,6 +396,22 @@ async def chat_completions(ai_name: str, request: Request):
         if hub.lookup_by_api_key(api_key_header) != ai_name:
             raise HTTPException(401, "invalid api key for this AI")
 
+        # Rate-limit enforcement (Phase 1.7)
+        rl_ok, retry_after = hub.check_rate_limit(ai_name, api_key_header)
+        if not rl_ok:
+            ra = max(1, int(round(retry_after or 1.0)))
+            return JSONResponse(
+                status_code=429,
+                content={
+                    "error": {
+                        "code": "rate_limited",
+                        "message": "rate limit exceeded for this api key",
+                        "retry_after": ra,
+                    },
+                },
+                headers={"Retry-After": str(ra)},
+            )
+
         messages = body.get("messages", [])
         model = body.get("model", "default")
         temperature = float(body.get("temperature", 0.4))