feat(compliance): documented allowlist-screen exceptions (waivers)

keel/cli.py

@@ -530,7 +530,8 @@ def _screen_product(
         if raw is not None
         else None
     )
-    return facts, screen_mod.screen_asset(facts, attestation)
+    waived = repo.get_screen_exceptions(asset)
+    return facts, screen_mod.screen_asset(facts, attestation, waived=waived)
 
 
 # Failure classes that are DOWNSTREAM of having no cached history: with zero bars `liquidity`
@@ -820,20 +821,102 @@ def assets_attest(
     click.echo(f"attested {asset}: sector={sector} backing={backing} pays_yield={pays_yield}")
 
 
+@assets_group.command("exempt")
+@click.option("--asset", required=True, help="Asset code, e.g. PAXG.")
+@click.option(
+    "--criterion",
+    required=True,
+    type=click.Choice(sorted(screen_mod.WAIVABLE_CRITERIA)),
+    help="The admission criterion to waive. Restricted to WAIVABLE_CRITERIA -- a DATA/market "
+    "criterion, never a shariah one.",
+)
+@click.option("--rationale", required=True, help="Why this waiver is granted.")
+@click.option("--granted-by", required=True, help="Who granted it.")
+@click.pass_context
+@with_disclaimer
+def assets_exempt(
+    ctx: click.Context, asset: str, criterion: str, rationale: str, granted_by: str
+) -> None:
+    """Record a DOCUMENTED exception waiving one admission criterion for one asset.
+
+    This waives ONLY a computed DATA/market criterion (history depth today) -- never a shariah
+    one (a missing attestation, haram sector, riba yield, or dayn/unknown backing): the
+    `--criterion` Choice is restricted to `screen_mod.WAIVABLE_CRITERIA`, so this command cannot
+    reach those checks no matter what is typed. The exception is recorded and then surfaced
+    loudly by `keel assets screen` as a WARNING, never silently -- it is not a default pass. It
+    is also self-retiring: once the underlying condition it was granted for no longer holds (the
+    asset accumulates enough history, say), `screen_asset` stops mentioning it at all.
+
+    Not passphrase-gated, for the same reason `keel assets attest` is not: recording an exception
+    cannot itself place an order or raise a cap, and the screen it feeds only ever ADMITS to a
+    list that `guards.py` rail 1 still enforces per-trade regardless.
+    """
+    if not rationale.strip():
+        # Mirrors the unsourced-attestation guard in `screen_asset` (`if not
+        # attestation.source.strip()`): an unsourced claim is not evidence, and a blank rationale
+        # is not documentation -- it would be an "undocumented documented exception."
+        raise click.BadParameter(
+            "rationale must be a non-empty documented reason", param_hint="--rationale"
+        )
+    asset = asset.upper()  # matches the uppercase asset `_screen_product` looks waivers up by
+    repo = _open_repo(ctx)
+    repo.upsert_screen_exception(
+        asset=asset,
+        criterion=criterion,
+        rationale=rationale,
+        granted_by=granted_by,
+        granted_at=int(time.time()),
+    )
+    click.echo(f"recorded exception: {asset} waives '{criterion}' criterion (by {granted_by})")
+
+
+@assets_group.command("unexempt")
+@click.option("--asset", required=True, help="Asset code, e.g. PAXG.")
+@click.option(
+    "--criterion",
+    required=True,
+    type=click.Choice(sorted(screen_mod.WAIVABLE_CRITERIA)),
+    help="The waived criterion to revoke.",
+)
+@click.pass_context
+@with_disclaimer
+def assets_unexempt(ctx: click.Context, asset: str, criterion: str) -> None:
+    """Revoke a documented allowlist-screen exception. A de-risking action, always allowed.
+
+    After this, `keel assets screen` re-evaluates the criterion normally -- if it still fails,
+    the asset is rejected again.
+    """
+    asset = asset.upper()  # matches the uppercase asset `assets exempt` records under
+    repo = _open_repo(ctx)
+    removed = repo.delete_screen_exception(asset, criterion)
+    if removed:
+        click.echo(f"revoked exception: {asset} no longer waives '{criterion}' criterion")
+    else:
+        click.echo(f"no such exception: {asset} has no '{criterion}' waiver")
+
+
 @assets_group.command("list")
 @click.pass_context
 def assets_list(ctx: click.Context) -> None:
-    """List recorded attestations."""
+    """List recorded attestations and any documented screen exceptions."""
     repo = _open_repo(ctx)
     rows = repo.get_asset_attestations()
-    if not rows:
+    exceptions = repo.list_screen_exceptions()
+    if not rows and not exceptions:
         click.echo("no attestations recorded")
         return
     for row in rows:
         click.echo(
             f"{row['asset']:<8} sector={row['sector']:<16} backing={row['backing']:<8} "
             f"pays_yield={bool(row['pays_yield'])!s:<5} by={row['attested_by']}"
         )
+    if exceptions:
+        click.echo("\nexceptions:")
+        for row in exceptions:
+            click.echo(
+                f"{row['asset']:<8} waives={row['criterion']:<10} by={row['granted_by']} -- "
+                f"{row['rationale']}"
+            )
 
 
 # -- withdrawals ------------------------------------------------------------------------------

keel/compliance/screen.py

@@ -22,6 +22,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping
 from dataclasses import dataclass, field
 from decimal import Decimal
 
@@ -46,6 +47,15 @@
 BACKING_NATIVE = "native"  # a base-layer coin, neither a claim nor a warehouse receipt
 KNOWN_BACKINGS = frozenset({BACKING_AYN, BACKING_DAYN, BACKING_NATIVE})
 
+#: Criteria a documented, human-recorded exception (`keel assets exempt`) may EVER waive. Only
+#: DATA/market criteria belong here -- history depth, liquidity, that kind of thing -- because
+#: those are facts about our own cache, not about the asset's shariah status. The shariah
+#: criteria (a missing attestation, `haram_sector`, `riba_yield`, `dayn`/unknown backing) and
+#: `settlement` can NEVER be waived: nothing in this module consults `waived` for them, and the
+#: CLI's `--criterion` Choice is restricted to this set. Expanding it is a deliberate future
+#: decision, not a default -- do not add to it to make a test pass.
+WAIVABLE_CRITERIA = frozenset({"history"})
+
 
 @dataclass(frozen=True)
 class AssetAttestation:
@@ -100,18 +110,45 @@ def screen_asset(
     facts: MarketFacts,
     attestation: AssetAttestation | None,
     policy: ScreenPolicy | None = None,
+    waived: Mapping[str, str] | None = None,
 ) -> ScreenResult:
-    """Deterministic admission decision. `attestation=None` fails closed."""
+    """Deterministic admission decision. `attestation=None` fails closed.
+
+    `waived` is `{criterion: rationale}` from a documented human exception (`keel assets
+    exempt` / `repository.get_screen_exceptions`). It is consulted ONLY when a check would
+    otherwise FAIL, and ONLY for criteria in `WAIVABLE_CRITERIA` -- a waiver for anything else
+    (a stray `screen_exceptions` row for, say, `attestation`) is silently ignored and that
+    criterion still fails closed. A waiver never affects any criterion other than its own, and a
+    blank/whitespace rationale is treated as no waiver at all (fail closed -- see the `.strip()`
+    check below, mirroring the unsourced-attestation guard further down).
+    """
     policy = policy or ScreenPolicy()
+    # Filtered ONCE, up front, rather than inline per-branch: this is the actual defense-in-depth
+    # for a criterion that is not in WAIVABLE_CRITERIA. `history` is currently the SOLE consumer
+    # of a waiver (screen_asset only ever reads `effective_waived["history"]`, so a shariah check
+    # is already structurally unreachable from `waived`) -- but filtering here means that even if
+    # a future edit wires a waiver lookup into another branch, it can never see an entry for a
+    # criterion nobody was allowed to grant one for, because it was dropped before any branch ran.
+    effective_waived = {c: r for c, r in (waived or {}).items() if c in WAIVABLE_CRITERIA}
     failures: list[str] = []
     warnings: list[str] = []
 
     # -- computed market facts -------------------------------------------------
     if facts.daily_bars < policy.min_daily_bars:
-        failures.append(
-            f"history: {facts.daily_bars} daily bars < {policy.min_daily_bars} required "
-            "(a rule cannot be validated on a series shorter than its evidence needs)"
-        )
+        history_rationale = effective_waived.get("history", "").strip()
+        if history_rationale:
+            # Self-retiring: this branch is only reached when the check WOULD fail, so a stale
+            # waiver on an asset that has since accumulated enough history produces no output at
+            # all -- see the `>=` branch below, which never looks at `effective_waived`.
+            warnings.append(
+                f"history: {facts.daily_bars} daily bars < {policy.min_daily_bars} required -- "
+                f"WAIVED by documented exception: {history_rationale}"
+            )
+        else:
+            failures.append(
+                f"history: {facts.daily_bars} daily bars < {policy.min_daily_bars} required "
+                "(a rule cannot be validated on a series shorter than its evidence needs)"
+            )
     if facts.median_daily_volume < policy.min_median_daily_volume:
         failures.append(
             f"liquidity: median daily volume {facts.median_daily_volume} < "

keel/data/db.py

@@ -19,7 +19,7 @@
 from pathlib import Path
 from typing import Any
 
-SCHEMA_VERSION = 8
+SCHEMA_VERSION = 9
 
 # Creation order matters for readability (and for backends that validate FK targets eagerly);
 # SQLite itself only checks FK targets at DML time, but we still declare referenced tables first.
@@ -239,6 +239,16 @@
     )
     """,
     """
+    CREATE TABLE IF NOT EXISTS screen_exceptions (
+        asset       TEXT NOT NULL,
+        criterion   TEXT NOT NULL,
+        rationale   TEXT NOT NULL,
+        granted_by  TEXT NOT NULL,
+        granted_at  INTEGER NOT NULL,
+        PRIMARY KEY (asset, criterion)
+    )
+    """,
+    """
     CREATE TABLE IF NOT EXISTS profile (
         id          INTEGER PRIMARY KEY CHECK (id = 1),
         autonomous  INTEGER NOT NULL DEFAULT 0,
@@ -395,6 +405,21 @@ def _migrate_v8_autonomy_expiry(conn: sqlite3.Connection) -> None:
         conn.execute("ALTER TABLE profile ADD COLUMN autonomous_until INTEGER")
 
 
+def _migrate_v9_screen_exceptions(conn: sqlite3.Connection) -> None:
+    """v9 adds `screen_exceptions`. Table creation is handled by `_SCHEMA_STATEMENTS`; there is
+    deliberately NO backfill.
+
+    A row asserts a human documented a waiver for one asset/criterion pair, with a rationale and
+    who granted it. Seeding one would fabricate an exception nobody granted; an empty table
+    correctly says nothing has been excepted yet.
+
+    Unlike v8's `profile.autonomous_until` ADD COLUMN, this is a genuine no-op: `migrate()` runs
+    every `_SCHEMA_STATEMENTS` statement (all `IF NOT EXISTS`) before the version loop below, so
+    a database already stamped at v8 picks up the new table from that pass alone. This step only
+    exists to advance the stamp.
+    """
+
+
 _MIGRATIONS: dict[int, Callable[[sqlite3.Connection], None]] = {
     2: _migrate_v2_broker_subscriptions,
     3: _migrate_v3_trade_outcomes,
@@ -403,6 +428,7 @@ def _migrate_v8_autonomy_expiry(conn: sqlite3.Connection) -> None:
     6: _migrate_v6_asset_attestations,
     7: _migrate_v7_profile,
     8: _migrate_v8_autonomy_expiry,
+    9: _migrate_v9_screen_exceptions,
 }
 
 

keel/data/repository.py

@@ -741,3 +741,51 @@ def get_asset_attestation(self, asset: str) -> dict | None:
     def get_asset_attestations(self) -> list[dict]:
         rows = self._conn.execute("SELECT * FROM asset_attestations ORDER BY asset").fetchall()
         return [dict(row) for row in rows]
+
+    # -- screen exceptions ------------------------------------------------------
+    # Documented, per-asset per-criterion waivers of an allowlist-screen admission criterion (KB
+    # PAXG/history case). See `keel/compliance/screen.py` -- only criteria in `WAIVABLE_CRITERIA`
+    # are ever honoured, so a row here can never bypass the shariah core.
+
+    def upsert_screen_exception(
+        self,
+        asset: str,
+        criterion: str,
+        rationale: str,
+        granted_by: str,
+        granted_at: int,
+    ) -> None:
+        self._conn.execute(
+            """
+            INSERT INTO screen_exceptions
+                (asset, criterion, rationale, granted_by, granted_at)
+            VALUES (?, ?, ?, ?, ?)
+            ON CONFLICT(asset, criterion) DO UPDATE SET
+                rationale = excluded.rationale,
+                granted_by = excluded.granted_by,
+                granted_at = excluded.granted_at
+            """,
+            (asset, criterion, rationale, granted_by, granted_at),
+        )
+        self._conn.commit()
+
+    def get_screen_exceptions(self, asset: str) -> dict[str, str]:
+        rows = self._conn.execute(
+            "SELECT criterion, rationale FROM screen_exceptions WHERE asset = ?", (asset,)
+        ).fetchall()
+        return {row["criterion"]: row["rationale"] for row in rows}
+
+    def list_screen_exceptions(self) -> list[dict]:
+        rows = self._conn.execute(
+            "SELECT * FROM screen_exceptions ORDER BY asset, criterion"
+        ).fetchall()
+        return [dict(row) for row in rows]
+
+    def delete_screen_exception(self, asset: str, criterion: str) -> int:
+        """Returns the number of rows removed (0 or 1), so a caller can tell a real revoke from
+        a no-op on a row that never existed rather than echoing success either way."""
+        cursor = self._conn.execute(
+            "DELETE FROM screen_exceptions WHERE asset = ? AND criterion = ?", (asset, criterion)
+        )
+        self._conn.commit()
+        return cursor.rowcount