Add user_id to routing decisions for per-user history isolation (#18)

* Add user_id to routing decisions for per-user history isolation

Migration 0003 adds a nullable user_id column to calls. CallStore.record(),
recent(), and count() accept an optional user_id parameter; recent/count
filter to that user when provided. RoutingService.route() and the HTTP
RouteRequest body thread user_id through; GET /api/history accepts a
user_id query param. All fields are optional for full backward compat.

* Address review: index on user_id, remove user_id from public history API

Add composite index (user_id, id) to migration 0003 and mark ORM column
index=True. Move WHERE before ORDER BY/LIMIT/OFFSET in recent(). Remove
user_id query param from GET /api/history (BOLA risk — per-user scoping
is the embedding app's responsibility via CallStore.recent(user_id=...)).

* Fix ORM/migration index mismatch; update user_id field description

Replace index=True on user_id column with __table_args__ composite index
sa.Index("ix_calls_user_id_id", "user_id", "id") to match migration 0003
exactly and prevent Alembic autogenerate from reporting false drift.
Reword user_id field description: no longer claims history filtering.

Author: qinxuye

src/xrouter_llm/migrations/versions/0003_add_user_id.py

@@ -0,0 +1,25 @@
+"""Add user_id column to calls table
+
+Revision ID: 0003
+Revises: 0002
+Create Date: 2026-06-28
+"""
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0003"
+down_revision: str = "0002"
+branch_labels: str | None = None
+depends_on: str | None = None
+
+
+def upgrade() -> None:
+    op.add_column("calls", sa.Column("user_id", sa.String(255), nullable=True))
+    op.create_index("ix_calls_user_id_id", "calls", ["user_id", "id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_calls_user_id_id", table_name="calls")
+    op.drop_column("calls", "user_id")

src/xrouter_llm/server.py

@@ -44,6 +44,7 @@ class RouteRequest(BaseModel):
         description="Candidate model IDs. Omit to use config models or all registered models.",
     )
     task: str | None = None
+    user_id: str | None = Field(default=None, max_length=255, description="Caller identity persisted with the routing decision.")
     completion_threshold: float | None = Field(default=None, ge=0.0, le=1.0)
     lambda_cost: float | None = Field(default=None, ge=0.0)
     lambda_latency: float | None = Field(default=None, ge=0.0)
@@ -96,6 +97,7 @@ def route(req: RouteRequest) -> dict[str, Any]:
                 lambda_latency=req.lambda_latency,
                 max_k=req.max_k,
                 fallback_quality_margin=req.fallback_quality_margin,
+                user_id=req.user_id,
             )
         except ValueError as exc:
             raise HTTPException(status_code=400, detail=str(exc)) from exc

src/xrouter_llm/serving.py

@@ -148,6 +148,7 @@ def route(
         lambda_latency: float | None = None,
         max_k: int | None = None,
         fallback_quality_margin: float | None = None,
+        user_id: str | None = None,
     ) -> dict[str, Any]:
         if not prompt.strip():
             raise ValueError("prompt must not be empty")
@@ -241,6 +242,7 @@ def route(
             expected_quality=float(breakdown.expected_quality),
             cost=float(breakdown.cost),
             latency=float(breakdown.latency),
+            user_id=user_id,
         )
         return {
             "id": call_id,

src/xrouter_llm/store.py

@@ -34,13 +34,19 @@ class CallRecord(Base):
     cost: Mapped[float | None] = mapped_column(sa.Float, nullable=True)
     latency: Mapped[float | None] = mapped_column(sa.Float, nullable=True)
     feedback: Mapped[Any] = mapped_column(sa.JSON, nullable=True)
+    user_id: Mapped[str | None] = mapped_column(sa.String(255), nullable=True)
+
+    __table_args__ = (
+        sa.Index("ix_calls_user_id_id", "user_id", "id"),
+    )
 
 
 _BASELINE_REVISION = "0001"
 # Each entry: (column added by that migration, revision). Keep in ascending order.
 # To add a new migration: append ("new_column", "000N") here.
 _SCHEMA_CHECKPOINTS: list[tuple[str, str]] = [
     ("feedback", "0002"),
+    ("user_id", "0003"),
 ]
 
 
@@ -141,6 +147,7 @@ def record(
         expected_quality: float,
         cost: float,
         latency: float,
+        user_id: str | None = None,
     ) -> int:
         with self._Session() as session:
             row = CallRecord(
@@ -153,32 +160,33 @@ def record(
                 expected_quality=expected_quality,
                 cost=cost,
                 latency=latency,
+                user_id=user_id,
             )
             session.add(row)
             session.flush()   # INSERT → DB assigns id; no extra SELECT needed
             call_id = row.id
             session.commit()
             return call_id
 
-    def recent(self, limit: int = 50, offset: int = 0) -> list[dict[str, Any]]:
+    def recent(
+        self, limit: int = 50, offset: int = 0, *, user_id: str | None = None
+    ) -> list[dict[str, Any]]:
         limit = max(1, min(int(limit), 1000))
         offset = max(0, int(offset))
+        stmt = sa.select(CallRecord)
+        if user_id is not None:
+            stmt = stmt.where(CallRecord.user_id == user_id)
+        stmt = stmt.order_by(CallRecord.id.desc()).limit(limit).offset(offset)
         with self._Session() as session:
-            rows = (
-                session.execute(
-                    sa.select(CallRecord)
-                    .order_by(CallRecord.id.desc())
-                    .limit(limit)
-                    .offset(offset)
-                )
-                .scalars()
-                .all()
-            )
+            rows = session.execute(stmt).scalars().all()
         return [_row_to_dict(r) for r in rows]
 
-    def count(self) -> int:
+    def count(self, *, user_id: str | None = None) -> int:
+        stmt = sa.select(sa.func.count(CallRecord.id))
+        if user_id is not None:
+            stmt = stmt.where(CallRecord.user_id == user_id)
         with self._Session() as session:
-            return session.execute(sa.select(sa.func.count(CallRecord.id))).scalar_one()
+            return session.execute(stmt).scalar_one()
 
     def delete(self, call_id: int) -> bool:
         with self._Session() as session:
@@ -225,4 +233,5 @@ def _row_to_dict(r: CallRecord) -> dict[str, Any]:
         "cost": r.cost,
         "latency": r.latency,
         "feedback": r.feedback,
+        "user_id": r.user_id,
     }

tests/test_serving.py

@@ -217,6 +217,27 @@ def test_feedback_endpoint(tmp_path) -> None:
                         json={"outcome": "good", "correct_model": "strong"}).status_code == 422
 
 
+def test_user_id_recorded_in_route(tmp_path) -> None:
+    from xrouter_llm.server import create_app
+
+    service = _service(tmp_path)
+    client = TestClient(create_app(service))
+
+    client.post("/api/route", json={"prompt": "hello", "models": ["cheap", "strong"], "user_id": "alice"})
+    client.post("/api/route", json={"prompt": "world", "models": ["cheap", "strong"], "user_id": "bob"})
+    client.post("/api/route", json={"prompt": "anon",  "models": ["cheap", "strong"]})
+
+    # /api/history is admin-only (no user filter); per-user scoping is done
+    # by the embedding app via CallStore.recent(user_id=...) directly.
+    assert client.get("/api/history").json()["total"] == 3
+
+    # user_id is persisted and accessible via store
+    assert service.store.count(user_id="alice") == 1
+    assert service.store.count(user_id="bob") == 1
+    alice_rows = service.store.recent(user_id="alice")
+    assert alice_rows[0]["user_id"] == "alice"
+
+
 def test_history_pagination(tmp_path) -> None:
     from xrouter_llm.server import create_app
 

tests/test_store.py

@@ -132,16 +132,17 @@ def _legacy_db_empty_version(tmp_path):
 
 
 def test_legacy_db_no_alembic_version(tmp_path) -> None:
-    """CallStore opens a pre-Alembic DB, runs pending migrations (adds feedback column)."""
+    """CallStore opens a pre-Alembic DB, runs pending migrations (adds feedback + user_id columns)."""
     url = _legacy_db(tmp_path)
     store = CallStore(url)
     store.record(
         ts=1.0, config="all", prompt="hello", task=None,
         selected=["m"], candidates=[], expected_quality=0.8, cost=0.0, latency=0.0,
     )
     assert store.count() == 1
-    # feedback column must exist after migration 0002 runs
-    assert store.recent()[0]["feedback"] is None
+    row = store.recent()[0]
+    assert row["feedback"] is None
+    assert row["user_id"] is None
 
 
 def test_legacy_db_empty_alembic_version(tmp_path) -> None:
@@ -174,6 +175,35 @@ def test_in_memory_sqlite_works() -> None:
     assert store.recent()[0]["prompt"] == "hello"
 
 
+def test_user_id_record_and_filter(store) -> None:
+    store.record(
+        ts=1.0, config="all", prompt="p1", task=None,
+        selected=["m"], candidates=[], expected_quality=0.8, cost=0.0, latency=0.0,
+        user_id="alice",
+    )
+    store.record(
+        ts=2.0, config="all", prompt="p2", task=None,
+        selected=["m"], candidates=[], expected_quality=0.8, cost=0.0, latency=0.0,
+        user_id="bob",
+    )
+    store.record(
+        ts=3.0, config="all", prompt="p3", task=None,
+        selected=["m"], candidates=[], expected_quality=0.8, cost=0.0, latency=0.0,
+    )
+    assert store.count() == 3
+    assert store.count(user_id="alice") == 1
+    assert store.count(user_id="bob") == 1
+
+    alice_rows = store.recent(user_id="alice")
+    assert len(alice_rows) == 1
+    assert alice_rows[0]["user_id"] == "alice"
+    assert alice_rows[0]["prompt"] == "p1"
+
+    # anonymous call has user_id None in the response
+    anon_rows = [r for r in store.recent() if r["user_id"] is None]
+    assert len(anon_rows) == 1
+
+
 def test_auto_migrate_false_skips_migration(tmp_path) -> None:
     """auto_migrate=False does not run migrations (table absent → OperationalError on first use)."""
     import pytest