Decentralized EIREL validation: validator-side execution, consensus weights, auditability

.env.validator.example

@@ -126,3 +126,34 @@ EIREL_VALIDATOR_MAX_PARALLEL=2
 #   thinking: full reasoning + tool use allowed, 10-minute hard ceiling.
 EIREL_MINER_INSTANT_LATENCY_BUDGET_SECONDS=120
 EIREL_MINER_THINKING_LATENCY_BUDGET_SECONDS=600
+
+# =============================================================================
+# Decentralized execution (EIREL_EXEC_MODE=local) — OPTIONAL
+#
+# Only needed if you run miners on YOUR OWN cluster instead of through the
+# owner proxy. Requires the local stack overlay:
+#   docker compose -f docker-compose.validator.yml -f docker-compose.validator.local.yml up -d
+# Leave EIREL_EXEC_MODE unset (or =owner) to use the owner proxy (default).
+#
+# You GENERATE these secrets yourself (random 32+ bytes); they must match across
+# the engine, provider-proxy and all four tool services, and never enter a
+# miner pod. The validator pays its own eval LLM/tool spend.
+# =============================================================================
+# EIREL_EXEC_MODE=local
+# EIREL_OWNER_HOTKEY=               # pin the subnet owner's ss58 (roster verify)
+# EIREL_TOOL_JOB_SIGNING_KEY=       # per-job HMAC key (engine mints, proxy/tools verify)
+# EIREL_PROVIDER_PROXY_MASTER_TOKEN=
+# EIREL_TOOL_LEDGER_TOKEN=          # tools -> ledger-sink write auth
+# EIREL_LOCAL_LEDGER_TOKEN=         # engine -> ledger-sink read auth (defaults to TOOL_LEDGER_TOKEN)
+# REDIS_PASSWORD=
+# REDIS_URL=redis://:CHANGEME@redis:6379/0
+# SERVER_A_HOST_IP=                 # validator host IP miner k3s pods NAT to
+# EIREL_PROVIDER_CHUTES_API_KEY=    # YOUR LLM spend during eval
+# EIREL_BRAVE_SEARCH_API_KEY=       # (or SERPER/TAVILY) for web_search
+# EIREL_RAG_EMBEDDING_API_KEY=      # for the rag tool
+# EIREL_RUN_BUDGET_USD=30.0         # per-miner-run LLM budget cap the proxy enforces
+# Validator k3s the runtime manager deploys miners into:
+# EIREL_VALIDATOR_KUBECONFIG_PATH=
+# EIREL_VALIDATOR_RUNTIME_NAMESPACE=eirel-miners
+# EIREL_VALIDATOR_RUNTIME_SYSTEM_NAMESPACE=eirel-system
+# EIREL_VALIDATOR_RUNTIME_IMAGE=eirel-miner-runtime:latest

control_plane/owner_api/_helpers.py

@@ -187,10 +187,10 @@ def _strip_sensitive_task_metadata(
 
     The validator-claim path passes ``strip_expected_output=False`` —
     validators run inside the operator's stack and need
-    ``expected_output.answer`` / ``must_not_claim`` to compute the
-    multi-metric per-task score. Miners never see ``expected_output``
-    regardless; that strip happens upstream when the task is dispatched
-    to the miner pod.
+    ``expected_output`` (e.g. ``required_tool``) plus the oracle-derived
+    ``expected_claims`` to compute the multi-metric per-task score.
+    Miners never see ``expected_output`` regardless; that strip happens
+    upstream when the task is dispatched to the miner pod.
     """
     task_metadata = dict(task_dict.get("metadata") or {})
     for key in _SENSITIVE_TASK_METADATA_KEYS:

control_plane/owner_api/app.py

@@ -15,7 +15,9 @@
 import os
 import uvicorn
 from starlette.types import ASGIApp, Message, Receive, Scope, Send
-from fastapi import FastAPI
+from dataclasses import dataclass
+
+from fastapi import FastAPI, HTTPException, Request
 from fastapi.middleware.cors import CORSMiddleware
 
 logging.basicConfig(
@@ -48,6 +50,7 @@
     internal,
     internal_eval,
     checkpoints,
+    roster,
 )
 from shared.common.request_context import RequestIdMiddleware
 from shared.common.security import create_replay_protector
@@ -254,6 +257,40 @@ async def _claim_expiry_sweeper_loop(app: FastAPI) -> None:
         await asyncio.sleep(interval)
 
 
+# ---------------------------------------------------------------------------
+# Ledger-only role (validator-local tool-call attestation sink)
+# ---------------------------------------------------------------------------
+
+# Resolved once at import so router registration (module scope) can gate on it.
+_OWNER_API_ROLE = os.getenv("EIREL_OWNER_API_ROLE", "full").strip().lower()
+_LEDGER_ROLE = _OWNER_API_ROLE == "ledger"
+
+
+@dataclass
+class _LedgerServices:
+    """Minimal ``app.state.services`` stand-in for the ledger role — the only
+    attributes the tool-call ledger + eval-feedback handlers touch are ``db``
+    and ``settings``. Avoids constructing the full ManagedOwnerServices (runtime
+    manager, scheduler, fee verifier) which a validator must not run."""
+
+    db: Database
+    settings: object
+
+
+async def _ledger_read_auth(request: Request) -> str:
+    """Read-auth for the ledger role. The validator reads its OWN local sink, so
+    instead of the metagraph ``validator_dependency`` (which needs chain access)
+    we accept a shared bearer token. When no token is configured (dev), allow."""
+    expected = (os.getenv("EIREL_LOCAL_LEDGER_TOKEN") or os.getenv("EIREL_TOOL_LEDGER_TOKEN") or "").strip()
+    if not expected:
+        return "local-ledger"
+    header = request.headers.get("Authorization", "")
+    token = header[7:].strip() if header.lower().startswith("bearer ") else ""
+    if token != expected:
+        raise HTTPException(status_code=401, detail="invalid local ledger token")
+    return "local-ledger"
+
+
 # ---------------------------------------------------------------------------
 # Lifespan
 # ---------------------------------------------------------------------------
@@ -262,6 +299,21 @@ async def _claim_expiry_sweeper_loop(app: FastAPI) -> None:
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     settings = get_settings()
+    if settings.owner_api_role == "ledger":
+        # Stripped startup: just the DB (small local store — SQLite ok). No
+        # runtime manager, no scheduler loops, no chain. create_all() gives the
+        # current schema directly, so the heavy production migration set is
+        # skipped.
+        db = Database(settings.database_url)
+        db.create_all()
+        app.state.services = _LedgerServices(db=db, settings=settings)
+        logger.info(
+            "owner-api LEDGER role: tool-call attestation sink only (db=%s)",
+            settings.database_url,
+        )
+        yield
+        db.engine.dispose()
+        return
     db = Database(settings.database_url)
     db.create_all()
     run_migrations(db.engine)
@@ -410,18 +462,37 @@ async def lifespan(app: FastAPI):
         "total_applied_count": 0,
         "last_result": None,
     }
-    await app.state.services.refresh_runtime_node_inventory()
-    await app.state.services.reconcile_all_active_deployments()
-    refresh_task = asyncio.create_task(_runtime_capacity_refresh_loop(app))
-    remediation_task = asyncio.create_task(_runtime_remediation_loop(app))
-    sweeper_task = asyncio.create_task(_claim_expiry_sweeper_loop(app))
-    health_task = asyncio.create_task(_container_health_loop(app))
-    reaper_task = asyncio.create_task(_retired_runtime_reaper_loop(app))
-    retry_task = asyncio.create_task(_pending_capacity_retry_loop(app))
+    # In decentralized execution the OWNER no longer manages miner runtimes —
+    # validators run submissions on their own clusters. An owner-api in that
+    # mode serves registry/roster/claim/archive but must NOT run the runtime
+    # reconcile/health/reaper loops (they would demote deployments it isn't
+    # actually hosting). Gated off by default so the owner-hosted path is
+    # byte-for-byte unchanged.
+    _disable_runtime_mgmt = os.getenv(
+        "EIREL_OWNER_DISABLE_RUNTIME_RECONCILE", ""
+    ).strip().lower() in ("1", "true", "yes", "on")
+    loop_tasks: list[asyncio.Task] = []
+    if _disable_runtime_mgmt:
+        logger.warning(
+            "EIREL_OWNER_DISABLE_RUNTIME_RECONCILE set — skipping runtime "
+            "node inventory, reconcile, and the runtime management loops "
+            "(decentralized-execution owner-api: registry/roster/claim only)"
+        )
+    else:
+        await app.state.services.refresh_runtime_node_inventory()
+        await app.state.services.reconcile_all_active_deployments()
+        loop_tasks = [
+            asyncio.create_task(_runtime_capacity_refresh_loop(app)),
+            asyncio.create_task(_runtime_remediation_loop(app)),
+            asyncio.create_task(_claim_expiry_sweeper_loop(app)),
+            asyncio.create_task(_container_health_loop(app)),
+            asyncio.create_task(_retired_runtime_reaper_loop(app)),
+            asyncio.create_task(_pending_capacity_retry_loop(app)),
+        ]
     yield
-    for task in (refresh_task, remediation_task, sweeper_task, health_task, reaper_task, retry_task):
+    for task in loop_tasks:
         task.cancel()
-    for task in (refresh_task, remediation_task, sweeper_task, health_task, reaper_task, retry_task):
+    for task in loop_tasks:
         try:
             await asyncio.wait_for(task, timeout=5.0)
         except (asyncio.CancelledError, asyncio.TimeoutError):
@@ -457,24 +528,37 @@ async def lifespan(app: FastAPI):
     allow_headers=["*"],
 )
 
-app.include_router(health.router)
-app.include_router(submissions.router)
-app.include_router(deployments.router)
-app.include_router(runs.router)
-app.include_router(scoring.router)
-app.include_router(evaluation_tasks.router)
-app.include_router(serving.router)
-app.include_router(workflows.router)
-app.include_router(operator.router)
-app.include_router(runtime.router)
-app.include_router(validators.router)
-app.include_router(internal.router)
-app.include_router(internal_eval.router)
-app.include_router(checkpoints.router)
-app.include_router(dashboard.router)
-
-from control_plane.owner_api.routers import admin as admin_router
-app.include_router(admin_router.router)
+if _LEDGER_ROLE:
+    # Validator-local sink: only the tool-call attestation ledger + a trivial
+    # healthz. Read-auth is the local bearer token, not the metagraph dependency.
+    from control_plane.owner_api.dependencies import validator_dependency
+
+    @app.get("/healthz")
+    async def _ledger_healthz() -> dict[str, str]:
+        return {"status": "ok", "role": "ledger"}
+
+    app.include_router(internal_eval.router)
+    app.dependency_overrides[validator_dependency] = _ledger_read_auth
+else:
+    app.include_router(health.router)
+    app.include_router(submissions.router)
+    app.include_router(deployments.router)
+    app.include_router(runs.router)
+    app.include_router(scoring.router)
+    app.include_router(evaluation_tasks.router)
+    app.include_router(serving.router)
+    app.include_router(workflows.router)
+    app.include_router(operator.router)
+    app.include_router(runtime.router)
+    app.include_router(validators.router)
+    app.include_router(internal.router)
+    app.include_router(internal_eval.router)
+    app.include_router(checkpoints.router)
+    app.include_router(roster.router)
+    app.include_router(dashboard.router)
+
+    from control_plane.owner_api.routers import admin as admin_router
+    app.include_router(admin_router.router)
 
 
 def main() -> None:

control_plane/owner_api/dashboard/cache.py

@@ -1,15 +1,35 @@
 from __future__ import annotations
 
 import time
+from collections import OrderedDict
 from typing import Any, Generic, TypeVar
 
 T = TypeVar("T")
 
+# Dashboard cache keys vary with run_id, miner hotkey, and pagination, so the
+# keyspace grows over the lifetime of the process. ``get()`` only evicts the
+# single key it is asked for, so without an active sweep and a hard cap,
+# expired entries for keys that are never fetched again are never reclaimed —
+# a slow leak that climbs the owner-api pod to its memory limit and OOMKills it
+# after a few days. The periodic sweep reclaims lapsed entries in bulk; the cap
+# bounds worst-case memory regardless of how fast the keyspace churns.
+_DEFAULT_MAX_ENTRIES = 512
+_SWEEP_INTERVAL_SECONDS = 30.0
+
 
 class TTLCache(Generic[T]):
-    def __init__(self, default_ttl_seconds: float = 30.0) -> None:
+    def __init__(
+        self,
+        default_ttl_seconds: float = 30.0,
+        *,
+        max_entries: int = _DEFAULT_MAX_ENTRIES,
+    ) -> None:
         self._default_ttl = default_ttl_seconds
-        self._store: dict[tuple[Any, ...], tuple[float, T]] = {}
+        self._max_entries = max(1, max_entries)
+        # Insertion-ordered so that, once the cap is hit and a sweep has not
+        # freed enough room, we can evict the oldest entry deterministically.
+        self._store: OrderedDict[tuple[Any, ...], tuple[float, T]] = OrderedDict()
+        self._last_sweep = time.monotonic()
 
     def get(self, key: tuple[Any, ...]) -> T | None:
         entry = self._store.get(key)
@@ -22,8 +42,27 @@ def get(self, key: tuple[Any, ...]) -> T | None:
         return value
 
     def set(self, key: tuple[Any, ...], value: T, ttl: float | None = None) -> None:
-        expires_at = time.monotonic() + (ttl if ttl is not None else self._default_ttl)
+        now = time.monotonic()
+        # Periodic full sweep: reclaim every lapsed entry, not just the one a
+        # caller happens to read back. This is what stops a churning keyspace
+        # (new run_ids, new hotkeys, paginated views) from accumulating dead
+        # entries indefinitely.
+        if now - self._last_sweep >= _SWEEP_INTERVAL_SECONDS:
+            self._sweep(now)
+        expires_at = now + (ttl if ttl is not None else self._default_ttl)
         self._store[key] = (expires_at, value)
+        self._store.move_to_end(key)
+        # Hard backstop between sweeps: never retain more than the cap, evicting
+        # oldest-first so the live working set survives.
+        while len(self._store) > self._max_entries:
+            self._store.popitem(last=False)
+
+    def _sweep(self, now: float) -> None:
+        expired = [k for k, (expires_at, _) in self._store.items() if expires_at < now]
+        for k in expired:
+            self._store.pop(k, None)
+        self._last_sweep = now
 
     def clear(self) -> None:
         self._store.clear()
+        self._last_sweep = time.monotonic()

control_plane/owner_api/dashboard/queries.py

@@ -3,7 +3,7 @@
 from collections import defaultdict
 from typing import Any
 
-from sqlalchemy import case, func, select
+from sqlalchemy import case, exists, func, select
 from sqlalchemy.orm import Session
 
 from eirel.groups import LAUNCH_FAMILIES, ensure_family_id
@@ -137,8 +137,8 @@ def _effective_verdict(verdict: str | None, final_task_score: float | None) -> s
 
     Rule: a task is in its pass bucket (matches / partially_matches /
     not_applicable) only if the composite gates (tool_attestation,
-    hallucination_knockout, grounded_gate, safety_gate, cost_attestation,
-    safety_attestation) didn't zero it. When ``final_task_score == 0``
+    grounded_gate, safety_gate, cost_attestation, safety_attestation)
+    didn't zero it. When ``final_task_score == 0``
     and the verdict was a pass type, the row becomes ``gate_knockout``
     so the displayed bucket counts and ``mean_agreement`` reflect the
     same reality as the canonical score.
@@ -348,9 +348,22 @@ def fetch_overview(session: Session, *, services: Any) -> OverviewResponse:
     #   received → pending_capacity → building → deployed_for_eval → retired
     #   build_failed is a terminal error branch.
     _QUEUED = ("received", "pending_capacity", "building")
-    _EVALUATING = ("deployed_for_eval",)
     queued = sum(int(sub_counts.get(s, 0)) for s in _QUEUED)
-    evaluating = sum(int(sub_counts.get(s, 0)) for s in _EVALUATING)
+    # "Evaluating" = deployed-for-eval submissions that aren't yet on the
+    # leaderboard. A submission keeps the sticky ``deployed_for_eval`` status
+    # even after it's been scored (and after the capacity manager parks its
+    # pod in standby), so the raw status count double-counts ranked miners.
+    # Exclude any submission that already has a DeploymentScoreRecord — those
+    # belong on the leaderboard, not the evaluating tally.
+    evaluating = int(session.scalar(
+        select(func.count(ManagedMinerSubmission.id))
+        .where(ManagedMinerSubmission.status == "deployed_for_eval")
+        .where(
+            ~exists().where(
+                DeploymentScoreRecord.submission_id == ManagedMinerSubmission.id
+            )
+        )
+    ) or 0)
     retired = int(sub_counts.get("retired", 0))
     build_failed = int(sub_counts.get("build_failed", 0))
     completed = retired + build_failed

control_plane/owner_api/evaluation/general_chat_scoring.py

@@ -9,9 +9,9 @@
     error              → not counted in the denominator
 
 Gate-aware bucketing: a task whose multi-metric composite was zeroed
-by a gate (tool_attestation / hallucination_knockout / grounded_gate /
-safety_gate / cost_attestation / safety_attestation) is reclassified
-OUT of its pass bucket and into ``gate_knockout``. The ``mean_agreement``
+by a gate (tool_attestation / grounded_gate / safety_gate /
+cost_attestation / safety_attestation) is reclassified OUT of its
+pass bucket and into ``gate_knockout``. The ``mean_agreement``
 treats those rows as contributing 0 to the numerator — so the displayed
 verdict counts and the agreement mean reflect the same reality as the
 canonical multi-metric score.

control_plane/owner_api/routers/internal_eval.py

@@ -66,7 +66,11 @@ def eval_feedback_enabled() -> bool:
 
 
 class ToolCallLogWriteRequest(BaseModel):
-    job_id: str = Field(min_length=1, max_length=64)
+    # 128 (not 64): the local-exec cost tag is
+    # ``task-eval={task_id};deployment={deployment_id}`` — a real task_id plus a
+    # 36-char deployment UUID runs ~80-98 chars, so a 64 cap 422s every
+    # validator-local tool-call write and silently empties tool attestation.
+    job_id: str = Field(min_length=1, max_length=128)
     tool_name: str = Field(min_length=1, max_length=64)
     args_hash: str = Field(default="", max_length=64)
     args_json: dict[str, Any] = Field(default_factory=dict)
@@ -148,7 +152,9 @@ async def write_tool_call(
 )
 async def read_job_ledger(
     request: Request,
-    job_id: str = Query(min_length=1, max_length=64),
+    # 128 to match ToolCallLogWriteRequest.job_id — the local-exec cost tag
+    # (task-eval={task_id};deployment={deployment_id}) overruns a 64 cap.
+    job_id: str = Query(min_length=1, max_length=128),
     validator_hotkey: str = Depends(validator_dependency),
 ) -> JobLedgerResponse:
     del validator_hotkey  # auth-only — registered active validators may read