TEN-240 Phase 7: Dual-write artifact storage (Hippius + R2)

.github/workflows/schema-migrations.yml

@@ -79,6 +79,14 @@ jobs:
         with:
           fetch-depth: 0
       - name: Scan new migrations for destructive patterns
+        env:
+          # Pass the PR body via env rather than ${{ }} interpolation —
+          # bodies routinely contain backticks, $vars, and quotes that
+          # break the runner's bash script when interpolated raw (the
+          # script source ends up containing PR-body text and bash treats
+          # backticks as command substitution). See PR #142 (TEN-240
+          # Phase 7) which tripped exit code 2 from a backtick-rich body.
+          PR_BODY: ${{ github.event.pull_request.body }}
         run: |
           NEW=$(git diff --diff-filter=A origin/${{ github.base_ref }}...HEAD \
                   -- 'database/migrations/*.sql' || true)
@@ -89,7 +97,7 @@ jobs:
           CONTENT=$(git diff --diff-filter=A origin/${{ github.base_ref }}...HEAD \
                       -- 'database/migrations/*.sql')
           if echo "$CONTENT" | grep -iE '^\+.*(DROP TABLE|DROP COLUMN|ALTER COLUMN [^,;]+ TYPE)' > /dev/null; then
-            if ! echo "${{ github.event.pull_request.body }}" | grep -q "ALLOW_DESTRUCTIVE"; then
+            if ! printf '%s' "$PR_BODY" | grep -q "ALLOW_DESTRUCTIVE"; then
               echo "::error::Destructive migration detected. Add 'ALLOW_DESTRUCTIVE' to the PR description to approve."
               exit 1
             fi

CLAUDE.md

@@ -82,7 +82,7 @@ scripts/         Localnet test script + Postgres startup
 | `shared/dedup.py` | Code similarity (provenance queries) |
 | `shared/scoring.py` | Size-gated Pareto frontier scoring (Phase C) |
 | `shared/commitment.py` | On-chain Docker image + endpoint commitment |
-| `shared/r2_audit.py` | R2 storage for checkpoints, snapshots, dispatch records |
+| `shared/r2_audit.py` | S3-compatible artifact storage. Hippius (Substrate) is the primary backend; Cloudflare R2 stays supported as a legacy fallback. Exports `HippiusStorage` (preferred) and `R2AuditLog` (alias). |
 | `database/server.py` | Centralized DB API (FastAPI, all experiment routes) |
 | `database/neuron.py` | Subnet owner process (Postgres + API server) |
 | `validator/neuron.py` | Main validator loop (3-phase: collect → train → evaluate) |
@@ -235,6 +235,29 @@ frontier/latest.json                                        # Current Pareto fro
 - [x] Validator reverse proxy for miners (`validator/db_proxy.py`)
 - [x] DatabaseClient for validator -> DB server communication (`shared/db_client.py`)
 - [x] Validator / dashboard deploy split (`RADAR_NEURON_MODE`)
+- [x] Hippius (Substrate) artifact backend with R2 legacy fallback (`shared/r2_audit.py`)
+
+## Artifact Storage (Hippius / R2)
+
+Primary backend is **Hippius**, a Substrate-based decentralized object store
+with an S3-compatible API at `https://s3.hippius.com`. Cloudflare R2 stays
+supported as a legacy fallback so existing deployments keep working without
+an env-var change. The client (`shared.r2_audit.HippiusStorage`, aliased as
+`R2AuditLog`) reads `HIPPIUS_*` first, then falls back to `R2_*`:
+
+| Var | Default | Purpose |
+|-----|---------|---------|
+| `HIPPIUS_ACCESS_KEY_ID` | `R2_ACCESS_KEY_ID` | Hippius access key (begins with `hip_`). Falls back to the legacy R2 var. |
+| `HIPPIUS_SECRET_ACCESS_KEY` | `R2_SECRET_ACCESS_KEY` | Hippius secret. Falls back to the legacy R2 var. |
+| `HIPPIUS_BUCKET` | `R2_BUCKET` | Bucket name for experiment artifacts. |
+| `HIPPIUS_REGION` | `decentralized` | Required by Hippius; leave as default. |
+| `HIPPIUS_ENDPOINT_URL` | `https://s3.hippius.com` | Override for private gateways. The legacy `R2_ACCOUNT_ID`-derived endpoint is only used when no `HIPPIUS_*` vars are set. |
+| `MOCK_S3_ENDPOINT` | unset | Localnet override (`scripts/mock_r2_server.py`). `MOCK_R2_ENDPOINT` remains an alias for back-compat. |
+
+Operators migrating from R2: leave the existing `R2_*` vars in place and
+add `HIPPIUS_*` once your Hippius keys are issued — the client will pick up
+the new backend automatically. To cut over fully, also set
+`HIPPIUS_BUCKET` (or unset `R2_BUCKET`) so dispatched URLs point at Hippius.
 
 ## Neuron Mode Environment Variables
 

README.md

@@ -166,7 +166,7 @@ python validator/neuron.py --netuid <N> --subtensor.network <network> --wallet.n
 
 ### Infrastructure Requirements
 
-* **R2 storage** for checkpoint storage and artifact sharing. Set `R2_ACCOUNT_ID`, `R2_ACCESS_KEY_ID`, `R2_SECRET_ACCESS_KEY`, `R2_BUCKET` env vars.
+* **Artifact storage** (S3-compatible) for checkpoints and cross-validator sharing. The default backend is **Hippius** (Substrate-based decentralized object store at `https://s3.hippius.com`); set `HIPPIUS_ACCESS_KEY_ID`, `HIPPIUS_SECRET_ACCESS_KEY`, `HIPPIUS_BUCKET`. Cloudflare R2 stays supported as a legacy fallback — if `HIPPIUS_*` is unset the client transparently reads `R2_ACCOUNT_ID`, `R2_ACCESS_KEY_ID`, `R2_SECRET_ACCESS_KEY`, `R2_BUCKET` instead, so existing deployments keep working without an env-var change.
 * **CPU for Phase C eval.** Checkpoint evaluation runs on CPU by default (override with `RADAR_EVAL_DEVICE`). Each eval takes seconds. No GPU required for validators.
 * **DB proxy.** The validator runs a reverse-proxy FastAPI on `RADAR_PROXY_PORT` (default 8080). It forwards `/experiments/*`, `/challenge`, `/frontier`, `/provenance/*` to the subnet owner's centralised DB server, and hosts `/desearch/*` and `/llm/v1/*` locally with per-miner rate limits. The DB itself (Postgres) is run by the subnet owner, not the validator.
 * **DB server deployable modes.** The subnet-owner `database/neuron.py` binary supports three modes via `RADAR_NEURON_MODE`:
@@ -233,7 +233,7 @@ shared/                          Core libraries (validator + miner)
   protocol.py                      Challenge/Proposal wire format (JSON)
   auth.py                          Epistula signing/verification (SR25519)
   challenge.py                     Deterministic challenge + size buckets
-  r2_audit.py                      R2 storage, upload/download/hash verification
+  r2_audit.py                      Hippius/R2 artifact storage, upload/download/hash verification
   task.py                          TaskSpec + Objective, pluggable task definitions
   pg_schema.py                     Postgres DDL + row conversion
   pg_store.py                      Async Postgres experiment store

config.py

@@ -54,13 +54,63 @@ class Config:
     # Comma-separated list of enabled task names. Empty or "all" = all built-in tasks.
     ENABLED_TASKS: str = os.getenv("RADAR_ENABLED_TASKS", "ts_forecasting")
 
-    # ── R2 Audit Log ────────────────────────────────────────────────
+    # ── Artifact Storage (Hippius primary, R2 legacy fallback) ──────
+    # Primary backend is Hippius — the Substrate-based decentralized object
+    # store at https://s3.hippius.com. Cloudflare R2 stays supported as a
+    # legacy fallback during the migration: any unset HIPPIUS_* var falls
+    # back to its R2_* counterpart so existing deployments keep working
+    # without an env-var change.
+    HIPPIUS_ACCESS_KEY_ID: str = os.getenv("HIPPIUS_ACCESS_KEY_ID", "") \
+        or os.getenv("R2_ACCESS_KEY_ID", "")
+    HIPPIUS_SECRET_ACCESS_KEY: str = os.getenv("HIPPIUS_SECRET_ACCESS_KEY", "") \
+        or os.getenv("R2_SECRET_ACCESS_KEY", "")
+    HIPPIUS_BUCKET: str = os.getenv("HIPPIUS_BUCKET", "") \
+        or os.getenv("R2_BUCKET", "")
+    HIPPIUS_ENDPOINT_URL: str = os.getenv("HIPPIUS_ENDPOINT_URL", "")
+    HIPPIUS_REGION: str = os.getenv("HIPPIUS_REGION", "decentralized")
+    PRESIGNED_TTL: int = int(os.getenv("RADAR_PRESIGNED_TTL", "5400"))
+
+    # Legacy R2 names — read directly from env so an operator who has only
+    # set HIPPIUS_* vars sees these as empty (and code that gates on them
+    # behaves correctly). Code that needs "is artifact storage configured?"
+    # should consult HIPPIUS_BUCKET instead, since that's already the
+    # resolved value.
     R2_ACCOUNT_ID: str = os.getenv("R2_ACCOUNT_ID", "")
     R2_ACCESS_KEY_ID: str = os.getenv("R2_ACCESS_KEY_ID", "")
     R2_SECRET_ACCESS_KEY: str = os.getenv("R2_SECRET_ACCESS_KEY", "")
-    R2_BUCKET: str = os.getenv("R2_BUCKET", "")
+    R2_BUCKET: str = os.getenv("R2_BUCKET", "") or os.getenv("HIPPIUS_BUCKET", "")
     R2_PRESIGNED_TTL: int = int(os.getenv("RADAR_PRESIGNED_TTL", "5400"))
 
+    # ── Hippius / Substrate Phase C publishing ──────────────────────
+    # Phase 1 of the Hippius migration ships the signing/serialisation
+    # contract (`shared/substrate.py`) but leaves the publishing pipeline
+    # off by default — existing validators are not affected by merging
+    # this code. Operators opt in by setting HIPPIUS_ENABLED=true once the
+    # downstream phases (IPFS upload, on-chain extrinsic) are wired up.
+    HIPPIUS_ENABLED: bool = os.getenv("HIPPIUS_ENABLED", "false").lower() == "true"
+    HIPPIUS_IPFS_API_URL: str = os.getenv("HIPPIUS_IPFS_API_URL", "")
+    HIPPIUS_KEY: str = os.getenv("HIPPIUS_KEY", "")
+    HIPPIUS_SUBSTRATE_RPC: str = os.getenv("HIPPIUS_SUBSTRATE_RPC", "")
+    SUBSTRATE_APP_TAG: str = os.getenv("SUBSTRATE_APP_TAG", "radar")
+    SUBSTRATE_SCHEMA_VERSION: str = os.getenv(
+        "SUBSTRATE_SCHEMA_VERSION", "radar.substrate.v1",
+    )
+
+    # ── Hippius artifact dual-write (TEN-240 Phase 7) ───────────────
+    # When enabled, validator-side artifact writes (dispatch records,
+    # frontier snapshots, training_meta cache) fan out to both R2 and
+    # Hippius. Defaults off so existing deployments are unaffected by
+    # merging — operators flip in staging first. See docs/HIPPIUS_STORAGE.md.
+    DUAL_WRITE_ARTIFACTS: bool = (
+        os.getenv("RADAR_DUAL_WRITE_ARTIFACTS", "false").lower() == "true"
+    )
+    # When enabled, ArtifactStore.get_bytes falls back to the other backend
+    # when the primary one fails. Off by default during rollout so failures
+    # stay loud and we don't accidentally paper over a misconfiguration.
+    HIPPIUS_ARTIFACT_FALLBACK: bool = (
+        os.getenv("RADAR_HIPPIUS_ARTIFACT_FALLBACK", "false").lower() == "true"
+    )
+
     # ── Round Timing ───────────────────────────────────────────────
     #
     # Timing is controlled at three distinct layers. Read this carefully —

database/migrations/002_substrate_cids.sql

@@ -0,0 +1,37 @@
+-- TEN-240: Phase C substrate audit trail.
+--
+-- Adds a per-experiment list of Hippius CIDs (one per validator that
+-- published a Phase C bundle covering this row). The column is JSONB so
+-- consumers can query into individual entries without a join, and the
+-- GIN index makes "find every experiment a given validator published"
+-- and "find every experiment backed by CID X" cheap.
+--
+-- Each element shape:
+--     {"kind": "phase_c_record",
+--      "validator_hotkey": "5G...",
+--      "cid": "bafy...",
+--      "round_id": 123}
+--
+-- Defaults to '[]' so legacy rows remain valid.
+--
+-- Guarded on ``experiments`` existing because the migration runner is
+-- exercised against an empty schema in CI before init_schema() runs
+-- (see workflows/schema-migrations.yml). On a fresh deploy the base
+-- CREATE TABLE in shared/pg_schema.py already declares the column, so
+-- this migration is a no-op there. On existing deployments init_schema()
+-- runs first and the table exists, so the ALTER applies normally.
+
+DO $$
+BEGIN
+    IF EXISTS (
+        SELECT 1
+        FROM information_schema.tables
+        WHERE table_schema = 'CURRENT_SCHEMA'
+          AND table_name = 'experiments'
+    ) THEN
+        ALTER TABLE CURRENT_SCHEMA.experiments
+            ADD COLUMN IF NOT EXISTS substrate_cids JSONB NOT NULL DEFAULT '[]';
+        CREATE INDEX IF NOT EXISTS idx_substrate_cids
+            ON CURRENT_SCHEMA.experiments USING GIN (substrate_cids);
+    END IF;
+END $$;

database/neuron.py

@@ -30,7 +30,7 @@
     app, include_validator_routes,
     set_db, set_auth, set_challenge, set_frontier,
     set_access_logger, set_hotkey_map, set_rate_limit, set_ip_rate_limit,
-    set_r2, set_pool,
+    set_r2, set_hippius, set_pool,
     get_current_challenge, get_current_frontier,
 )
 from shared.migrations import apply_migrations
@@ -165,11 +165,45 @@ def __init__(self, config):
             except Exception as e:
                 logger.warning("R2 audit log unavailable: %s", e)
 
+        # Hippius client — opt-in via Config.HIPPIUS_ENABLED. Powers the
+        # /experiments/{id}/verify endpoint on the validator surface; the
+        # rest of the DB server doesn't need it. Lazy import + tolerant
+        # construction mirrors the validator-side wiring (TEN-244).
+        self.hippius = self._init_hippius()
+
         # Pool, store, logger set in async init
         self.pool = None
         self.store = None
         self.access_logger = None
 
+    def _init_hippius(self):
+        """Construct the Hippius client when opted in; return None otherwise.
+
+        Mirrors `Validator._init_hippius`: lazy import so the module loads
+        without the TEN-242 wrapper present, tolerant constructor so a flap
+        at startup downgrades to disabled instead of crashing the DB server.
+        """
+        if not Config.HIPPIUS_ENABLED:
+            return None
+        try:
+            from shared.hippius_client import HippiusClient  # type: ignore
+        except ImportError:
+            logger.warning(
+                "HIPPIUS_ENABLED=true but shared.hippius_client is not "
+                "available yet (TEN-242). /experiments/{id}/verify will "
+                "respond 503 until the wrapper ships."
+            )
+            return None
+        try:
+            return HippiusClient(
+                ipfs_api_url=Config.HIPPIUS_IPFS_API_URL,
+                hippius_key=Config.HIPPIUS_KEY,
+                substrate_rpc=Config.HIPPIUS_SUBSTRATE_RPC,
+            )
+        except Exception as e:  # noqa: BLE001
+            logger.warning("Hippius client init failed: %s", e)
+            return None
+
     def _build_task_bucket_resolver(self) -> Callable[[str], list[tuple[int, int]]]:
         """Closure that maps a task name to its FLOPs-equivalent bucket list.
 
@@ -339,6 +373,8 @@ async def _apply_timeout(conn):
         set_pool(self.pool)
         if self.r2:
             set_r2(self.r2)
+        if self.hippius:
+            set_hippius(self.hippius)
         set_access_logger(self.access_logger)
 
         # ── Validator-surface wiring (Epistula auth + rate limit) ──

database/server.py

@@ -35,6 +35,7 @@
 db = None  # PgExperimentStore
 _r2 = None  # R2AuditLog (for agent code storage)
 _pool = None  # asyncpg pool (for agent_submissions table)
+_hippius = None  # HippiusClient — used by GET /experiments/{id}/verify
 
 # Auth middleware reference
 _metagraph = None
@@ -93,6 +94,12 @@ def set_r2(r2):
     _r2 = r2
 
 
+def set_hippius(hippius):
+    """Inject a Hippius client (used by /experiments/{id}/verify)."""
+    global _hippius
+    _hippius = hippius
+
+
 def set_pool(pool):
     global _pool
     _pool = pool
@@ -198,8 +205,21 @@ class SearchRequest(BaseModel):
 
 
 class AddExperimentRequest(BaseModel):
-    """Validator POSTs experiment data after Phase C."""
+    """Validator POSTs experiment data after Phase C.
+
+    ``substrate_cid`` and ``validator_hotkey`` are optional; when set, the
+    server appends a ``{kind: "phase_c_record", validator_hotkey, cid,
+    round_id}`` entry to the row's ``substrate_cids`` audit list.
+
+    ``artifact_cids`` (TEN-240 Phase 7) carries Hippius CIDs for the
+    dual-written non-DB artifacts (checkpoint, architecture,
+    training_meta). Each element is appended to the audit list verbatim;
+    callers are responsible for the ``kind`` field.
+    """
     data: dict
+    substrate_cid: str = ""
+    validator_hotkey: str = ""
+    artifact_cids: list[dict] = []
 
 
 class RecordComponentsRequest(BaseModel):
@@ -508,15 +528,67 @@ async def get_experiment(index: int):
     return elem.to_api_dict()
 
 
+@validator_router.get("/experiments/{index}/verify")
+async def verify_experiment_route(index: int):
+    """Independently verify the substrate audit trail for one experiment.
+
+    For each CID in ``substrate_cids``: download the bundle from Hippius,
+    locate the (round_id, miner_uid) record, sr25519-verify the signature,
+    and compare the record's metric dict with what's in the DB. Returns
+    per-CID flags + discrepancy strings, never raises on per-CID errors.
+
+    503 when the server has no Hippius client wired up.
+    """
+    d = _require_db()
+    elem = await d.get(index)
+    if elem is None:
+        raise HTTPException(status_code=404, detail=f"Experiment {index} not found")
+    if _hippius is None:
+        raise HTTPException(
+            status_code=503,
+            detail="Hippius not configured — substrate verification unavailable",
+        )
+    from database.verify import verify_experiment
+    verifications = await verify_experiment(hippius=_hippius, element=elem)
+    return {
+        "experiment_id": index,
+        "substrate_cids": list(elem.substrate_cids or []),
+        "verifications": verifications,
+    }
+
+
 # ── Write endpoints (new — validators only) ──────────────
 
 @validator_router.post("/experiments/add")
 async def add_experiment(req: AddExperimentRequest):
-    """Validator writes a DataElement after Phase C."""
+    """Validator writes a DataElement after Phase C.
+
+    When ``req.substrate_cid`` is set, append a ``phase_c_record`` entry to
+    the element's ``substrate_cids`` audit list before insert. The entry
+    captures who published the bundle (validator_hotkey), where it lives
+    (cid), and which round it covers (taken from element.round_id).
+    """
     d = _require_db()
     from shared.database import DataElement
     try:
         element = DataElement.from_dict(req.data)
+        # Defensive: from_dict may have produced something unexpected.
+        if not isinstance(element.substrate_cids, list):
+            element.substrate_cids = []
+        cids = list(element.substrate_cids)
+        if req.substrate_cid:
+            cids.append({
+                "kind": "phase_c_record",
+                "validator_hotkey": req.validator_hotkey,
+                "cid": req.substrate_cid,
+                "round_id": int(element.round_id),
+            })
+        # Append artifact CIDs verbatim (TEN-240 Phase 7). Filter to dicts
+        # so a malformed caller can't poison the list with non-objects.
+        for entry in req.artifact_cids or []:
+            if isinstance(entry, dict) and entry.get("cid"):
+                cids.append(entry)
+        element.substrate_cids = cids
         idx = await d.add(element)
     except Exception as e:
         logger.error("add_experiment failed: %s", e, exc_info=True)