feat(P01): TASK-01-04..07 Postgres DB foundation — migrate + schema + PgStore + tests

- db/pg_migrate.py: asyncpg migration runner with _pg_migrations tracking table, ordered .sql, transactional, 3x retry on connection failure (R-MT-02). - db/pg_schema.sql + db/pg_migrations/0001_operator_tier.sql: 5 operator-tier tables (operators, issued_credentials, mastery_gate_events, cohort_aggregates, issuer_keys) using gen_random_uuid() (PG16 core, no extension). cohort_aggregates is a plain table, NOT partitioned (D-050). - db/pg_store.py: PgStore class implementing the IssuerKeyStore protocol (init/get_active/get_public_key_row/set_superseded) plus operator CRUD, cohort aggregate read/write, credential methods, gate events. get_public_key_row queries by id (not status) → finds superseded keys (R-VC-MIG-01 verification fallback, D-051). No cross-DB FKs (D-031). - tests/test_pg_store.py: 13 integration tests (skip if PRAXIS_PG_DSN unset). ---ci--- project: praxis phase: 1 milestone: v0.4 status: execute persona: data-engineer task: 01-04,01-05,01-06,01-07 requirements: covered: [REQ-MT-01, REQ-NFR-MT-01, REQ-MT-02] ---/ci---
2026-08-04 00:47:14 +00:00
parent 6ada2560ba
commit b0cb6280d7
5 changed files with 702 additions and 0 deletions
@@ -0,0 +1,71 @@
+"""Postgres migration runner — applies db/pg_migrations/*.sql in order.
+
+Mirrors db/migrate.py: ordered .sql files tracked in a `_pg_migrations`
+table so re-running is idempotent. Uses an asyncpg pool. Retries on
+connection failure (3 attempts, 2s backoff — R-MT-02 mitigation).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import datetime as _dt
+from pathlib import Path
+
+import asyncpg
+
+_DEFAULT_MIGRATIONS_DIR = Path(__file__).resolve().parent / "pg_migrations"
+_RETRY_ATTEMPTS = 3
+_RETRY_BACKOFF_S = 2.0
+
+
+async def apply_pg_migrations(
+    pool: asyncpg.Pool,
+    migrations_dir: Path | None = None,
+) -> list[str]:
+    """Apply all pending Postgres migrations in order. Returns applied names.
+
+    Idempotent — no-op if all migrations are already applied. Each migration
+    runs within a transaction; the `_pg_migrations` tracking row is inserted
+    in the same transaction so a failure rolls back cleanly.
+    """
+    mdir = migrations_dir or _DEFAULT_MIGRATIONS_DIR
+    if not mdir.exists():
+        return []
+
+    async def _run() -> list[str]:
+        async with pool.acquire() as conn:
+            await conn.execute(
+                "CREATE TABLE IF NOT EXISTS _pg_migrations ("
+                "id TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now()"
+                ")"
+            )
+            rows = await conn.fetch("SELECT id FROM _pg_migrations")
+            applied_ids = {r["id"] for r in rows}
+            applied: list[str] = []
+            for sql_path in sorted(mdir.glob("*.sql")):
+                mid = sql_path.stem
+                if mid in applied_ids:
+                    continue
+                sql = sql_path.read_text(encoding="utf-8")
+                async with conn.transaction():
+                    await conn.execute(sql)
+                    await conn.execute(
+                        "INSERT INTO _pg_migrations (id) VALUES ($1)", mid
+                    )
+                    applied.append(mid)
+            return applied
+
+    last_exc: Exception | None = None
+    for attempt in range(1, _RETRY_ATTEMPTS + 1):
+        try:
+            return await _run()
+        except (asyncpg.PostgresConnectionError, ConnectionError, OSError) as exc:
+            last_exc = exc
+            if attempt < _RETRY_ATTEMPTS:
+                await asyncio.sleep(_RETRY_BACKOFF_S)
+            continue
+    assert last_exc is not None
+    raise last_exc
+
+
+__all__ = ["apply_pg_migrations"]
@@ -0,0 +1,59 @@
+-- Praxis v0.4 operator-tier schema migration 0001.
+-- Creates the 5 operator-tier tables. Uses gen_random_uuid() (PG16 core).
+-- Idempotent via IF NOT EXISTS (also safe through pg_migrate tracking).
+
+CREATE TABLE IF NOT EXISTS operators (
+    id            UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    username      TEXT UNIQUE NOT NULL,
+    password_hash TEXT NOT NULL,
+    display_name  TEXT,
+    role          TEXT NOT NULL DEFAULT 'operator',
+    is_active     BOOLEAN NOT NULL DEFAULT TRUE,
+    created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
+    last_login_at TIMESTAMPTZ
+);
+
+CREATE TABLE IF NOT EXISTS issued_credentials (
+    id            UUID PRIMARY KEY,
+    operator_id   UUID REFERENCES operators(id),
+    learner_ref   TEXT NOT NULL,
+    vc_type       TEXT,
+    payload_jsonb JSONB NOT NULL,
+    signature_b64 TEXT NOT NULL,
+    status        TEXT NOT NULL DEFAULT 'active',
+    issued_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
+    revoked_at    TIMESTAMPTZ
+);
+
+CREATE TABLE IF NOT EXISTS mastery_gate_events (
+    id                  UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    learner_ref         TEXT NOT NULL,
+    scenario_id         TEXT,
+    path_id             TEXT NOT NULL,
+    gate_outcome        TEXT,
+    rubric_scores_jsonb JSONB,
+    recorded_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
+    source              TEXT NOT NULL DEFAULT 'sync'
+);
+
+CREATE TABLE IF NOT EXISTS cohort_aggregates (
+    path            TEXT NOT NULL,
+    metric          TEXT NOT NULL,
+    window_start    DATE NOT NULL,
+    window_end      DATE NOT NULL,
+    value           NUMERIC,
+    cell_count      INTEGER NOT NULL DEFAULT 0,
+    cell_suppressed BOOLEAN NOT NULL DEFAULT FALSE,
+    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (path, metric, window_start)
+);
+CREATE INDEX IF NOT EXISTS cohort_aggregates_path_window_idx
+    ON cohort_aggregates (path, window_start);
+
+CREATE TABLE IF NOT EXISTS issuer_keys (
+    id               TEXT PRIMARY KEY,
+    public_key       TEXT NOT NULL,
+    private_key_enc  BYTEA,
+    status           TEXT NOT NULL DEFAULT 'active',
+    created_at       TIMESTAMPTZ NOT NULL DEFAULT now()
+);
@@ -0,0 +1,71 @@
+-- Praxis v0.4 operator-tier Postgres schema (reference).
+-- Applied in order by db/pg_migrate.py via db/pg_migrations/*.sql.
+-- The canonical migration is 0001_operator_tier.sql; this file is the
+-- human-readable reference (kept in sync). Uses gen_random_uuid() which
+-- is in PG16 core (no extension needed — R-MT-05 verified).
+--
+-- Tables:
+--   operators            — operator accounts (argon2id password hash)
+--   issued_credentials   — VC issuance log (learner_ref is opaque, no FK)
+--   mastery_gate_events  — mastery gate audit log (REQ-NFR-MAST-02)
+--   cohort_aggregates    — k-anonymized cohort metrics (plain table, D-050)
+--   issuer_keys          — Ed25519 issuer key lifecycle (active/superseded)
+--
+-- No cross-DB FKs (D-031). learner_ref is an opaque string in Postgres.
+
+CREATE TABLE IF NOT EXISTS operators (
+    id            UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    username      TEXT UNIQUE NOT NULL,
+    password_hash TEXT NOT NULL,
+    display_name  TEXT,
+    role          TEXT NOT NULL DEFAULT 'operator',
+    is_active     BOOLEAN NOT NULL DEFAULT TRUE,
+    created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
+    last_login_at TIMESTAMPTZ
+);
+
+CREATE TABLE IF NOT EXISTS issued_credentials (
+    id            UUID PRIMARY KEY,
+    operator_id   UUID REFERENCES operators(id),
+    learner_ref   TEXT NOT NULL,
+    vc_type       TEXT,
+    payload_jsonb JSONB NOT NULL,
+    signature_b64 TEXT NOT NULL,
+    status        TEXT NOT NULL DEFAULT 'active',
+    issued_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
+    revoked_at    TIMESTAMPTZ
+);
+
+CREATE TABLE IF NOT EXISTS mastery_gate_events (
+    id                  UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    learner_ref         TEXT NOT NULL,
+    scenario_id         TEXT,
+    path_id             TEXT NOT NULL,
+    gate_outcome        TEXT,
+    rubric_scores_jsonb JSONB,
+    recorded_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
+    source              TEXT NOT NULL DEFAULT 'sync'
+);
+
+CREATE TABLE IF NOT EXISTS cohort_aggregates (
+    path           TEXT NOT NULL,
+    metric         TEXT NOT NULL,
+    window_start   DATE NOT NULL,
+    window_end     DATE NOT NULL,
+    value          NUMERIC,
+    cell_count     INTEGER NOT NULL DEFAULT 0,
+    cell_suppressed BOOLEAN NOT NULL DEFAULT FALSE,
+    updated_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (path, metric, window_start)
+);
+-- Plain table, NOT partitioned (D-050..D-053; add partitioning post-pilot).
+CREATE INDEX IF NOT EXISTS cohort_aggregates_path_window_idx
+    ON cohort_aggregates (path, window_start);
+
+CREATE TABLE IF NOT EXISTS issuer_keys (
+    id               TEXT PRIMARY KEY,
+    public_key       TEXT NOT NULL,
+    private_key_enc  BYTEA,
+    status           TEXT NOT NULL DEFAULT 'active',
+    created_at       TIMESTAMPTZ NOT NULL DEFAULT now()
+);
@@ -0,0 +1,280 @@
+"""Postgres store — operator-tier access layer (D-040, D-050, TASK-01-06).
+
+Async access via an asyncpg.Pool. Implements the IssuerKeyStore protocol
+(server/vc/issuer_keys.py) so VC verification can use either PraxisStore
+(SQLite, v0.3) or PgStore (Postgres, v0.4). No cross-DB joins (D-031);
+`learner_ref` is an opaque string in Postgres (not a FK to SQLite).
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from typing import Any
+
+import asyncpg
+
+
+class PgStore:
+    """Async Postgres store for the v0.4 operator tier."""
+
+    def __init__(self, pool: asyncpg.Pool) -> None:
+        self.pool = pool
+
+    # ── Operator CRUD ────────────────────────────────────────────────────
+
+    async def get_operator_by_username(self, username: str) -> dict | None:
+        async with self.pool.acquire() as conn:
+            row = await conn.fetchrow(
+                "SELECT id, username, password_hash, display_name, role, "
+                "is_active, created_at, last_login_at "
+                "FROM operators WHERE username = $1",
+                username,
+            )
+            return dict(row) if row else None
+
+    async def get_operator_by_id(self, operator_id: str) -> dict | None:
+        async with self.pool.acquire() as conn:
+            row = await conn.fetchrow(
+                "SELECT id, username, password_hash, display_name, role, "
+                "is_active, created_at, last_login_at "
+                "FROM operators WHERE id = $1",
+                operator_id,
+            )
+            return dict(row) if row else None
+
+    async def update_last_login(self, operator_id: str) -> None:
+        async with self.pool.acquire() as conn:
+            await conn.execute(
+                "UPDATE operators SET last_login_at = now() WHERE id = $1",
+                operator_id,
+            )
+
+    async def insert_operator(
+        self,
+        username: str,
+        password_hash: str,
+        display_name: str | None = None,
+        *,
+        on_conflict_update: bool = False,
+    ) -> str | None:
+        """Insert an operator (idempotent on username). Returns the id, or
+        None if the row already existed and on_conflict_update is False."""
+        async with self.pool.acquire() as conn:
+            if on_conflict_update:
+                row = await conn.fetchrow(
+                    "INSERT INTO operators (username, password_hash, display_name) "
+                    "VALUES ($1, $2, $3) "
+                    "ON CONFLICT (username) DO UPDATE SET "
+                    "password_hash = excluded.password_hash, "
+                    "display_name = excluded.display_name "
+                    "RETURNING id",
+                    username,
+                    password_hash,
+                    display_name,
+                )
+                return str(row["id"]) if row else None
+            row = await conn.fetchrow(
+                "INSERT INTO operators (username, password_hash, display_name) "
+                "VALUES ($1, $2, $3) "
+                "ON CONFLICT (username) DO NOTHING "
+                "RETURNING id",
+                username,
+                password_hash,
+                display_name,
+            )
+            return str(row["id"]) if row else None
+
+    # ── Cohort aggregate read/write ──────────────────────────────────────
+
+    async def get_cohort_aggregates(
+        self,
+        path: str,
+        metric: str,
+        since_date: Any,
+    ) -> list[dict]:
+        async with self.pool.acquire() as conn:
+            rows = await conn.fetch(
+                "SELECT path, metric, window_start, window_end, value, "
+                "cell_count, cell_suppressed, updated_at "
+                "FROM cohort_aggregates "
+                "WHERE path = $1 AND metric = $2 AND window_start >= $3 "
+                "ORDER BY window_start",
+                path,
+                metric,
+                since_date,
+            )
+            return [dict(r) for r in rows]
+
+    async def upsert_cohort_aggregate(
+        self,
+        path: str,
+        metric: str,
+        window_start: Any,
+        window_end: Any,
+        value: float | None,
+        cell_count: int,
+        cell_suppressed: bool,
+    ) -> None:
+        async with self.pool.acquire() as conn:
+            await conn.execute(
+                "INSERT INTO cohort_aggregates "
+                "(path, metric, window_start, window_end, value, cell_count, "
+                "cell_suppressed, updated_at) "
+                "VALUES ($1, $2, $3, $4, $5, $6, $7, now()) "
+                "ON CONFLICT (path, metric, window_start) DO UPDATE SET "
+                "window_end = excluded.window_end, value = excluded.value, "
+                "cell_count = excluded.cell_count, "
+                "cell_suppressed = excluded.cell_suppressed, "
+                "updated_at = now()",
+                path,
+                metric,
+                window_start,
+                window_end,
+                value,
+                cell_count,
+                cell_suppressed,
+            )
+
+    # ── IssuerKeyStore protocol (D-051, TASK-04-02) ──────────────────────
+
+    async def init_issuer_key(
+        self,
+        key_id: str,
+        public_key: str,
+        private_key_enc: bytes | None,
+    ) -> None:
+        async with self.pool.acquire() as conn:
+            await conn.execute(
+                "INSERT INTO issuer_keys (id, public_key, private_key_enc, status) "
+                "VALUES ($1, $2, $3, 'active') "
+                "ON CONFLICT (id) DO NOTHING",
+                key_id,
+                public_key,
+                private_key_enc if private_key_enc is not None else b"",
+            )
+
+    async def get_active_signing_key_row(self) -> dict | None:
+        async with self.pool.acquire() as conn:
+            row = await conn.fetchrow(
+                "SELECT id, public_key, private_key_enc, status, created_at "
+                "FROM issuer_keys WHERE status = 'active' "
+                "ORDER BY created_at DESC LIMIT 1"
+            )
+            return dict(row) if row else None
+
+    async def get_public_key_row(self, key_id: str) -> dict | None:
+        # Queries by id (NOT status) so superseded keys are found too —
+        # this is the R-VC-MIG-01 verification fallback (D-051).
+        async with self.pool.acquire() as conn:
+            row = await conn.fetchrow(
+                "SELECT id, public_key, private_key_enc, status, created_at "
+                "FROM issuer_keys WHERE id = $1",
+                key_id,
+            )
+            return dict(row) if row else None
+
+    async def set_issuer_key_superseded(self, key_id: str) -> None:
+        async with self.pool.acquire() as conn:
+            await conn.execute(
+                "UPDATE issuer_keys SET status = 'superseded' WHERE id = $1",
+                key_id,
+            )
+
+    # ── Credential methods ───────────────────────────────────────────────
+
+    async def insert_credential(
+        self,
+        cred_id: str,
+        learner_ref: str,
+        payload_json: str,
+        signature_b64: str,
+        *,
+        operator_id: str | None = None,
+        vc_type: str = "MasteryCredential",
+    ) -> None:
+        async with self.pool.acquire() as conn:
+            await conn.execute(
+                "INSERT INTO issued_credentials "
+                "(id, operator_id, learner_ref, vc_type, payload_jsonb, "
+                "signature_b64, status) "
+                "VALUES ($1, $2, $3, $4, $5::jsonb, $6, 'active')",
+                cred_id,
+                operator_id,
+                learner_ref,
+                vc_type,
+                payload_json,
+                signature_b64,
+            )
+
+    async def get_credential(self, cred_id: str) -> dict | None:
+        # Returns a row shaped like PraxisStore.get_credential so the
+        # verification code can use either store interchangeably.
+        async with self.pool.acquire() as conn:
+            row = await conn.fetchrow(
+                "SELECT id, learner_ref, "
+                "payload_jsonb::text AS vc_payload_json, signature_b64, "
+                "status, issued_at "
+                "FROM issued_credentials WHERE id = $1",
+                cred_id,
+            )
+            return dict(row) if row else None
+
+    async def set_credential_status(self, cred_id: str, status: str) -> None:
+        extra = ", revoked_at = now()" if status == "revoked" else ""
+        async with self.pool.acquire() as conn:
+            await conn.execute(
+                f"UPDATE issued_credentials SET status = $1{extra} WHERE id = $2",
+                status,
+                cred_id,
+            )
+
+    async def list_credentials(self, operator_id: str | None = None) -> list[dict]:
+        async with self.pool.acquire() as conn:
+            if operator_id is None:
+                rows = await conn.fetch(
+                    "SELECT id, learner_ref, vc_type, status, issued_at, "
+                    "revoked_at FROM issued_credentials ORDER BY issued_at DESC"
+                )
+            else:
+                rows = await conn.fetch(
+                    "SELECT id, learner_ref, vc_type, status, issued_at, "
+                    "revoked_at FROM issued_credentials "
+                    "WHERE operator_id = $1 ORDER BY issued_at DESC",
+                    operator_id,
+                )
+            return [dict(r) for r in rows]
+
+    # ── Mastery gate event ───────────────────────────────────────────────
+
+    async def record_gate_event(
+        self,
+        learner_ref: str,
+        path_id: str,
+        scenario_id: str | None = None,
+        gate_outcome: str | None = None,
+        rubric_scores_jsonb: Any | None = None,
+    ) -> str:
+        event_id = str(uuid.uuid4())
+        scores_json = (
+            rubric_scores_jsonb
+            if isinstance(rubric_scores_jsonb, str)
+            else (json.dumps(rubric_scores_jsonb) if rubric_scores_jsonb is not None else None)
+        )
+        async with self.pool.acquire() as conn:
+            await conn.execute(
+                "INSERT INTO mastery_gate_events "
+                "(id, learner_ref, scenario_id, path_id, gate_outcome, "
+                "rubric_scores_jsonb, source) "
+                "VALUES ($1, $2, $3, $4, $5, $6::jsonb, 'sync')",
+                event_id,
+                learner_ref,
+                scenario_id,
+                path_id,
+                gate_outcome,
+                scores_json,
+            )
+        return event_id
+
+
+__all__ = ["PgStore"]
@@ -0,0 +1,221 @@
+"""PgStore + asyncpg pool integration test (TASK-01-07).
+
+Requires a live Postgres instance. Skips gracefully when PRAXIS_PG_DSN is
+unset so the test suite has no hard CI dependency on Postgres.
+"""
+
+from __future__ import annotations
+
+import os
+import uuid
+from datetime import date
+
+import asyncpg
+import pytest
+
+from db.pg_migrate import apply_pg_migrations
+from db.pg_store import PgStore
+
+pytestmark = pytest.mark.skipif(
+    "PRAXIS_PG_DSN" not in os.environ,
+    reason="PRAXIS_PG_DSN not set — Postgres integration tests skipped (dev mode).",
+)
+
+
+@pytest.fixture(scope="module")
+async def pool() -> asyncpg.Pool:
+    p = await asyncpg.create_pool(
+        dsn=os.environ["PRAXIS_PG_DSN"],
+        min_size=1,
+        max_size=5,
+        command_timeout=10,
+    )
+    try:
+        await apply_pg_migrations(p)
+        yield p
+    finally:
+        await p.close()
+
+
+@pytest.fixture(autouse=True)
+async def _clean_tables(pool: asyncpg.Pool):
+    async with pool.acquire() as conn:
+        await conn.execute(
+            "TRUNCATE operators, issued_credentials, mastery_gate_events, "
+            "cohort_aggregates, issuer_keys RESTART IDENTITY CASCADE"
+        )
+    yield
+
+
+@pytest.mark.asyncio
+async def test_migration_creates_tables(pool: asyncpg.Pool):
+    async with pool.acquire() as conn:
+        tables = await conn.fetch(
+            "SELECT tablename FROM pg_tables WHERE schemaname = 'public' "
+            "ORDER BY tablename"
+        )
+        names = {r["tablename"] for r in tables}
+        assert {"operators", "issued_credentials", "mastery_gate_events",
+                "cohort_aggregates", "issuer_keys"}.issubset(names)
+
+
+@pytest.mark.asyncio
+async def test_migration_idempotent(pool: asyncpg.Pool):
+    applied = await apply_pg_migrations(pool)
+    assert applied == []
+
+
+@pytest.mark.asyncio
+async def test_operator_insert_and_lookup(pool: asyncpg.Pool):
+    store = PgStore(pool)
+    oid = await store.insert_operator(
+        "alice", "$argon2id$fakehash", "Alice"
+    )
+    assert oid is not None
+    op = await store.get_operator_by_username("alice")
+    assert op is not None
+    assert op["username"] == "alice"
+    assert op["display_name"] == "Alice"
+    assert op["is_active"] is True
+    by_id = await store.get_operator_by_id(oid)
+    assert by_id is not None
+    assert by_id["id"] == op["id"]
+
+
+@pytest.mark.asyncio
+async def test_operator_insert_idempotent(pool: asyncpg.Pool):
+    store = PgStore(pool)
+    first = await store.insert_operator("bob", "$argon2id$h1", "Bob")
+    assert first is not None
+    second = await store.insert_operator("bob", "$argon2id$h2", "Bob")
+    assert second is None
+
+
+@pytest.mark.asyncio
+async def test_operator_on_conflict_update(pool: asyncpg.Pool):
+    store = PgStore(pool)
+    await store.insert_operator("carol", "$argon2id$old", "Carol")
+    updated = await store.insert_operator(
+        "carol", "$argon2id$new", "Carol", on_conflict_update=True
+    )
+    assert updated is not None
+    op = await store.get_operator_by_username("carol")
+    assert op["password_hash"] == "$argon2id$new"
+
+
+@pytest.mark.asyncio
+async def test_update_last_login(pool: asyncpg.Pool):
+    store = PgStore(pool)
+    oid = await store.insert_operator("dave", "$argon2id$h", "Dave")
+    assert oid is not None
+    assert (await store.get_operator_by_id(oid))["last_login_at"] is None
+    await store.update_last_login(oid)
+    assert (await store.get_operator_by_id(oid))["last_login_at"] is not None
+
+
+@pytest.mark.asyncio
+async def test_cohort_aggregate_upsert_idempotent(pool: asyncpg.Pool):
+    store = PgStore(pool)
+    ws, we = date(2026, 8, 1), date(2026, 8, 7)
+    await store.upsert_cohort_aggregate(
+        "cs-refund", "sessions_count", ws, we, 42.0, 15, False
+    )
+    await store.upsert_cohort_aggregate(
+        "cs-refund", "sessions_count", ws, we, 42.0, 15, False
+    )
+    rows = await store.get_cohort_aggregates(
+        "cs-refund", "sessions_count", date(2026, 7, 1)
+    )
+    assert len(rows) == 1
+    assert rows[0]["value"] == 42.0
+    assert rows[0]["cell_count"] == 15
+
+
+@pytest.mark.asyncio
+async def test_cohort_aggregate_suppressed_cell(pool: asyncpg.Pool):
+    store = PgStore(pool)
+    ws, we = date(2026, 8, 1), date(2026, 8, 7)
+    await store.upsert_cohort_aggregate(
+        "cs-refund", "active_learners", ws, we, None, 9, True
+    )
+    rows = await store.get_cohort_aggregates(
+        "cs-refund", "active_learners", date(2026, 7, 1)
+    )
+    assert len(rows) == 1
+    assert rows[0]["cell_suppressed"] is True
+    assert rows[0]["value"] is None
+
+
+@pytest.mark.asyncio
+async def test_issuer_key_init_active_then_superseded(pool: asyncpg.Pool):
+    store = PgStore(pool)
+    kid = f"key-{uuid.uuid4().hex[:12]}"
+    await store.init_issuer_key(kid, "pub-b64-aaa", b"\x01\x02\x03")
+    active = await store.get_active_signing_key_row()
+    assert active is not None
+    assert active["id"] == kid
+    assert active["status"] == "active"
+    await store.set_issuer_key_superseded(kid)
+    assert await store.get_active_signing_key_row() is None
+    archived = await store.get_public_key_row(kid)
+    assert archived is not None
+    assert archived["status"] == "superseded"
+    assert archived["public_key"] == "pub-b64-aaa"
+
+
+@pytest.mark.asyncio
+async def test_get_public_key_row_finds_superseded(pool: asyncpg.Pool):
+    store = PgStore(pool)
+    kid = f"key-{uuid.uuid4().hex[:12]}"
+    await store.init_issuer_key(kid, "pub-b64-bbb", b"\x04\x05")
+    await store.set_issuer_key_superseded(kid)
+    row = await store.get_public_key_row(kid)
+    assert row is not None
+    assert row["status"] == "superseded"
+
+
+@pytest.mark.asyncio
+async def test_credential_insert_and_get(pool: asyncpg.Pool):
+    store = PgStore(pool)
+    oid = await store.insert_operator("ed", "$argon2id$h", "Ed")
+    cid = f"vc-{uuid.uuid4().hex[:16]}"
+    await store.insert_credential(
+        cid, "learner-1", '{"id":"vc-x"}', "sig-b64",
+        operator_id=oid,
+    )
+    row = await store.get_credential(cid)
+    assert row is not None
+    assert row["id"] == cid
+    assert row["learner_ref"] == "learner-1"
+    assert row["signature_b64"] == "sig-b64"
+    assert row["status"] == "active"
+    assert row["vc_payload_json"] == '{"id":"vc-x"}'
+
+
+@pytest.mark.asyncio
+async def test_credential_status_revoke(pool: asyncpg.Pool):
+    store = PgStore(pool)
+    cid = f"vc-{uuid.uuid4().hex[:16]}"
+    await store.insert_credential(cid, "learner-2", "{}", "sig")
+    await store.set_credential_status(cid, "revoked")
+    row = await store.get_credential(cid)
+    assert row["status"] == "revoked"
+    assert row["revoked_at"] is not None
+
+
+@pytest.mark.asyncio
+async def test_record_gate_event(pool: asyncpg.Pool):
+    store = PgStore(pool)
+    eid = await store.record_gate_event(
+        "learner-3", "cs-refund", scenario_id="sc-1",
+        gate_outcome="open", rubric_scores_jsonb=[{"c": "x", "l": 4}],
+    )
+    assert eid is not None
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            "SELECT * FROM mastery_gate_events WHERE id = $1", eid
+        )
+        assert row is not None
+        assert row["learner_ref"] == "learner-3"
+        assert row["gate_outcome"] == "open"
+        assert row["source"] == "sync"