PC3 Phase 1.3: backend avatar loadout persistence

New avatar_loadouts table (one row per user, loadout stored as a single
JSONB document matching the client's AvatarLoadout.to_json() shape) and
GET/PUT /api/avatar/loadout.

The server has no copy of the asset catalog, so it validates only the
envelope (JSON object, non-empty base_body, object slots, size cap) --
which-asset-in-which-slot validation stays client-side. New accounts get a
server default (base_human_male_01, no slots) with is_default:true so the
client can tell "never customised" from "customised back to base".

- models/avatar_loadout.py + 0004 migration + User.avatar_loadout relation
- routers/avatar.py wired into main.py
- tests/test_avatar_loadout.py (7 tests: default, save/read, replace,
  validation, auth) -- full suite 68 passed
- lifted the duplicated _override_db fixture from 3 test files into
  conftest.py (new test files need it too)
- docs/architecture/avatar.md §9a: the persistence design

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015jtxAiu5dAW4WxCNdGkrkJ

backend/alembic/env.py

@@ -8,7 +8,7 @@ from sqlalchemy.ext.asyncio import create_async_engine
 
 from app.config import settings
 from app.database import Base
-from app.models import user, session, world, asset  # noqa: F401
+from app.models import user, session, world, asset, avatar_loadout  # noqa: F401
 
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.

backend/alembic/versions/0004_add_avatar_loadouts_table.py

@@ -0,0 +1,39 @@
+"""add avatar_loadouts table
+
+Revision ID: 0004_add_avatar_loadouts_table
+Revises: 0003_add_refresh_tokens_table
+Create Date: 2026-09-05 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "0004_add_avatar_loadouts_table"
+down_revision: Union[str, None] = "0003_add_refresh_tokens_table"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Create the avatar_loadouts table (one row per user)."""
+    op.create_table(
+        "avatar_loadouts",
+        sa.Column(
+            "user_id",
+            sa.Integer(),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            primary_key=True,
+        ),
+        sa.Column("data", postgresql.JSONB(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+    )
+
+
+def downgrade() -> None:
+    """Drop the avatar_loadouts table."""
+    op.drop_table("avatar_loadouts")

backend/app/main.py

@@ -6,7 +6,7 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
 from app.config import settings
-from app.routers import auth, users, worlds, assets
+from app.routers import auth, users, worlds, assets, avatar
 
 
 @asynccontextmanager
@@ -38,6 +38,7 @@ app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
 app.include_router(users.router, prefix="/api/users", tags=["users"])
 app.include_router(worlds.router, prefix="/api/worlds", tags=["worlds"])
 app.include_router(assets.router, prefix="/api/assets", tags=["assets"])
+app.include_router(avatar.router, prefix="/api/avatar", tags=["avatar"])
 
 
 @app.get("/api/health")

backend/app/models/__init__.py

@@ -2,5 +2,6 @@ from app.models.user import User
 from app.models.session import Session
 from app.models.world import World
 from app.models.asset import Asset
+from app.models.avatar_loadout import AvatarLoadout
 
-__all__ = ["User", "Session", "World", "Asset"]
+__all__ = ["User", "Session", "World", "Asset", "AvatarLoadout"]

backend/app/models/avatar_loadout.py

@@ -0,0 +1,41 @@
+from sqlalchemy import Column, Integer, ForeignKey, DateTime
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import relationship
+from sqlalchemy.sql import func
+
+from app.database import Base
+
+
+# The default a brand-new account gets before they touch the customization
+# screen. Kept in sync with assets/avatars/catalog.json (the base_body id) and
+# with the client's AvatarLoadout.to_json() shape.
+DEFAULT_LOADOUT: dict = {
+    "base_body": "base_human_male_01",
+    "slots": {},
+}
+
+
+class AvatarLoadout(Base):
+    """A user's equipped avatar composition.
+
+    One row per user. The whole loadout is stored as a single JSON document
+    matching the client's ``AvatarLoadout.to_json()`` shape
+    (``{"base_body": str, "slots": {...}, "skin_tone"?: {...}}``) rather than
+    exploded into columns -- the server has no copy of the asset catalog, so
+    deep validation (does this asset exist, is it valid for this slot) stays
+    on the client. The server only enforces the envelope: valid JSON object
+    with a non-empty ``base_body``.
+    """
+
+    __tablename__ = "avatar_loadouts"
+
+    user_id = Column(
+        Integer,
+        ForeignKey("users.id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    data = Column(JSONB, nullable=False)
+    created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
+
+    user = relationship("User", back_populates="avatar_loadout")

backend/app/models/user.py

@@ -24,6 +24,12 @@ class User(Base):
     refresh_tokens = relationship("RefreshToken", back_populates="user", cascade="all, delete-orphan")
     worlds = relationship("World", back_populates="owner")
     assets = relationship("Asset", back_populates="owner")
+    avatar_loadout = relationship(
+        "AvatarLoadout",
+        back_populates="user",
+        uselist=False,
+        cascade="all, delete-orphan",
+    )
 
     @property
     def is_adult(self) -> bool:

backend/app/routers/avatar.py

@@ -0,0 +1,78 @@
+"""Avatar router - the current user's persisted avatar loadout.
+
+The loadout is a single JSON document matching the client's
+``AvatarLoadout.to_json()`` shape. The server has no copy of the asset
+catalog, so it only validates the envelope (JSON object, non-empty
+``base_body``); which assets are valid for which slot is the client's job.
+"""
+
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel, Field
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.database import get_db
+from app.dependencies import CurrentUser
+from app.models.avatar_loadout import AvatarLoadout, DEFAULT_LOADOUT
+
+router = APIRouter()
+
+DbDep = Annotated[AsyncSession, Depends(get_db)]
+
+# A loadout document is small; reject anything absurd outright.
+MAX_LOADOUT_BYTES = 16 * 1024
+
+
+class LoadoutResponse(BaseModel):
+    data: dict[str, Any]
+    is_default: bool = Field(
+        description="True when the user has never saved a loadout and this is "
+        "the server default."
+    )
+
+
+class LoadoutUpdate(BaseModel):
+    data: dict[str, Any]
+
+
+def _validate_envelope(data: dict[str, Any]) -> None:
+    base_body = data.get("base_body")
+    if not isinstance(base_body, str) or not base_body.strip():
+        raise HTTPException(
+            status_code=422,
+            detail="loadout.data must have a non-empty string 'base_body'",
+        )
+    slots = data.get("slots", {})
+    if not isinstance(slots, dict):
+        raise HTTPException(status_code=422, detail="loadout.data.slots must be an object")
+    if len(str(data)) > MAX_LOADOUT_BYTES:
+        raise HTTPException(status_code=413, detail="loadout document too large")
+
+
+@router.get("/loadout", response_model=LoadoutResponse)
+async def get_loadout(current_user: CurrentUser, db: DbDep):
+    """Return the current user's saved loadout, or the server default if they
+    have never saved one."""
+    row = await db.get(AvatarLoadout, current_user.id)
+    if row is None:
+        return LoadoutResponse(data=dict(DEFAULT_LOADOUT), is_default=True)
+    return LoadoutResponse(data=row.data, is_default=False)
+
+
+@router.put("/loadout", response_model=LoadoutResponse)
+async def put_loadout(payload: LoadoutUpdate, current_user: CurrentUser, db: DbDep):
+    """Replace the current user's loadout."""
+    _validate_envelope(payload.data)
+
+    row = await db.get(AvatarLoadout, current_user.id)
+    if row is None:
+        row = AvatarLoadout(user_id=current_user.id, data=payload.data)
+        db.add(row)
+    else:
+        row.data = payload.data
+
+    await db.commit()
+    await db.refresh(row)
+    return LoadoutResponse(data=row.data, is_default=False)

backend/tests/conftest.py

@@ -15,11 +15,14 @@ import asyncio
 
 import pytest
 from sqlalchemy import text
-from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy.pool import NullPool
 
 from app.config import settings
+from app.database import get_db
+from app.main import app
 
-TEST_TABLES = ("assets", "worlds", "sessions", "refresh_tokens", "users")
+TEST_TABLES = ("avatar_loadouts", "assets", "worlds", "sessions", "refresh_tokens", "users")
 
 
 @pytest.fixture(scope="session", autouse=True)
@@ -36,3 +39,38 @@ def _reset_database():
 
     asyncio.run(_truncate())
     yield
+
+
+@pytest.fixture(autouse=True)
+def _override_db():
+    """Override get_db so the app uses a fresh session per test.
+
+    Deliberately does NOT reuse app.database's shared engine: FastAPI's
+    (sync) TestClient runs each request through anyio's blocking-portal
+    machinery, which can hand different requests different event loops.
+    asyncpg connections are bound to the loop they were created on, so
+    reusing one from the shared pool across requests raises "attached to a
+    different loop" the moment a test makes more than one request. NullPool
+    sidesteps it: every checkout is a brand-new connection.
+
+    This lived (identically) in three separate test files before it was
+    lifted here; any file still defining its own local `_override_db`
+    simply shadows this one with the same behaviour.
+    """
+    test_engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
+    test_session_factory = async_sessionmaker(test_engine, expire_on_commit=False)
+
+    async def _get_db():
+        async_session = test_session_factory()
+        try:
+            yield async_session
+            await async_session.commit()
+        except Exception:
+            await async_session.rollback()
+            raise
+        finally:
+            await async_session.close()
+
+    app.dependency_overrides[get_db] = _get_db
+    yield
+    app.dependency_overrides.clear()

backend/tests/test_avatar_loadout.py

@@ -0,0 +1,91 @@
+"""Integration tests for GET/PUT /api/avatar/loadout.
+
+Hits the real app against the configured Postgres (same contract as the
+other integration tests -- migrations applied first).
+"""
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.main import app
+from app.models.avatar_loadout import DEFAULT_LOADOUT
+
+
+def _auth_headers(client: TestClient, username: str) -> dict:
+    resp = client.post(
+        "/api/auth/register",
+        json={
+            "username": username,
+            "email": f"{username}@example.com",
+            "password": "secret123",
+            "over_18": True,
+        },
+    )
+    assert resp.status_code == 201, resp.text
+    return {"Authorization": f"Bearer {resp.json()['access_token']}"}
+
+
+class TestGetLoadout:
+    def test_new_user_gets_the_default(self):
+        client = TestClient(app)
+        headers = _auth_headers(client, "av_default")
+        resp = client.get("/api/avatar/loadout", headers=headers)
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["is_default"] is True
+        assert body["data"] == DEFAULT_LOADOUT
+
+    def test_requires_auth(self):
+        client = TestClient(app)
+        assert client.get("/api/avatar/loadout").status_code == 401
+
+
+class TestPutLoadout:
+    def test_save_then_read_back(self):
+        client = TestClient(app)
+        headers = _auth_headers(client, "av_save")
+        loadout = {
+            "base_body": "base_human_male_01",
+            "slots": {"hair": "hair_mohawk_01", "accessory": ["accessory_belt_01"]},
+            "skin_tone": {"r": 0.8, "g": 0.6, "b": 0.5, "a": 1.0},
+        }
+        put = client.put("/api/avatar/loadout", headers=headers, json={"data": loadout})
+        assert put.status_code == 200, put.text
+        assert put.json()["is_default"] is False
+
+        got = client.get("/api/avatar/loadout", headers=headers)
+        assert got.status_code == 200
+        assert got.json()["is_default"] is False
+        assert got.json()["data"] == loadout
+
+    def test_put_is_idempotent_replace(self):
+        client = TestClient(app)
+        headers = _auth_headers(client, "av_replace")
+        client.put(
+            "/api/avatar/loadout", headers=headers,
+            json={"data": {"base_body": "base_human_male_01", "slots": {"hair": "hair_mohawk_01"}}},
+        )
+        second = {"base_body": "base_human_male_01", "slots": {"chest": "chest_bandit_01"}}
+        client.put("/api/avatar/loadout", headers=headers, json={"data": second})
+        got = client.get("/api/avatar/loadout", headers=headers)
+        assert got.json()["data"] == second
+
+    def test_rejects_missing_base_body(self):
+        client = TestClient(app)
+        headers = _auth_headers(client, "av_nobase")
+        resp = client.put("/api/avatar/loadout", headers=headers, json={"data": {"slots": {}}})
+        assert resp.status_code == 422
+
+    def test_rejects_non_object_slots(self):
+        client = TestClient(app)
+        headers = _auth_headers(client, "av_badslots")
+        resp = client.put(
+            "/api/avatar/loadout", headers=headers,
+            json={"data": {"base_body": "base_human_male_01", "slots": "nope"}},
+        )
+        assert resp.status_code == 422
+
+    def test_requires_auth(self):
+        client = TestClient(app)
+        resp = client.put("/api/avatar/loadout", json={"data": {"base_body": "x", "slots": {}}})
+        assert resp.status_code == 401