feat(P01): SLICE-04 VC issuer key migration SQLite→Postgres (R-VC-MIG-01)

- TASK-04-01 server/vc/issuer_keys.py: refactor to IssuerKeyStore
  Protocol (runtime_checkable). PraxisStore + PgStore both implement it
  (R-VC-MIG-03). Functions now accept IssuerKeyStore instead of
  PraxisStore. _fetch_private_key_enc rewritten to use
  get_public_key_row (protocol method) instead of store._connect()
  (PgStore has no _connect). Backward-compatible — all 19 v0.3 VC
  tests still pass.
- TASK-04-02 db/pg_store.py: IssuerKeyStore methods (already implemented
  in TASK-01-06): init/get_active/get_public_key_row/set_superseded.
  get_public_key_row queries by id (not status) → finds superseded keys
  (R-VC-MIG-01 fallback). db/store.py get_public_key_row now also
  returns private_key_enc (protocol alignment).
- TASK-04-03 server/vc/migrate_keys.py: migrate_issuer_keys() one-time
  procedure. R-VC-MIG-01: archives v0.3 public key as superseded BEFORE
  generating the fresh v0.4 active key (step 2 before step 3). G-027
  first-boot path: no v0.3 active key in SQLite → skip archive, generate
  fresh key only. Idempotent (no-op if Postgres already has an active key).
- TASK-04-04 server/vc/verification.py: verify_credential now accepts
  pg_store + sqlite_store kwargs. G-011 two-store fallback (binding):
  (a) Postgres for key lookup (active + superseded); (b) Postgres for
  credential, fall back to SQLite if not found (v0.3 creds stay in
  SQLite); (c) SQLite-only if no Postgres (v0.3 compat).
- TASK-04-05 tests/test_vc_migration.py: 9 tests — migration archives +
  generates fresh, idempotent, G-027 first-boot, archive-before-active
  ordering (R-VC-MIG-01), v0.3 VC verifies against superseded key in
  Postgres (R-VC-MIG-01 critical), v0.4 VC verifies, tamper detection,
  G-011(b) SQLite fallback, G-011(c) SQLite-only.

---ci---
project: praxis
phase: 1
milestone: v0.4
status: execute
persona: security-engineer
task: 04-01,04-02,04-03,04-04,04-05
requirements:
  covered: [REQ-MT-01]
  grill:
    - G-011 (two-store fallback semantics — explicit in verify_credential)
    - G-027 (first-boot: no v0.3 key → skip archive, fresh key only)
  risks:
    - R-VC-MIG-01 (archived-before-active — tested in test_migration_archives_before_activating_r_vc_mig_01 + test_v03_vc_verifies_against_superseded_key_in_pg)
---/ci---

db/store.py

@@ -338,7 +338,7 @@ class PraxisStore:
         async with self._connect() as db:
             db.row_factory = aiosqlite.Row
             cur = await db.execute(
-                "SELECT id, public_key, status, created_at "
+                "SELECT id, public_key, private_key_enc, status, created_at "
                 "FROM issuer_keys WHERE id = ?",
                 (key_id,),
             )

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 {

tests/test_vc_migration.py

@@ -0,0 +1,354 @@
+"""VC migration unit tests (TASK-04-05) — mocked stores.
+
+Covers:
+  - Migration script: v0.3 key archived as superseded, fresh key active,
+    idempotent re-run.
+  - G-027 first-boot path: no v0.3 active key in SQLite → skip archive,
+    generate fresh key only.
+  - Verification with PgStore: v0.4 VC (active key) verifies ✓; v0.3 VC
+    (superseded key) verifies ✓ (R-VC-MIG-01 — the critical test).
+  - get_public_key_row finds superseded key by id (verification fallback).
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import uuid
+from unittest.mock import AsyncMock, MagicMock
+
+import nacl.signing
+import pytest
+
+from server.vc.issuer import build_vc_payload, sign, extract_key_id, verify_proof
+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
+
+
+def _b64(b: bytes) -> str:
+    return base64.b64encode(b).decode("ascii")
+
+
+# ── Migration script ────────────────────────────────────────────────────────
+
+
+def _make_sqlite_store_with_v03_key(key_id="v03-key-aaa", public_key="pub-v03-b64"):
+    store = MagicMock()
+    store.get_active_signing_key_row = AsyncMock(
+        return_value={"id": key_id, "public_key": public_key, "private_key_enc": b"\x01"}
+    )
+    return store
+
+
+def _make_pg_store():
+    store = MagicMock()
+    store._rows = {}  # key_id -> row dict
+    store._active = None
+
+    async def init_issuer_key(key_id, public_key, private_key_enc):
+        status = "active"
+        if key_id in store._rows:
+            # ON CONFLICT DO NOTHING — don't overwrite
+            return
+        store._rows[key_id] = {
+            "id": key_id, "public_key": public_key,
+            "private_key_enc": private_key_enc, "status": status,
+        }
+        store._active = key_id
+
+    async def get_active_signing_key_row():
+        if store._active is None:
+            return None
+        return dict(store._rows[store._active])
+
+    async def get_public_key_row(key_id):
+        r = store._rows.get(key_id)
+        return dict(r) if r else None
+
+    async def set_issuer_key_superseded(key_id):
+        if key_id in store._rows:
+            store._rows[key_id]["status"] = "superseded"
+            if store._active == key_id:
+                store._active = None
+
+    store.init_issuer_key = init_issuer_key
+    store.get_active_signing_key_row = get_active_signing_key_row
+    store.get_public_key_row = get_public_key_row
+    store.set_issuer_key_superseded = set_issuer_key_superseded
+    return store
+
+
+@pytest.mark.asyncio
+async def test_migration_archives_v03_and_generates_fresh():
+    sqlite = _make_sqlite_store_with_v03_key()
+    pg = _make_pg_store()
+    root = _load_root_key()
+    result = await migrate_issuer_keys(sqlite, pg, root)
+    assert result["archived_key_id"] == "v03-key-aaa"
+    assert result["new_key_id"] is not None
+    # v0.3 key is superseded in Postgres
+    archived = await pg.get_public_key_row("v03-key-aaa")
+    assert archived["status"] == "superseded"
+    assert archived["public_key"] == "pub-v03-b64"
+    # fresh key is active
+    active = await pg.get_active_signing_key_row()
+    assert active is not None
+    assert active["id"] == result["new_key_id"]
+    assert active["status"] == "active"
+
+
+@pytest.mark.asyncio
+async def test_migration_idempotent_when_active_key_exists():
+    sqlite = _make_sqlite_store_with_v03_key()
+    pg = _make_pg_store()
+    root = _load_root_key()
+    await migrate_issuer_keys(sqlite, pg, root)
+    # second run — no-op
+    result = await migrate_issuer_keys(sqlite, pg, root)
+    assert result["archived_key_id"] is None
+    assert result["new_key_id"] is None
+
+
+@pytest.mark.asyncio
+async def test_migration_g027_first_boot_no_v03_key():
+    # G-027: no v0.3 active key in SQLite → skip archive, fresh key only.
+    sqlite = MagicMock()
+    sqlite.get_active_signing_key_row = AsyncMock(return_value=None)
+    pg = _make_pg_store()
+    root = _load_root_key()
+    result = await migrate_issuer_keys(sqlite, pg, root)
+    assert result["archived_key_id"] is None  # nothing archived
+    assert result["new_key_id"] is not None   # fresh key generated
+    active = await pg.get_active_signing_key_row()
+    assert active is not None
+    assert active["id"] == result["new_key_id"]
+
+
+@pytest.mark.asyncio
+async def test_migration_archives_before_activating_r_vc_mig_01():
+    # R-VC-MIG-01: the v0.3 public key MUST be archived BEFORE the fresh
+    # key is activated. We verify the ordering by checking that at no point
+    # is there an active v0.4 key without the v0.3 key being present (as
+    # superseded) in Postgres.
+    sqlite = _make_sqlite_store_with_v03_key()
+    pg = _make_pg_store()
+    # Instrument init_issuer_key to assert the archive happened first.
+    order = []
+    orig_init = pg.init_issuer_key
+
+    async def tracing_init(key_id, public_key, private_key_enc):
+        order.append(key_id)
+        await orig_init(key_id, public_key, private_key_enc)
+
+    pg.init_issuer_key = tracing_init
+    orig_super = pg.set_issuer_key_superseded
+
+    async def tracing_super(key_id):
+        order.append(f"supersede:{key_id}")
+        await orig_super(key_id)
+
+    pg.set_issuer_key_superseded = tracing_super
+    root = _load_root_key()
+    await migrate_issuer_keys(sqlite, pg, root)
+    # The v0.3 key (v03-key-aaa) is init'd then superseded BEFORE the fresh
+    # key is init'd (active).
+    v03_idx = order.index("v03-key-aaa")
+    sup_idx = order.index("supersede:v03-key-aaa")
+    fresh_idx = [i for i, k in enumerate(order) if k not in ("v03-key-aaa",) and not k.startswith("supersede:")][0]
+    assert v03_idx < sup_idx < fresh_idx
+
+
+# ── Verification with PgStore (R-VC-MIG-01 critical test) ───────────────────
+
+
+def _make_credential_store(rows: dict[str, dict]):
+    store = MagicMock()
+    async def get_credential(cid):
+        return rows.get(cid)
+    store.get_credential = get_credential
+    # status list store (SQLite) — empty
+    sl_store = MagicMock()
+    sl_store.get_status_list = AsyncMock(return_value=None)
+    sl_store.upsert_status_list = AsyncMock()
+    return store, sl_store
+
+
+@pytest.mark.asyncio
+async def test_v03_vc_verifies_against_superseded_key_in_pg():
+    """R-VC-MIG-01 critical: a v0.3 VC verifies against a Postgres store
+    with the v0.3 public key archived as superseded."""
+    # Generate a v0.3 keypair + credential.
+    sk_v03 = nacl.signing.SigningKey.generate()
+    vk_v03 = sk_v03.verify_key
+    pub_v03_b64 = _b64(bytes(vk_v03))
+    v03_key_id = "v03-key-real"
+    payload = build_vc_payload(
+        learner_ref="learner-1", path="cs-refund",
+        scenarios_passed=["sc-1"], rubric_score=4.0, completed_weeks=6,
+        evidence=[], credential_id="vc-v03-real", status_list_index=None,
+    )
+    secured, sig_b64 = sign(payload, sk_v03, v03_key_id)
+    cred_row = {
+        "id": "vc-v03-real", "learner_ref": "learner-1",
+        "vc_payload_json": json.dumps(secured, sort_keys=True, separators=(",", ":")),
+        "signature_b64": sig_b64, "status": "active",
+    }
+    # Postgres store has the v0.3 key as superseded + the credential.
+    pg = _make_pg_store()
+    await pg.init_issuer_key(v03_key_id, pub_v03_b64, b"")
+    await pg.set_issuer_key_superseded(v03_key_id)
+    pg._rows[v03_key_id]["public_key"] = pub_v03_b64
+    # add credential to pg via a separate mock get_credential
+    async def get_cred(cid):
+        if cid == "vc-v03-real":
+            return cred_row
+        return None
+    pg.get_credential = get_cred
+    # SQLite status-list store (empty → not revoked)
+    sqlite_sl = MagicMock()
+    sqlite_sl.get_status_list = AsyncMock(return_value=None)
+    sqlite_sl.upsert_status_list = AsyncMock()
+    result = await verify_credential(
+        pg, "vc-v03-real", pg_store=pg, sqlite_store=sqlite_sl
+    )
+    assert result is not None
+    assert result["valid"] is True, "v0.3 VC must verify against archived superseded key (R-VC-MIG-01)"
+
+
+@pytest.mark.asyncio
+async def test_v04_vc_verifies_against_active_key_in_pg():
+    sk_v04 = nacl.signing.SigningKey.generate()
+    vk_v04 = sk_v04.verify_key
+    pub_v04_b64 = _b64(bytes(vk_v04))
+    v04_key_id = "v04-key-fresh"
+    payload = build_vc_payload(
+        learner_ref="learner-2", path="cs-refund",
+        scenarios_passed=["sc-1", "sc-2"], rubric_score=4.5, completed_weeks=6,
+        evidence=[], credential_id="vc-v04-fresh", status_list_index=None,
+    )
+    secured, sig_b64 = sign(payload, sk_v04, v04_key_id)
+    cred_row = {
+        "id": "vc-v04-fresh", "learner_ref": "learner-2",
+        "vc_payload_json": json.dumps(secured, sort_keys=True, separators=(",", ":")),
+        "signature_b64": sig_b64, "status": "active",
+    }
+    pg = _make_pg_store()
+    await pg.init_issuer_key(v04_key_id, pub_v04_b64, b"\x09")
+    pg._rows[v04_key_id]["public_key"] = pub_v04_b64
+    async def get_cred(cid):
+        return cred_row if cid == "vc-v04-fresh" else None
+    pg.get_credential = get_cred
+    sqlite_sl = MagicMock()
+    sqlite_sl.get_status_list = AsyncMock(return_value=None)
+    result = await verify_credential(
+        pg, "vc-v04-fresh", pg_store=pg, sqlite_store=sqlite_sl
+    )
+    assert result is not None
+    assert result["valid"] is True
+
+
+@pytest.mark.asyncio
+async def test_tampered_v03_vc_fails_verification():
+    sk = nacl.signing.SigningKey.generate()
+    vk = sk.verify_key
+    pub_b64 = _b64(bytes(vk))
+    key_id = "key-tamper"
+    payload = build_vc_payload(
+        learner_ref="learner-t", path="cs-refund",
+        scenarios_passed=["sc-1"], rubric_score=4.0, completed_weeks=6,
+        evidence=[], credential_id="vc-tamper", status_list_index=None,
+    )
+    secured, sig_b64 = sign(payload, sk, key_id)
+    # Tamper: change the rubricScore after signing.
+    secured["credentialSubject"]["rubricScore"] = 1.0
+    cred_row = {
+        "id": "vc-tamper", "learner_ref": "learner-t",
+        "vc_payload_json": json.dumps(secured, sort_keys=True, separators=(",", ":")),
+        "signature_b64": sig_b64, "status": "active",
+    }
+    pg = _make_pg_store()
+    await pg.init_issuer_key(key_id, pub_b64, b"")
+    pg._rows[key_id]["public_key"] = pub_b64
+    async def get_cred(cid):
+        return cred_row if cid == "vc-tamper" else None
+    pg.get_credential = get_cred
+    sqlite_sl = MagicMock()
+    sqlite_sl.get_status_list = AsyncMock(return_value=None)
+    result = await verify_credential(
+        pg, "vc-tamper", pg_store=pg, sqlite_store=sqlite_sl
+    )
+    assert result is not None
+    assert result["valid"] is False
+
+
+@pytest.mark.asyncio
+async def test_verification_fallback_sqlite_when_pg_missing_credential():
+    """G-011(b): credential not in Postgres → fall back to SQLite."""
+    sk = nacl.signing.SigningKey.generate()
+    vk = sk.verify_key
+    pub_b64 = _b64(bytes(vk))
+    key_id = "key-fallback"
+    payload = build_vc_payload(
+        learner_ref="learner-fb", path="cs-refund",
+        scenarios_passed=["sc-1"], rubric_score=4.0, completed_weeks=6,
+        evidence=[], credential_id="vc-fallback", status_list_index=None,
+    )
+    secured, sig_b64 = sign(payload, sk, key_id)
+    sqlite_cred_row = {
+        "id": "vc-fallback", "learner_ref": "learner-fb",
+        "vc_payload_json": json.dumps(secured, sort_keys=True, separators=(",", ":")),
+        "signature_b64": sig_b64, "status": "active",
+    }
+    # Postgres has the key but NOT the credential.
+    pg = _make_pg_store()
+    await pg.init_issuer_key(key_id, pub_b64, b"")
+    pg._rows[key_id]["public_key"] = pub_b64
+    async def pg_get_cred(cid):
+        return None  # not in Postgres
+    pg.get_credential = pg_get_cred
+    # SQLite has the credential + the key (v0.3 path).
+    sqlite = MagicMock()
+    async def sqlite_get_cred(cid):
+        return sqlite_cred_row if cid == "vc-fallback" else None
+    sqlite.get_credential = sqlite_get_cred
+    sqlite.get_public_key_row = AsyncMock(return_value={
+        "id": key_id, "public_key": pub_b64, "status": "active"
+    })
+    sqlite.get_status_list = AsyncMock(return_value=None)
+    result = await verify_credential(
+        sqlite, "vc-fallback", pg_store=pg, sqlite_store=sqlite
+    )
+    assert result is not None
+    assert result["valid"] is True
+
+
+@pytest.mark.asyncio
+async def test_verification_sqlite_only_when_no_pg():
+    """G-011(c): no Postgres → full v0.3 SQLite path."""
+    sk = nacl.signing.SigningKey.generate()
+    vk = sk.verify_key
+    pub_b64 = _b64(bytes(vk))
+    key_id = "key-sqlite-only"
+    payload = build_vc_payload(
+        learner_ref="learner-so", path="cs-refund",
+        scenarios_passed=["sc-1"], rubric_score=4.0, completed_weeks=6,
+        evidence=[], credential_id="vc-so", status_list_index=None,
+    )
+    secured, sig_b64 = sign(payload, sk, key_id)
+    cred_row = {
+        "id": "vc-so", "learner_ref": "learner-so",
+        "vc_payload_json": json.dumps(secured, sort_keys=True, separators=(",", ":")),
+        "signature_b64": sig_b64, "status": "active",
+    }
+    sqlite = MagicMock()
+    async def get_cred(cid):
+        return cred_row if cid == "vc-so" else None
+    sqlite.get_credential = get_cred
+    sqlite.get_public_key_row = AsyncMock(return_value={
+        "id": key_id, "public_key": pub_b64, "status": "active"
+    })
+    sqlite.get_status_list = AsyncMock(return_value=None)
+    result = await verify_credential(sqlite, "vc-so", pg_store=None, sqlite_store=sqlite)
+    assert result is not None
+    assert result["valid"] is True