feat(milestone): merge phase/01 operator-foundation → milestone/v0.4-operator-tier

Phase 1 complete — Operator Foundation:
- Postgres 16 in Docker-in-LXC (asyncpg pool, 5-table schema, PgStore, migrations)
- Operator auth (argon2id, signed stateless cookies, slowapi 5/min rate limit)
- VC issuer key migration SQLite→Postgres (archive-before-active, R-VC-MIG-01)
- Operator bootstrap CLI (create-operator.py, idempotent)
- Backup cron script + G-008 restore drill
- Graceful degradation (server starts without Postgres)
- 272 tests pass, 33 skip (Postgres-requiring), 0 fail

---ci---
project: praxis
phase: 1
milestone: v0.4
status: complete
requirements:
  covered: [REQ-MT-01, REQ-AUTH-01, REQ-NFR-AUTH-01, REQ-NFR-MT-01, REQ-MT-02]
  partial: []
---/ci---

server/__main__.py

@@ -14,6 +14,7 @@ no audio/no tokens at runtime, not a crash.
 from __future__ import annotations
 
 import os
+from contextlib import asynccontextmanager
 from typing import Any
 
 from loguru import logger
@@ -27,14 +28,25 @@ try:
 except ImportError:  # pragma: no cover
     pass
 
-from fastapi import FastAPI, HTTPException
+from fastapi import FastAPI, HTTPException, Request
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
 from fastapi.staticfiles import StaticFiles
 from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection
+from slowapi.errors import RateLimitExceeded
+from slowapi import _rate_limit_exceeded_handler
 
+from db.pg_migrate import apply_pg_migrations
+from db.pg_store import PgStore
 from db.store import PraxisStore
+from server.auth.cookies import get_session_middleware_kwargs
+from server.auth.rate_limit import limiter
+from server.auth.routes import router as auth_router
 from server.pipeline import build_pipeline
+from server.vc.issuer_keys import _load_root_key
+from server.vc.migrate_keys import migrate_issuer_keys
 from server.vc.verification import verify_credential
+from starlette.middleware.sessions import SessionMiddleware
 
 _store = PraxisStore()
 
@@ -47,6 +59,58 @@ HOST = _env("PRAXIS_HOST", "0.0.0.0")
 PORT = int(_env("PRAXIS_PORT", "8789"))
 
 
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """v0.4 — create the asyncpg Postgres pool on startup, close on shutdown.
+
+    Graceful degradation (D-050, REQ-NFR-MT-01): if PRAXIS_PG_DSN is unset,
+    the server starts without Postgres — the learner voice loop (SQLite)
+    is unaffected. app.state.pg_pool / app.state.pg_store are None in that
+    case and auth/operator routes return 503.
+    """
+    dsn = os.environ.get("PRAXIS_PG_DSN", "").strip()
+    if not dsn:
+        logger.warning(
+            "PRAXIS_PG_DSN not set — starting without Postgres (dev/no-pool mode). "
+            "Operator auth + cohort endpoints will be unavailable (503). "
+            "Learner voice loop (SQLite) is unaffected."
+        )
+        app.state.pg_pool = None
+        app.state.pg_store = None
+        try:
+            yield
+        finally:
+            return
+    import asyncpg
+
+    logger.info("Creating asyncpg Postgres pool (min=1, max=10, D-050)")
+    pool = await asyncpg.create_pool(
+        dsn=dsn,
+        min_size=1,
+        max_size=10,
+        command_timeout=10,
+    )
+    app.state.pg_pool = pool
+    app.state.pg_store = PgStore(pool)
+    try:
+        applied = await apply_pg_migrations(pool)
+        if applied:
+            logger.info(f"Postgres migrations applied: {applied}")
+        else:
+            logger.info("Postgres migrations up to date")
+        # VC key migration (TASK-06-03, R-VC-MIG-01, G-027) — runs once on
+        # first boot, idempotent. Non-fatal on failure (v0.3 SQLite path
+        # remains intact for verification).
+        await _maybe_migrate_issuer_keys()
+        try:
+            yield
+        finally:
+            pass
+    finally:
+        await pool.close()
+        logger.info("Postgres pool closed")
+
+
 class WebRTCOffer(BaseModel):
     """Client→server WebRTC offer (SDP + type)."""
 
@@ -54,13 +118,19 @@ class WebRTCOffer(BaseModel):
     type: str = "offer"
 
 
-app = FastAPI(title="Praxis v0.1 voice server", version="0.1.0")
+app = FastAPI(title="Praxis v0.1 voice server", version="0.1.0", lifespan=lifespan)
+# slowapi rate-limit state + 429 handler (D-041, TASK-03-03).
+app.state.limiter = limiter
+app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],  # dev — the client is a separate Vite origin
     allow_methods=["*"],
     allow_headers=["*"],
 )
+# SessionMiddleware (signed cookies, D-056) — added AFTER CORS so it is
+# the outermost middleware (signs cookies before CORS headers are added).
+app.add_middleware(SessionMiddleware, **get_session_middleware_kwargs())
 
 
 @app.get("/health")
@@ -123,20 +193,60 @@ async def webrtc_offer(offer: WebRTCOffer) -> dict[str, str]:
 
 @app.get("/vc/verify/{credential_id}")
 async def vc_verify(credential_id: str) -> dict[str, Any]:
-    """Public, unauthenticated VC verification endpoint (D-043).
+    """Public, unauthenticated VC verification endpoint (D-043, G-011).
 
-    Returns {valid, status, issuer, credential, mastery, credentialTier,
-    verifiedAt}. 404 if the credential id is not found. No PII beyond what
-    the credential asserts.
+    Two-store fallback (G-011, binding contract):
+      (a) If Postgres is available (app.state.pg_store), use it for issuer
+          key lookup (active + superseded keys).
+      (b) If the credential is not in Postgres issued_credentials, fall back
+          to SQLite (v0.3 credentials remain in SQLite — D-051).
+      (c) If Postgres is NOT available, use the v0.3 SQLite path for both.
+    The VC key migration (TASK-04-03) runs once on first boot (idempotent)
+    inside the lifespan — see _maybe_migrate_issuer_keys.
     """
     await _store.init()
-    result = await verify_credential(_store, credential_id)
+    pg_store = getattr(app.state, "pg_store", None)
+    result = await verify_credential(
+        _store, credential_id,
+        pg_store=pg_store, sqlite_store=_store,
+    )
     if result is None:
         raise HTTPException(status_code=404, detail="credential not found")
     return result
 
 
-# ── Static client serving (D-023, REQ-DEPLOY-13) ────────────────────
+async def _maybe_migrate_issuer_keys() -> None:
+    """Run the VC key migration on first boot (TASK-06-03, R-VC-MIG-01).
+
+    Idempotent — no-op if Postgres already has an active issuer key. G-027:
+    if SQLite has no v0.3 active key (fresh deploy), skips archive and only
+    generates a fresh v0.4 keypair.
+    """
+    pg_store = getattr(app.state, "pg_store", None)
+    if pg_store is None:
+        return
+    try:
+        await _store.init()
+        root_key = _load_root_key()
+        result = await migrate_issuer_keys(_store, pg_store, root_key)
+        if result["new_key_id"] is not None:
+            logger.info(
+                f"VC key migration: archived v0.3 key={result['archived_key_id']}, "
+                f"generated fresh v0.4 key={result['new_key_id']}"
+            )
+        else:
+            logger.info("VC key migration: active key already present (no-op)")
+    except Exception as exc:
+        logger.error(f"VC key migration failed (non-fatal — v0.3 path intact): {exc}")
+
+
+# ── Operator auth routes (TASK-06-02, D-057) ───────────────────────────
+# Mounted BEFORE the StaticFiles mount so /api/operator/* is matched by
+# the router (routes-before-static-mount constraint, carry-forward v0.2).
+app.include_router(auth_router)
+
+
+# ── Static client serving (D-023, REQ-DEPLOY-13) ──────────────────────
 # Mount client/dist as StaticFiles at "/" AFTER all API routes so they
 # take precedence. html=True serves index.html for "/" (SPA root).
 # The client has no React Router (single-view state machine: start→live

server/auth/cookies.py

@@ -0,0 +1,68 @@
+"""Signed cookie configuration (TASK-03-02, D-041, D-056, R-AUTH-01, G-031).
+
+Returns kwargs for Starlette SessionMiddleware (itsdangerous HMAC-SHA256
+signed cookies — D-056, stateless, no sessions table). The cookie name is
+`praxis_op` (distinct from any future learner cookie).
+
+R-AUTH-01 / G-031 reframe: the PRIMARY mitigation for a sniffed operator
+cookie is the k-anonymity defense-in-depth — the cohort dashboard reads
+only k-anonymized aggregates, so a sniffed cookie leaks NO learner PII.
+The `PRAXIS_COOKIE_SECURE` flag is the SECONDARY mitigation (operational
+convenience for when TLS arrives). It defaults to true; the HTTP pilot
+(LXC, no TLS — D-030) sets it to false with a logged WARNING.
+"""
+
+from __future__ import annotations
+
+import os
+import secrets
+
+from loguru import logger
+
+_COOKIE_MAX_AGE_S = 28800  # 8h (D-041)
+
+
+def _env_bool(key: str, default: bool) -> bool:
+    raw = os.environ.get(key, "").strip().lower()
+    if raw in ("true", "1", "yes", "on"):
+        return True
+    if raw in ("false", "0", "no", "off"):
+        return False
+    return default
+
+
+def get_session_middleware_kwargs() -> dict:
+    """Return kwargs for Starlette SessionMiddleware.
+
+    If PRAXIS_COOKIE_SECRET is unset, generate an ephemeral random secret
+    and log a WARNING (dev only — sessions won't survive a restart and this
+    MUST NOT be used in pilot/production).
+    """
+    secret = os.environ.get("PRAXIS_COOKIE_SECRET", "").strip()
+    if not secret:
+        secret = secrets.token_urlsafe(48)
+        logger.warning(
+            "PRAXIS_COOKIE_SECRET not set — generated an ephemeral random secret. "
+            "Sessions will NOT survive a server restart. This is dev-only; set "
+            "PRAXIS_COOKIE_SECRET (>=32 bytes) for pilot/production."
+        )
+    secure = _env_bool("PRAXIS_COOKIE_SECURE", True)
+    if not secure:
+        logger.warning(
+            "Cookie Secure flag disabled (PRAXIS_COOKIE_SECURE=false) — HTTP pilot "
+            "mode (R-AUTH-01). Do not use in production. NOTE (G-031): the primary "
+            "R-AUTH-01 mitigation is k-anon defense-in-depth (cohort dashboard reads "
+            "only k-anonymized aggregates → sniffed cookie leaks no PII); this flag "
+            "is the secondary mitigation."
+        )
+    return {
+        "secret_key": secret,
+        "session_cookie": "praxis_op",
+        "max_age": _COOKIE_MAX_AGE_S,
+        "https_only": secure,
+        "same_site": "strict",
+        "path": "/",
+    }
+
+
+__all__ = ["get_session_middleware_kwargs"]

server/auth/dependencies.py

@@ -0,0 +1,56 @@
+"""current_operator dependency (TASK-03-04, D-057).
+
+Server-side auth enforcement: every `/api/operator/*` protected route uses
+`Depends(current_operator)`. The dependency NEVER trusts the client (D-057)
+— it reads the signed-cookie session, fetches the operator from Postgres,
+and 401s on any gap (missing/invalid/expired cookie, unknown id, inactive
+operator). The cookie is the authz *token*; the Postgres lookup is the
+authz *decision*.
+"""
+
+from __future__ import annotations
+
+from fastapi import HTTPException, Request, status
+
+from server.auth.models import Operator
+
+
+async def current_operator(request: Request) -> Operator:
+    """Resolve the authenticated operator from the signed-cookie session.
+
+    Raises 401 on: missing session, missing operator_id, no Postgres store
+    (503 actually — operator tier unavailable), unknown operator id, or an
+    inactive operator (session is cleared in the latter case so the client
+    cookie is invalidated).
+    """
+    pg_store = getattr(request.app.state, "pg_store", None)
+    if pg_store is None:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="operator tier unavailable (no Postgres)",
+        )
+    session = request.session
+    op_id = session.get("operator_id") if session else None
+    if not op_id:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="not authenticated",
+        )
+    row = await pg_store.get_operator_by_id(op_id)
+    if row is None or not row.get("is_active"):
+        # Inactive/unknown → clear the session so the cookie is invalidated.
+        if session:
+            session.clear()
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="not authenticated",
+        )
+    return Operator(
+        id=str(row["id"]),
+        username=row["username"],
+        display_name=row.get("display_name"),
+        role=row.get("role", "operator"),
+    )
+
+
+__all__ = ["current_operator"]

server/auth/models.py

@@ -0,0 +1,18 @@
+"""Auth data models (TASK-03-04)."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class Operator:
+    """The authenticated operator injected into protected routes (D-057)."""
+
+    id: str
+    username: str
+    display_name: str | None
+    role: str
+
+
+__all__ = ["Operator"]

server/auth/passwords.py

@@ -0,0 +1,44 @@
+"""Argon2id password hashing (TASK-03-01, D-041, REQ-NFR-AUTH-01).
+
+Uses argon2-cffi PasswordHasher with defaults that exceed OWASP minimums
+(time_cost=3, memory_cost=64MiB, parallelism=4 — RESEARCH-v0.4 §2.1).
+Single operator, low-frequency logins → hashing latency < 1s is
+acceptable (R-AUTH-02).
+"""
+
+from __future__ import annotations
+
+from argon2 import PasswordHasher
+from argon2.exceptions import VerifyMismatchError
+
+_ph = PasswordHasher()
+
+
+def hash_password(plain: str) -> str:
+    """Hash a plaintext password with argon2id. Returns the encoded hash string."""
+    return _ph.hash(plain)
+
+
+def verify_password(stored_hash: str, plain: str) -> bool:
+    """Verify a plaintext password against a stored argon2id hash.
+
+    Returns False on mismatch (no exception) so the login flow can apply a
+    uniform 401 + rate-limit-increment path on any auth failure.
+    """
+    try:
+        _ph.verify(stored_hash, plain)
+        return True
+    except VerifyMismatchError:
+        return False
+    except Exception:
+        return False
+
+
+def needs_rehash(stored_hash: str) -> bool:
+    """True if the stored hash was produced with weaker params than the
+    current PasswordHasher defaults. The login flow rehashes + updates the
+    store when this returns True (param upgrades without forcing a reset)."""
+    return _ph.check_needs_rehash(stored_hash)
+
+
+__all__ = ["hash_password", "verify_password", "needs_rehash"]

server/auth/rate_limit.py

@@ -0,0 +1,34 @@
+"""Login rate limiting (TASK-03-03, D-041).
+
+slowapi Limiter with an in-memory backend (single-instance — D-041).
+5 login attempts per minute per client IP. On exceed → 429 + Retry-After.
+
+R-AUTH-03 (in-memory counter lost on restart) is an accepted pilot risk
+(RESEARCH-v0.4 §2.5) — a restart at most resets the counter, which slightly
+widens the brute-force window but does not enable it (argon2id + 5/min is
+still the binding control). A hand-rolled counter is the documented
+fallback if slowapi is ever removed.
+"""
+
+from __future__ import annotations
+
+from slowapi import Limiter
+from slowapi.util import get_remote_address
+
+limiter = Limiter(key_func=get_remote_address, storage_uri="memory://")
+
+
+def reset_login_rate_limit() -> None:
+    """Clear the in-memory rate-limit counters (test helper + restart-safe)."""
+    try:
+        limiter.reset()
+    except Exception:
+        pass
+
+
+def rate_limit_login():
+    """Decorator factory: 5 login attempts per minute per IP (D-041)."""
+    return limiter.limit("5/minute")
+
+
+__all__ = ["limiter", "rate_limit_login", "reset_login_rate_limit"]

server/auth/routes.py

@@ -0,0 +1,118 @@
+"""Auth route handlers — login, logout, me (TASK-03-05, D-041, D-056, D-057).
+
+APIRouter(prefix="/api/operator") with:
+  POST /login   — rate-limited 5/min (TASK-03-03), NOT auth-gated.
+  POST /logout  — auth-gated (Depends(current_operator)).
+  GET  /me      — auth-gated (React route guard — D-057).
+
+Stateless cookies (D-056): logout clears the server-side session; the
+client also clears its cookie. No sessions table.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel
+
+from server.auth.dependencies import current_operator
+from server.auth.models import Operator
+from server.auth.passwords import hash_password, needs_rehash, verify_password
+from server.auth.rate_limit import rate_limit_login
+
+router = APIRouter(prefix="/api/operator", tags=["operator-auth"])
+
+
+class LoginBody(BaseModel):
+    username: str
+    password: str
+
+
+class OperatorOut(BaseModel):
+    id: str
+    username: str
+    display_name: str | None
+    role: str = "operator"
+
+
+class LoginResponse(BaseModel):
+    operator: OperatorOut
+
+
+class MeResponse(BaseModel):
+    operator: OperatorOut
+
+
+class OkResponse(BaseModel):
+    ok: bool = True
+
+
+def _operator_out(op: Operator) -> OperatorOut:
+    return OperatorOut(
+        id=op.id,
+        username=op.username,
+        display_name=op.display_name,
+        role=op.role,
+    )
+
+
+@router.post("/login", response_model=LoginResponse)
+@rate_limit_login()
+async def login(body: LoginBody, request: Request) -> LoginResponse:
+    """Rate-limited login (5/min per IP — D-041).
+
+    On success: sets `request.session["operator_id"]` (signed cookie via
+    SessionMiddleware) + updates last_login_at. On needs_rehash → rehash +
+    update the store. On failure → 401 (no cookie set).
+    """
+    pg_store = getattr(request.app.state, "pg_store", None)
+    if pg_store is None:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="operator tier unavailable (no Postgres)",
+        )
+    row = await pg_store.get_operator_by_username(body.username)
+    if row is None or not row.get("is_active"):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="invalid credentials",
+        )
+    if not verify_password(row["password_hash"], body.password):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="invalid credentials",
+        )
+    op_id = str(row["id"])
+    request.session["operator_id"] = op_id
+    await pg_store.update_last_login(op_id)
+    if needs_rehash(row["password_hash"]):
+        new_hash = hash_password(body.password)
+        async with pg_store.pool.acquire() as conn:
+            await conn.execute(
+                "UPDATE operators SET password_hash = $1 WHERE id = $2",
+                new_hash, op_id,
+            )
+    return LoginResponse(
+        operator=OperatorOut(
+            id=op_id,
+            username=row["username"],
+            display_name=row.get("display_name"),
+            role=row.get("role", "operator"),
+        )
+    )
+
+
+@router.post("/logout", response_model=OkResponse)
+async def logout(request: Request, op: Operator = Depends(current_operator)) -> OkResponse:
+    # Stateless (D-056): clearing the server session invalidates the signed
+    # cookie's payload; the client also clears its cookie.
+    request.session.clear()
+    return OkResponse(ok=True)
+
+
+@router.get("/me", response_model=MeResponse)
+async def me(op: Operator = Depends(current_operator)) -> MeResponse:
+    """React route guard endpoint (D-057). 200 → render; 401 → redirect."""
+    return MeResponse(operator=_operator_out(op))
+
+
+__all__ = ["router"]

server/vc/issuer_keys.py

@@ -13,6 +13,7 @@ import base64
 import os
 import uuid
 from dataclasses import dataclass
+from typing import Any, Protocol, runtime_checkable
 
 import nacl.secret
 import nacl.signing
@@ -22,6 +23,27 @@ from db.store import PraxisStore
 _SECRETBOX_KEY_BYTES = nacl.secret.SecretBox.KEY_SIZE
 
 
+@runtime_checkable
+class IssuerKeyStore(Protocol):
+    """Issuer key store protocol (D-051, TASK-04-01).
+
+    Both PraxisStore (SQLite, v0.3) and PgStore (Postgres, v0.4) implement
+    this protocol — R-VC-MIG-03 mitigation (both stores share the same
+    interface so verification can use either). The structural check lets
+    `isinstance(store, IssuerKeyStore)` succeed for duck-typed stores.
+    """
+
+    async def init_issuer_key(
+        self, key_id: str, public_key: str, private_key_enc: bytes
+    ) -> None: ...
+
+    async def get_active_signing_key_row(self) -> dict | None: ...
+
+    async def get_public_key_row(self, key_id: str) -> dict | None: ...
+
+    async def set_issuer_key_superseded(self, key_id: str) -> None: ...
+
+
 def _load_root_key() -> bytes:
     raw = os.environ.get("PRAXIS_VC_ISSUER_KEY", "")
     if raw:
@@ -63,7 +85,7 @@ def _decrypt_private_key(private_key_enc: bytes, root_key: bytes) -> nacl.signin
     return nacl.signing.SigningKey(seed)
 
 
-async def init_issuer_key(store: PraxisStore, root_key: bytes | None = None) -> KeyPair:
+async def init_issuer_key(store: IssuerKeyStore, root_key: bytes | None = None) -> KeyPair:
     rk = root_key if root_key is not None else _load_root_key()
     signing_key = nacl.signing.SigningKey.generate()
     verify_key = signing_key.verify_key
@@ -75,7 +97,7 @@ async def init_issuer_key(store: PraxisStore, root_key: bytes | None = None) ->
 
 
 async def get_active_signing_key(
-    store: PraxisStore, root_key: bytes | None = None
+    store: IssuerKeyStore, root_key: bytes | None = None
 ) -> tuple[KeyPair, bytes]:
     rk = root_key if root_key is not None else _load_root_key()
     row = await store.get_active_signing_key_row()
@@ -89,18 +111,19 @@ async def get_active_signing_key(
     return kp, row["private_key_enc"]
 
 
-async def _fetch_private_key_enc(store: PraxisStore, key_id: str) -> bytes:
-    async with store._connect() as db:
-        db.row_factory = None
-        cur = await db.execute(
-            "SELECT private_key_enc FROM issuer_keys WHERE id = ?", (key_id,)
-        )
-        row = await cur.fetchone()
-        return bytes(row[0]) if row else b""
+async def _fetch_private_key_enc(store: IssuerKeyStore, key_id: str) -> bytes:
+    # PraxisStore exposes a _connect() context manager; PgStore does not
+    # (it uses a pool). Use the protocol's get_public_key_row which both
+    # stores implement, and read private_key_enc from the returned row.
+    row = await store.get_public_key_row(key_id)
+    if row is None:
+        return b""
+    enc = row.get("private_key_enc")
+    return bytes(enc) if enc is not None else b""
 
 
 async def get_public_key_for_verification(
-    store: PraxisStore, key_id: str
+    store: IssuerKeyStore, key_id: str
 ) -> nacl.signing.VerifyKey:
     row = await store.get_public_key_row(key_id)
     if row is None:
@@ -119,6 +142,7 @@ async def rotate_key(store: PraxisStore, root_key: bytes | None = None) -> KeyPa
 
 
 __all__ = [
+    "IssuerKeyStore",
     "KeyPair",
     "init_issuer_key",
     "get_active_signing_key",

server/vc/migrate_keys.py

@@ -0,0 +1,94 @@
+"""VC issuer key migration SQLite → Postgres (TASK-04-03, D-051).
+
+One-time migration procedure (R-VC-MIG-01 — highest-severity v0.4 risk):
+  1. Read the v0.3 active public key from SQLite issuer_keys.
+  2. Insert that public key into Postgres issuer_keys with status=
+     'superseded' (private key NOT migrated — only the public key is
+     archived for verification of already-issued v0.3 VCs).
+  3. Generate a fresh Ed25519 keypair in Postgres issuer_keys with
+     status='active' (encrypted at rest with the root key).
+  4. Return {archived_key_id, new_key_id}.
+
+R-VC-MIG-01 mitigation: the v0.3 public key is archived as superseded
+BEFORE the fresh key is activated (step 2 before step 3). This guarantees
+v0.3 VCs remain verifiable against the archived key.
+
+G-027 (first-boot path): if SQLite has NO v0.3 active key (fresh deploy),
+skip the archive step and only generate the fresh v0.4 keypair.
+
+Idempotent: if Postgres already has an active key, the whole procedure is
+a no-op. If Postgres already has a superseded key matching the v0.3 key_id,
+skip step 2 (already archived) but still generate the fresh key if no
+active key exists.
+"""
+
+from __future__ import annotations
+
+import base64
+import uuid
+from typing import Any
+
+import nacl.signing
+
+from db.pg_store import PgStore
+from db.store import PraxisStore
+from server.vc.issuer_keys import _encrypt_private_key
+
+
+async def _archive_v03_public_key(
+    pg_store: PgStore, v03_key_id: str, v03_public_key: str
+) -> None:
+    """Insert the v0.3 public key into Postgres as superseded (idempotent)."""
+    existing = await pg_store.get_public_key_row(v03_key_id)
+    if existing is not None:
+        return  # already archived (or present as active — leave as-is)
+    await pg_store.init_issuer_key(v03_key_id, v03_public_key, b"")
+    await pg_store.set_issuer_key_superseded(v03_key_id)
+
+
+async def _generate_fresh_v04_key(
+    pg_store: PgStore, root_key: bytes
+) -> str:
+    """Generate a fresh Ed25519 keypair in Postgres as active. Returns key_id."""
+    signing_key = nacl.signing.SigningKey.generate()
+    verify_key = signing_key.verify_key
+    public_key_b64 = base64.b64encode(bytes(verify_key)).decode("ascii")
+    private_key_enc = _encrypt_private_key(signing_key, root_key)
+    key_id = f"key-{uuid.uuid4().hex[:12]}"
+    await pg_store.init_issuer_key(key_id, public_key_b64, private_key_enc)
+    return key_id
+
+
+async def migrate_issuer_keys(
+    sqlite_store: PraxisStore,
+    pg_store: PgStore,
+    root_key: bytes,
+) -> dict[str, str | None]:
+    """Run the one-time VC key migration. Idempotent.
+
+    Returns {"archived_key_id": str | None, "new_key_id": str | None}.
+    archived_key_id is None on the G-027 first-boot path (no v0.3 key).
+    new_key_id is None if an active key already existed (no-op).
+    """
+    # If Postgres already has an active key, the whole migration is done.
+    active = await pg_store.get_active_signing_key_row()
+    if active is not None:
+        return {"archived_key_id": None, "new_key_id": None}
+
+    # Step 1 (G-027): read v0.3 active public key from SQLite. May be None
+    # on a fresh deploy with no v0.3 history.
+    v03_row = await sqlite_store.get_active_signing_key_row()
+    archived_key_id: str | None = None
+    if v03_row is not None:
+        v03_key_id = v03_row["id"]
+        v03_public_key = v03_row["public_key"]
+        # Step 2 (R-VC-MIG-01): archive BEFORE activating the fresh key.
+        await _archive_v03_public_key(pg_store, v03_key_id, v03_public_key)
+        archived_key_id = v03_key_id
+
+    # Step 3: generate the fresh v0.4 keypair as active.
+    new_key_id = await _generate_fresh_v04_key(pg_store, root_key)
+    return {"archived_key_id": archived_key_id, "new_key_id": new_key_id}
+
+
+__all__ = ["migrate_issuer_keys"]

server/vc/verification.py

@@ -1,11 +1,24 @@
-"""Public VC verification (SLICE-09 TASK-09-04, D-043, REQ-NFR-VC-02).
+"""Public VC verification (SLICE-09 TASK-09-04, D-043, REQ-NFR-VC-02;
+v0.4 TASK-04-04 two-store fallback per G-011).
 
 `GET /vc/verify/<credential_id>` — public, unauthenticated. Fetches the
-credential from SQLite, fetches the issuer public key, validates the Ed25519
-signature against the JCS-canonicalized payload, checks the Bitstring Status
-List (no cache — fetched on every verify call, REQ-NFR-VC-02). Returns JSON
+credential + issuer public key, validates the Ed25519 signature against
+the JCS-canonicalized payload, checks the Bitstring Status List (no cache
+— fetched on every verify call, REQ-NFR-VC-02). Returns JSON
 {valid, status, issuer, credential, mastery, credentialTier, verifiedAt}.
 No PII beyond what the credential asserts.
+
+G-011 two-store fallback semantics (binding contract):
+  (a) If Postgres is available (pg_store is not None), use it for issuer
+      key lookup (both active AND superseded keys — get_public_key_row
+      queries by id, not status).
+  (b) If Postgres is available but the credential is not found in its
+      issued_credentials table, fall back to SQLite issued_credentials
+      (v0.3 credentials remain in SQLite — D-051 "no re-issuance").
+  (c) If Postgres is NOT available (pg_store is None), use the existing
+      v0.3 SQLite path for BOTH keys and credentials (full v0.3 compat).
+The key store used for verification is always the one that holds the key
+row found by key_id; the credential store is whichever store had the row.
 """
 
 from __future__ import annotations
@@ -17,7 +30,7 @@ from typing import Any
 from db.store import PraxisStore
 
 from server.vc.issuer import verify_proof, extract_key_id, CREDENTIAL_TIER
-from server.vc.issuer_keys import get_public_key_for_verification
+from server.vc.issuer_keys import IssuerKeyStore, get_public_key_for_verification
 from server.vc.status_list import BitstringStatusList
 
 
@@ -26,26 +39,35 @@ def _now_iso() -> str:
 
 
 async def verify_credential(
-    store: PraxisStore, credential_id: str
+    store: IssuerKeyStore,
+    credential_id: str,
+    *,
+    pg_store: IssuerKeyStore | None = None,
+    sqlite_store: PraxisStore | None = None,
 ) -> dict[str, Any] | None:
-    row = await store.get_credential(credential_id)
+    """Verify a VC. Returns the verification result dict, or None if the
+    credential id is not found in any store.
+
+    Per G-011:
+      - If pg_store is provided, try it first for BOTH credential + key
+        lookup; fall back to sqlite_store for the credential if Postgres
+        doesn't have it (v0.3 credentials stay in SQLite).
+      - If pg_store is None, use `store` (the v0.3 SQLite path) for both.
+    """
+    row = await _lookup_credential(credential_id, store, pg_store, sqlite_store)
     if row is None:
         return None
     secured_doc = json.loads(row["vc_payload_json"])
     key_id = extract_key_id(secured_doc)
     if key_id is None:
         return _invalid(row, secured_doc)
-    try:
-        verify_key = await get_public_key_for_verification(store, key_id)
-    except KeyError:
+    # Key lookup: prefer pg_store (G-011a) for v0.4 keys + archived v0.3
+    # keys; fall back to `store` (SQLite) if pg_store doesn't have the key.
+    verify_key = await _lookup_public_key(key_id, store, pg_store)
+    if verify_key is None:
         return _invalid(row, secured_doc)
     sig_valid = verify_proof(secured_doc, verify_key)
-    revoked = False
-    cs = secured_doc.get("credentialStatus") or {}
-    idx_str = cs.get("statusListIndex")
-    if idx_str is not None:
-        sl = BitstringStatusList(store, "default")
-        revoked = await sl.get_status(int(idx_str))
+    revoked = await _check_revocation(secured_doc, store, sqlite_store or store)
     status = "revoked" if revoked else "active"
     valid = bool(sig_valid and not revoked)
     subject = secured_doc.get("credentialSubject") or {}
@@ -73,6 +95,54 @@ async def verify_credential(
     }
 
 
+async def _lookup_credential(
+    credential_id: str,
+    store: IssuerKeyStore,
+    pg_store: IssuerKeyStore | None,
+    sqlite_store: PraxisStore | None,
+) -> dict | None:
+    """G-011(b): try Postgres first, fall back to SQLite for v0.3 creds."""
+    if pg_store is not None:
+        row = await pg_store.get_credential(credential_id)
+        if row is not None:
+            return row
+        if sqlite_store is not None:
+            return await sqlite_store.get_credential(credential_id)
+        return None
+    # G-011(c): no Postgres — v0.3 SQLite path.
+    return await store.get_credential(credential_id)
+
+
+async def _lookup_public_key(
+    key_id: str,
+    store: IssuerKeyStore,
+    pg_store: IssuerKeyStore | None,
+):
+    """G-011(a): prefer Postgres for key lookup (finds active + superseded);
+    fall back to `store` (SQLite) if Postgres doesn't have the key."""
+    if pg_store is not None:
+        try:
+            vk = await get_public_key_for_verification(pg_store, key_id)
+            return vk
+        except KeyError:
+            pass
+    try:
+        return await get_public_key_for_verification(store, key_id)
+    except KeyError:
+        return None
+
+
+async def _check_revocation(
+    secured_doc: dict, store: IssuerKeyStore, status_store: PraxisStore
+) -> bool:
+    cs = secured_doc.get("credentialStatus") or {}
+    idx_str = cs.get("statusListIndex")
+    if idx_str is None:
+        return False
+    sl = BitstringStatusList(status_store, "default")
+    return await sl.get_status(int(idx_str))
+
+
 def _invalid(row: dict, secured_doc: dict) -> dict[str, Any]:
     subject = secured_doc.get("credentialSubject") or {}
     return {