diff --git a/control_plane/owner_api/dashboard/queries.py b/control_plane/owner_api/dashboard/queries.py index 0cf9d6d..d3acbbc 100644 --- a/control_plane/owner_api/dashboard/queries.py +++ b/control_plane/owner_api/dashboard/queries.py @@ -907,6 +907,8 @@ def _task_evaluation_from_row( *, bundle_task: dict[str, Any], baseline_response_json: dict[str, Any] | None = None, + validator_hotkey: str | None = None, + redact_sensitive: bool = False, ) -> TaskEvaluation: jo = row.judge_output_json or {} @@ -1058,13 +1060,14 @@ def _opt_score(value: Any) -> float | None: category=category, difficulty=difficulty, web_search=web_search, + validator_hotkey=validator_hotkey, task_status=status, evaluated_at=row.created_at.isoformat() if row.created_at else None, prompt=prompt_val if isinstance(prompt_val, str) else None, turn_count=turn_count, user_turns=user_turns, - miner_response=row.miner_response_json, - baseline_response_text=baseline_text, + miner_response=None if redact_sensitive else row.miner_response_json, + baseline_response_text=None if redact_sensitive else baseline_text, agreement_verdict=row.agreement_verdict, agreement_score=( float(row.agreement_score) if row.agreement_score is not None else None @@ -1076,7 +1079,7 @@ def _opt_score(value: Any) -> float | None: if row.miner_latency_seconds else None ), - judge_rationale=jo.get("rationale"), + judge_rationale=None if redact_sensitive else jo.get("rationale"), # Multi-metric breakdown — None for legacy rows. task_type=getattr(row, "task_type", None), pairwise_preference_score=_opt_score(getattr(row, "pairwise_preference_score", None)), @@ -1098,8 +1101,14 @@ def _opt_score(value: Any) -> float | None: composite_knockout_reason=composite_knockout_reason if isinstance(composite_knockout_reason, str) else None, weighted_sum_score=weighted_sum_score_val, eval_outcome=eval_outcome if eval_outcome in ("correct", "partial", "wrong", "hallucinated", "refused", "disputed") else None, - eval_failure_mode=eval_failure_mode if isinstance(eval_failure_mode, str) else None, - eval_guidance=eval_guidance if isinstance(eval_guidance, str) else None, + eval_failure_mode=( + None if redact_sensitive + else (eval_failure_mode if isinstance(eval_failure_mode, str) else None) + ), + eval_guidance=( + None if redact_sensitive + else (eval_guidance if isinstance(eval_guidance, str) else None) + ), ledger_tools=ledger_tools, capability=capability if isinstance(capability, str) else None, domain=domain if isinstance(domain, str) else None, @@ -1161,25 +1170,34 @@ def fetch_run_detail( if tid: tasks_by_id[tid] = _strip_sensitive_task_metadata(task_def) - # Fetch baseline responses alongside the task evals so we can surface - # OpenAI's citations next to the miner's on each task row. + # Fetch baseline responses + claimed_by_validator alongside the task + # evals so we can surface OpenAI's citations and which validator + # graded the row. from shared.common.models import TaskEvaluation as TaskEvaluationRow - baseline_rows = session.execute( - select(TaskEvaluationRow.task_id, TaskEvaluationRow.baseline_response_json) - .where( + eval_rows = session.execute( + select( + TaskEvaluationRow.id, + TaskEvaluationRow.baseline_response_json, + TaskEvaluationRow.claimed_by_validator, + ).where( TaskEvaluationRow.run_id == run_id, TaskEvaluationRow.family_id == family_id, ) ).all() - baseline_by_task: dict[str, dict[str, Any] | None] = { - task_id: baseline for (task_id, baseline) in baseline_rows - } + baseline_by_eval_id: dict[str, dict[str, Any] | None] = {} + validator_by_eval_id: dict[str, str | None] = {} + for eval_id, baseline, claimed_by in eval_rows: + baseline_by_eval_id[eval_id] = baseline + validator_by_eval_id[eval_id] = claimed_by + redact_sensitive = run.status != "completed" tasks = [ _task_evaluation_from_row( row, bundle_task=tasks_by_id.get(row.task_id, {}), - baseline_response_json=baseline_by_task.get(row.task_id), + baseline_response_json=baseline_by_eval_id.get(row.task_evaluation_id), + validator_hotkey=validator_by_eval_id.get(row.task_evaluation_id), + redact_sensitive=redact_sensitive, ) for row in task_rows ] diff --git a/control_plane/owner_api/deployment/deployment_manager.py b/control_plane/owner_api/deployment/deployment_manager.py index d4caed0..606ecd6 100644 --- a/control_plane/owner_api/deployment/deployment_manager.py +++ b/control_plane/owner_api/deployment/deployment_manager.py @@ -13,7 +13,7 @@ logger = logging.getLogger(__name__) -from sqlalchemy import select +from sqlalchemy import func, select from sqlalchemy.orm import Session from shared.common.models import ( @@ -287,6 +287,106 @@ def _remaining_capacity( ) return remaining_cpu, remaining_memory, remaining_pods + def _derive_pool_capacity(self, session: Session) -> int: + """K = how many eval pods fit on healthy nodes right now. + + Sum over verified+schedulable nodes of + ``floor((allocatable_cpu - cpu_headroom) / per_submission_cpu)``, + minus slots already held by ``active`` serving (product) pods — + those aren't part of the eval pool. Recomputed every reconcile + from live node snapshots so adding nodes grows K automatically. + """ + per_sub_cpu = max( + 1, int(self._owner.settings.owner_runtime_submission_cpu_millis) + ) + headroom = self._owner.runtime_cpu_headroom_millis + nodes = list( + session.execute( + select(RuntimeNodeSnapshot) + .where(RuntimeNodeSnapshot.verified.is_(True)) + .where(RuntimeNodeSnapshot.schedulable.is_(True)) + ).scalars() + ) + total_slots = 0 + for node in nodes: + usable = max(0, int(node.allocatable_cpu_millis or 0) - headroom) + total_slots += usable // per_sub_cpu + active_serving = int( + session.execute( + select(func.count()) + .select_from(ManagedDeployment) + .where(ManagedDeployment.status == "active") + ).scalar() + or 0 + ) + return max(0, total_slots - active_serving) + + def _pool_keep_and_done( + self, session: Session, *, family_id: str + ) -> tuple[set[str], set[str], dict[str, Any]]: + """(keep_ids, done_ids, stats) for capacity-aware pool eval. + + ``keep`` = active serving pods ∪ the top-K eligible-not-done + members by submission recency. ``done`` = fully-evaluated + members (their slot can be recycled). Non-kept healthy + deployments are drained → standby by the caller's existing + drain pass, which is exactly the slot-recycle mechanism. + """ + k = self._derive_pool_capacity(session) + pool = self._owner.pool_keep_deployment_ids( + session, family_id=family_id, capacity_k=k + ) + active_ids = { + d_id + for (d_id,) in session.execute( + select(ManagedDeployment.id) + .where(ManagedDeployment.family_id == family_id) + .where(ManagedDeployment.status == "active") + ) + } + if pool["snapshot_total"] == 0: + # Pre-freeze bootstrap: no open snapshot yet (it's created + # lazily on the first validator claim from whatever + # deployments exist). Keep up to K bootstrapping/eval pods so + # the first freeze has live candidates and the first claim + # works — otherwise the scheduler would drain every fresh + # submission before the snapshot ever names it. + boot = list( + session.execute( + select(ManagedDeployment.id) + .where(ManagedDeployment.family_id == family_id) + .where( + ManagedDeployment.status.in_( + ( + "queued", "received", "pending_capacity", + "building", "deploying", "deployed_for_eval", + "eligible", + ) + ) + ) + .order_by(ManagedDeployment.created_at.desc()) + .limit(max(0, k)) + ) + ) + keep = {d_id for (d_id,) in boot} | active_ids + stats = { + "k": k, + "eligible_total": 0, + "snapshot_total": 0, + "done": 0, + "bootstrap": True, + } + return keep, set(), stats + keep = set(pool["keep"]) | active_ids + stats = { + "k": k, + "eligible_total": pool["eligible_total"], + "snapshot_total": pool["snapshot_total"], + "done": len(pool["done"]), + "bootstrap": False, + } + return keep, set(pool["done"]), stats + def _select_runtime_node( self, session: Session, @@ -826,24 +926,24 @@ async def reconcile_family_deployments(self, *, family_id: str) -> None: .where(ManagedDeployment.status != "retired") ).scalars() ) - pinned_ids = self._owner.open_run_pinned_deployment_ids(session, family_id=family_id) - bootstrapping_ids = { - item.id - for item in deployments - if item.status in {"building", "received", "deploying", "pending_capacity"} - or item.health_status == "starting" - } - to_keep = ( - {item.id for item in deployments if item.is_active} - | pinned_ids - | bootstrapping_ids - | { - item.id - for item in deployments - if item.status in {"deployed_for_eval", "eligible"} - } + # Capacity-aware continuous pool: deploy only the top-K + # eligible-not-done members (by submission recency) plus + # active serving pods. Members beyond K wait for a freed + # slot; fully-evaluated ("done") members are not started so + # the drain pass recycles their slot. This replaces the old + # "pin every snapshot member" behavior that stranded every + # submission past node capacity in pending_capacity forever. + keep_ids, done_ids, pool_stats = self._pool_keep_and_done( + session, family_id=family_id ) - to_start = [item.id for item in deployments if item.id in to_keep] + logger.info( + "pool reconcile family=%s k=%d snapshot=%d eligible=%d " + "done=%d keep=%d", + family_id, pool_stats["k"], pool_stats["snapshot_total"], + pool_stats["eligible_total"], pool_stats["done"], + len(keep_ids), + ) + to_start = [item.id for item in deployments if item.id in keep_ids] for deployment_id in to_start: try: await self.ensure_deployment_runtime(deployment_id=deployment_id) @@ -859,23 +959,14 @@ async def reconcile_family_deployments(self, *, family_id: str) -> None: .where(ManagedDeployment.status != "retired") ).scalars() ) - pinned_ids = self._owner.open_run_pinned_deployment_ids(session, family_id=family_id) - bootstrapping_ids = { - item.id - for item in deployments - if item.status in {"building", "received", "deploying", "pending_capacity"} - or item.health_status == "starting" - } - desired_keep = ( - {item.id for item in deployments if item.is_active} - | pinned_ids - | bootstrapping_ids - | { - item.id - for item in deployments - if item.status == "deployed_for_eval" - } + # Same capacity-bounded keep set as the deploy pass. Anything + # not kept (over-capacity members AND fully-evaluated "done" + # members) drains → standby_cold below, freeing its slot for + # the next eligible miner on the following reconcile tick. + keep_ids, _done_ids, _pool_stats = self._pool_keep_and_done( + session, family_id=family_id ) + desired_keep = set(keep_ids) now = utcnow() for deployment in deployments: if deployment.id not in desired_keep: diff --git a/control_plane/owner_api/evaluation/evaluation_task_manager.py b/control_plane/owner_api/evaluation/evaluation_task_manager.py index 7bc1af0..6e1b933 100644 --- a/control_plane/owner_api/evaluation/evaluation_task_manager.py +++ b/control_plane/owner_api/evaluation/evaluation_task_manager.py @@ -28,6 +28,12 @@ from control_plane.owner_api._helpers import _strip_sensitive_task_metadata +# Oracle baseline cache lifetime. Within this window every pool cycle +# that re-claims a task reuses the frozen baseline (flat oracle cost + +# batch-independent scoring); past it the next claim recomputes and +# resets the first-compute stamp. Confirmed product decision: 12h. +_ORACLE_BASELINE_TTL = timedelta(hours=12) + def _opt_float(value: Any) -> float | None: """Coerce numeric scores to float; preserve ``None`` for N/A dimensions.""" @@ -54,6 +60,11 @@ def _upsert_eval_feedback( unique constraint. Caller has already verified ``eval_outcome`` is present in ``judge_meta``. """ + from control_plane.owner_api.routers.internal_eval import ( + eval_feedback_enabled, + ) + if not eval_feedback_enabled(): + return knockout_reasons = judge_meta.get("eval_knockout_reasons") if not isinstance(knockout_reasons, list): knockout_reasons = [] @@ -345,8 +356,23 @@ def submit_task_result( "remaining_task_count": -1, } - task.baseline_response_json = baseline_response - task.oracle_cost_usd = max(0.0, float(oracle_cost_usd or 0.0)) + # Oracle baseline 12h TTL. Keep the first-computed baseline (and + # its cost + first-compute stamp) for the whole TTL window so + # every pool cycle that re-claims this task is judged against an + # identical baseline and the oracle is billed once. Only accept + # (and re-stamp) a freshly computed baseline when the cache is + # absent or expired — that's also the only case the validator + # actually computes one (see build_claim_items / engine.py). + cached_at = task.baseline_cached_at + cache_fresh = ( + cached_at is not None + and task.baseline_response_json is not None + and (now - cached_at) < _ORACLE_BASELINE_TTL + ) + if not cache_fresh and baseline_response is not None: + task.baseline_response_json = baseline_response + task.oracle_cost_usd = max(0.0, float(oracle_cost_usd or 0.0)) + task.baseline_cached_at = now task.status = "evaluated" task.evaluated_at = now task.updated_at = now @@ -427,13 +453,55 @@ def submit_task_result( session.flush() remaining = self._remaining_tasks(session, task) - family_complete = remaining == 0 - if family_complete: - self._on_family_evaluation_complete( - session, - run_id=task.run_id, - family_id=task.family_id, - ) + family_complete = False + if remaining == 0: + # All tasks evaluated against the *current* pool batch. Under + # continuous-pool eval that's not run completion — other + # eligible miners still owe every task. Re-arm the tasks so + # the next batch can claim them; the claim-time filter sends + # each re-armed task only to not-yet-scored miners, and the + # 12h baseline cache means the oracle isn't re-billed. The + # run is finished only once no eligible-not-done member + # remains (pool fully drained). + # + # If pool eligibility can't be computed (degraded owner / + # transient error), fall back to the legacy "complete when + # all tasks evaluated" semantics so submission never wedges. + try: + pool = self._owner.pool_keep_deployment_ids( + session, family_id=task.family_id, capacity_k=1_000_000, + ) + eligible_total = int(pool["eligible_total"]) + except Exception as exc: # noqa: BLE001 — never wedge submit + logger.warning( + "pool eligibility check failed (%s); finalizing on " + "all-tasks-evaluated (legacy behavior)", exc, + ) + eligible_total = 0 + if eligible_total > 0: + session.execute( + update(TaskEvaluation) + .where(TaskEvaluation.run_id == task.run_id) + .where(TaskEvaluation.family_id == task.family_id) + .where(TaskEvaluation.status == "evaluated") + .values( + status="pending", + claimed_by_validator=None, + claimed_at=None, + claim_expires_at=None, + claim_attempt_count=0, + evaluated_at=None, + updated_at=now, + ) + ) + session.flush() + else: + family_complete = True + self._on_family_evaluation_complete( + session, + run_id=task.run_id, + family_id=task.family_id, + ) return { "status": "accepted", @@ -834,7 +902,22 @@ def build_claim_items( run_id: str, family_id: str, ) -> list[dict[str, Any]]: - """Package claimed tasks with task payload + miner fan-out list.""" + """Package claimed tasks with task payload + miner fan-out list. + + Per-task miner list excludes any hotkey that already has a + TaskMinerResult row for that ``(run_id, task_id)`` — so a miner + drops out of fanout the instant it's fully scored, which is the + signal the pool scheduler uses to recycle its slot. Under + capacity-aware continuous-pool evaluation the validator thus + only ever judges the live K-pod batch, and re-judging a reset + task never overwrites already-scored miners. + + When a task's oracle baseline is cached and within the 12h TTL, + it's attached as ``cached_baseline`` so the validator skips + oracle enrichment entirely (flat oracle cost across pool cycles; + identical baseline → batch-independent scoring). + """ + now = utcnow() snapshot = session.execute( select(EpochTargetSnapshot).where( EpochTargetSnapshot.run_id == run_id, @@ -844,18 +927,58 @@ def build_claim_items( if snapshot is None: return [] - miners = [] + # Capacity-aware pool: only fan out to miners whose deployment is + # live right now. The scheduler keeps just K pods deployed and + # cycles the rest through; a member without a healthy pod is + # simply not in this batch yet (it'll be picked up once its pod + # is up). Combined with the already-scored exclusion below, the + # validator only ever judges the current K-pod batch. + from shared.common.models import ManagedDeployment + + _SERVING_STATES = {"deployed_for_eval", "eligible", "active"} + member_dep_ids = [ + str((m.get("metadata") or {}).get("deployment_id") or "").strip() + for m in (snapshot.members_json or []) + ] + member_dep_ids = [d for d in member_dep_ids if d] + healthy_dep_ids: set[str] = set() + if member_dep_ids: + for dep_id, dep_status, dep_health in session.execute( + select( + ManagedDeployment.id, + ManagedDeployment.status, + ManagedDeployment.health_status, + ).where(ManagedDeployment.id.in_(member_dep_ids)) + ): + if dep_status in _SERVING_STATES and dep_health == "healthy": + healthy_dep_ids.add(dep_id) + + all_miners: list[dict[str, Any]] = [] for member in (snapshot.members_json or []): hk = member.get("hotkey", "") if not hk: continue metadata = member.get("metadata", {}) or {} - miners.append({ + dep_id = str(metadata.get("deployment_id") or "").strip() + if dep_id not in healthy_dep_ids: + continue + all_miners.append({ "hotkey": hk, "endpoint": metadata.get("validator_endpoint") or member.get("endpoint", ""), "auth_headers": metadata.get("auth_headers", {}), }) + claimed_task_ids = [t.task_id for t in claimed_tasks] + scored_by_task: dict[str, set[str]] = {tid: set() for tid in claimed_task_ids} + if claimed_task_ids: + for tid, hk in session.execute( + select(TaskMinerResult.task_id, TaskMinerResult.miner_hotkey) + .where(TaskMinerResult.run_id == run_id) + .where(TaskMinerResult.family_id == family_id) + .where(TaskMinerResult.task_id.in_(claimed_task_ids)) + ): + scored_by_task.setdefault(tid, set()).add(hk) + bundle = self._owner.run_evaluation_bundle( session, run_id=run_id, family_id=family_id, ) @@ -870,23 +993,28 @@ def build_claim_items( for task in claimed_tasks: task_payload = tasks_by_id.get(task.task_id, {}) if isinstance(task_payload, dict): - # Validators run inside the operator's stack and need - # ``expected_output.answer`` to compute multi-metric - # per-task scores. Sensitive metadata keys (seed_id, - # hidden_fixture, etc.) are still scrubbed. task_payload = _strip_sensitive_task_metadata( task_payload, strip_expected_output=False, ) - items.append({ + already_scored = scored_by_task.get(task.task_id, set()) + miners_for_task = [m for m in all_miners if m["hotkey"] not in already_scored] + item = { "task_evaluation_id": task.id, "run_id": task.run_id, "family_id": task.family_id, "task_id": task.task_id, "task_index": task.task_index, "task_payload": task_payload, - "miners": miners, + "miners": miners_for_task, "claim_expires_at": task.claim_expires_at.isoformat() if task.claim_expires_at else "", "rubric_version": snapshot.rubric_version, "benchmark_version": snapshot.benchmark_version, - }) + } + if ( + task.baseline_response_json is not None + and task.baseline_cached_at is not None + and (now - task.baseline_cached_at) < _ORACLE_BASELINE_TTL + ): + item["cached_baseline"] = task.baseline_response_json + items.append(item) return items diff --git a/control_plane/owner_api/evaluation/run_manager.py b/control_plane/owner_api/evaluation/run_manager.py index 8953a3c..7aa4ce0 100644 --- a/control_plane/owner_api/evaluation/run_manager.py +++ b/control_plane/owner_api/evaluation/run_manager.py @@ -16,7 +16,7 @@ logger = logging.getLogger(__name__) -from sqlalchemy import delete, select, update +from sqlalchemy import delete, func, select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session @@ -29,6 +29,7 @@ ManagedDeployment, ManagedMinerSubmission, TaskEvaluation, + TaskMinerResult, RunFamilyResult, ) from shared.contracts.models import BenchmarkRunRecord @@ -886,6 +887,113 @@ def open_run_pinned_deployment_ids( deployment_ids.add(deployment_id) return deployment_ids + def pool_keep_deployment_ids( + self, + session: Session, + *, + family_id: str, + capacity_k: int, + ) -> dict[str, Any]: + """Capacity-aware continuous-pool selection for the open run. + + Replaces "pin every snapshot member" (which silently strands + every submission past node capacity in ``pending_capacity``). + Of the open snapshot's members: + + * ``done`` — has a TaskMinerResult row for *every* task in the + run → fully evaluated; its pod can be retired and the slot + recycled. + * ``eligible`` — not yet done; ordered by submission recency + (most recent first, per product decision). The first + ``capacity_k`` of these are ``keep`` (stay deployed); the + rest wait for a freed slot. + + Returns ``{keep, done, eligible_total, snapshot_total}`` as + sets/ints. ``eligible_total`` drives run-completion gating — + the run isn't ``scored`` until it reaches 0. + """ + snapshot = session.execute( + select(EpochTargetSnapshot) + .where(EpochTargetSnapshot.status == "open") + .where(EpochTargetSnapshot.family_id == ensure_family_id(family_id)) + .limit(1) + ).scalar_one_or_none() + if snapshot is None: + return { + "keep": set(), + "done": set(), + "eligible_total": 0, + "snapshot_total": 0, + } + + run_id = snapshot.run_id + total_tasks = int( + session.execute( + select(func.count()) + .select_from(TaskEvaluation) + .where(TaskEvaluation.run_id == run_id) + .where(TaskEvaluation.family_id == ensure_family_id(family_id)) + ).scalar() + or 0 + ) + + # Per-miner distinct evaluated task count for this run. + done_counts: dict[str, int] = {} + if total_tasks > 0: + for hk, n in session.execute( + select( + TaskMinerResult.miner_hotkey, + func.count(func.distinct(TaskMinerResult.task_id)), + ) + .where(TaskMinerResult.run_id == run_id) + .where(TaskMinerResult.family_id == ensure_family_id(family_id)) + .group_by(TaskMinerResult.miner_hotkey) + ): + done_counts[hk] = int(n or 0) + + members = list(snapshot.members_json or []) + sub_ids = [ + str((m.get("metadata") or {}).get("submission_id") or "").strip() + for m in members + ] + sub_ids = [s for s in sub_ids if s] + sub_created: dict[str, Any] = {} + if sub_ids: + for sid, created in session.execute( + select(ManagedMinerSubmission.id, ManagedMinerSubmission.created_at) + .where(ManagedMinerSubmission.id.in_(sub_ids)) + ): + sub_created[sid] = created + + done: set[str] = set() + eligible: list[tuple[Any, str]] = [] # (created_at, deployment_id) + for member in members: + meta = member.get("metadata") or {} + dep_id = str(meta.get("deployment_id") or "").strip() + sub_id = str(meta.get("submission_id") or "").strip() + hk = str(member.get("hotkey") or "").strip() + if not dep_id: + continue + if total_tasks > 0 and done_counts.get(hk, 0) >= total_tasks: + done.add(dep_id) + continue + eligible.append((sub_created.get(sub_id), dep_id)) + + # Most-recent submission first. None-created sorts last (oldest). + from datetime import datetime as _dt + + eligible.sort( + key=lambda t: t[0] or _dt.min, + reverse=True, + ) + keep = {dep for _, dep in eligible[: max(0, capacity_k)]} + return { + "keep": keep, + "done": done, + "eligible_total": len(eligible), + "snapshot_total": len(members), + } + def resolve_run_member( self, session: Session, diff --git a/control_plane/owner_api/routers/internal_eval.py b/control_plane/owner_api/routers/internal_eval.py index 6d28f43..313cad1 100644 --- a/control_plane/owner_api/routers/internal_eval.py +++ b/control_plane/owner_api/routers/internal_eval.py @@ -35,6 +35,7 @@ from __future__ import annotations import logging +import os from datetime import datetime from typing import Any @@ -55,6 +56,15 @@ router = APIRouter(tags=["internal_eval"]) +_TRUTHY = {"1", "true", "yes", "on"} + + +def eval_feedback_enabled() -> bool: + """Return True iff the EvalFeedback subsystem is enabled. + """ + return os.getenv("EIREL_EVAL_FEEDBACK_ENABLED", "1").strip().lower() in _TRUTHY + + class ToolCallLogWriteRequest(BaseModel): job_id: str = Field(min_length=1, max_length=64) tool_name: str = Field(min_length=1, max_length=64) @@ -226,6 +236,11 @@ async def read_eval_feedback( callers cannot read another miner's feedback by passing a different hotkey in a query param. No proxy / no internal token. """ + if not eval_feedback_enabled(): + raise HTTPException( + status_code=503, + detail="eval feedback is temporarily disabled", + ) services: ManagedOwnerServices = request.app.state.services with services.db.sessionmaker() as session: rows = list( diff --git a/control_plane/owner_api/routers/scoring.py b/control_plane/owner_api/routers/scoring.py index 1f7f63a..cfee592 100644 --- a/control_plane/owner_api/routers/scoring.py +++ b/control_plane/owner_api/routers/scoring.py @@ -135,7 +135,6 @@ async def aggregate_family_scores( async def get_weights( request: Request, run_id: str | None = None, - validator_hotkey: str = Depends(validator_dependency), ) -> WeightsResponse: """Return the current weight table for validators to set on-chain. @@ -147,7 +146,6 @@ async def get_weights( Validators poll this every ~180 blocks, resolve hotkey→UID via their own metagraph sync, and call ``subtensor.set_weights()``. """ - del validator_hotkey services: ManagedOwnerServices = request.app.state.services with services.db.sessionmaker() as session: from shared.common.models import EvaluationRun, RunFamilyResult diff --git a/deploy/k8s/owner-api-hybrid/configmap.env.example b/deploy/k8s/owner-api-hybrid/configmap.env.example index b210bbf..95762df 100644 --- a/deploy/k8s/owner-api-hybrid/configmap.env.example +++ b/deploy/k8s/owner-api-hybrid/configmap.env.example @@ -95,3 +95,7 @@ OWNER_API_PUBLIC_URL= # CORS allowlist (comma-separated). Empty / "*" = allow any origin (dev only). EIREL_CORS_ALLOW_ORIGINS= + +EIREL_DB_POOL_SIZE=20 + +EIREL_EVAL_FEEDBACK_ENABLED=1 diff --git a/deploy/k8s/owner-api-hybrid/deployment.yaml b/deploy/k8s/owner-api-hybrid/deployment.yaml index e34cdd7..254bfe3 100644 --- a/deploy/k8s/owner-api-hybrid/deployment.yaml +++ b/deploy/k8s/owner-api-hybrid/deployment.yaml @@ -61,12 +61,16 @@ spec: port: 8000 initialDelaySeconds: 5 periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 3 livenessProbe: httpGet: path: /healthz port: 8000 initialDelaySeconds: 15 periodSeconds: 20 + timeoutSeconds: 5 + failureThreshold: 5 resources: requests: cpu: 250m diff --git a/infra/ansible/playbooks/create-runtime-secrets.yml b/infra/ansible/playbooks/create-runtime-secrets.yml index 8207d8c..94d13de 100644 --- a/infra/ansible/playbooks/create-runtime-secrets.yml +++ b/infra/ansible/playbooks/create-runtime-secrets.yml @@ -11,6 +11,8 @@ --from-literal=PROVIDER_PROXY_TOKEN={{ provider_proxy_token }} --from-literal=EIREL_WEB_SEARCH_TOKEN={{ web_search_tool_token }} --from-literal=EIREL_SANDBOX_TOKEN={{ sandbox_tool_token }} + --from-literal=EIREL_URL_FETCH_TOKEN={{ url_fetch_tool_token }} + --from-literal=EIREL_RAG_TOOL_API_TOKEN={{ rag_tool_token }} --dry-run=client -o yaml | kubectl apply -f - environment: KUBECONFIG: /etc/rancher/k3s/k3s.yaml diff --git a/infra/ansible/playbooks/rotate-runtime-secrets.yml b/infra/ansible/playbooks/rotate-runtime-secrets.yml index 5bdf7e5..bbec74d 100644 --- a/infra/ansible/playbooks/rotate-runtime-secrets.yml +++ b/infra/ansible/playbooks/rotate-runtime-secrets.yml @@ -13,6 +13,8 @@ --from-literal=PROVIDER_PROXY_TOKEN={{ provider_proxy_token }} --from-literal=EIREL_WEB_SEARCH_TOKEN={{ web_search_tool_token }} --from-literal=EIREL_SANDBOX_TOKEN={{ sandbox_tool_token }} + --from-literal=EIREL_URL_FETCH_TOKEN={{ url_fetch_tool_token }} + --from-literal=EIREL_RAG_TOOL_API_TOKEN={{ rag_tool_token }} --dry-run=client -o yaml | kubectl apply -f - environment: KUBECONFIG: /etc/rancher/k3s/k3s.yaml diff --git a/infra/miner_runtime/_k8s_helpers.py b/infra/miner_runtime/_k8s_helpers.py index aa606cf..411e81b 100644 --- a/infra/miner_runtime/_k8s_helpers.py +++ b/infra/miner_runtime/_k8s_helpers.py @@ -81,6 +81,22 @@ class DeploymentStatus: _CONFIGMAP_MAX_BYTES = 900 * 1024 +# Single ConfigMap key holding the whole cleaned submission as a gzipped +# tar. We deliberately do NOT map one file → one ConfigMap key: k8s +# ConfigMap keys must match ``[-._a-zA-Z0-9]+`` and are length-capped, so +# any submission with subdirectories (``pkg/mod.py``), unusual characters, +# or deep paths makes the ConfigMap API reject the whole object with 422. +# One opaque archive key sidesteps every key-charset / key-length / nested +# -path failure mode permanently; an init container unpacks it into a +# shared emptyDir before the runtime container starts. +_CODE_ARCHIVE_KEY = "code.tar.gz" +# Where the ConfigMap (archive) is mounted, and where the init container +# unpacks it for the runtime container to import from. +_CODE_ARCHIVE_MOUNT = "/submission-archive" +_CODE_EXTRACT_MOUNT = "/submission" +_CODE_EXTRACT_VOLUME = "submission-code-extracted" +_CODE_ARCHIVE_VOLUME = "submission-code" + _ARCHIVE_EXCLUDE_DIRS = {"__pycache__", ".git", ".venv", "venv", ".pytest_cache", ".mypy_cache", ".ruff_cache", "node_modules"} _ARCHIVE_EXCLUDE_SUFFIXES = {".pyc", ".pyo", ".pyd", ".swp", ".swo"} _ARCHIVE_EXCLUDE_NAMES = {".DS_Store", "Thumbs.db"} @@ -121,14 +137,47 @@ def _extract_archive_to_dict(archive_bytes: bytes) -> dict[str, bytes]: return files -def _check_configmap_size(files: dict[str, bytes]) -> None: - total = sum(len(v) for v in files.values()) - if total > _CONFIGMAP_MAX_BYTES: +def _repack_clean_archive(files: dict[str, bytes]) -> bytes: + """Re-tar the cleaned file set into one deterministic gzipped archive. + + ``files`` has already been path-validated and junk-filtered by + :func:`_extract_archive_to_dict`. We repack (rather than forwarding the + miner's original tar) so the archive shipped to the pod contains + exactly the sanitized tree — no ``__pycache__``/``.git`` bloat, no + path-escape members, no preserved ownership/mtime. Deterministic + output (sorted names, zeroed mtime/uid/gid) keeps the ConfigMap stable + across re-deploys so unchanged code doesn't churn the object. + """ + raw = io.BytesIO() + # mtime=0 → reproducible gzip header; sorted names → stable tar order. + with tarfile.open(fileobj=raw, mode="w:gz", compresslevel=9) as tar: + for path in sorted(files): + data = files[path] + info = tarfile.TarInfo(name=path) + info.size = len(data) + info.mtime = 0 + info.uid = info.gid = 0 + info.uname = info.gname = "" + info.mode = 0o644 + tar.addfile(info, io.BytesIO(data)) + return raw.getvalue() + + +def _check_archive_configmap_size(archive_bytes: bytes) -> None: + """Reject archives that won't fit in a ConfigMap. + + etcd stores the base64 of ``binaryData`` values, so the on-wire size + is ~4/3 the raw archive. Bound the base64 size, not the raw size, so + the check matches what the API server actually enforces (~1 MiB). + """ + encoded = (len(archive_bytes) + 2) // 3 * 4 + if encoded > _CONFIGMAP_MAX_BYTES: from .runtime_manager import RuntimeManagerError raise RuntimeManagerError( - f"archive too large for ConfigMap: {total} bytes " - f"(limit {_CONFIGMAP_MAX_BYTES} bytes)" + f"submission archive too large for ConfigMap: {encoded} bytes " + f"base64 (limit {_CONFIGMAP_MAX_BYTES} bytes); raw archive " + f"{len(archive_bytes)} bytes" ) @@ -149,8 +198,10 @@ def _build_code_configmap( *, name: str, submission_id: str, - files: dict[str, bytes], + archive_bytes: bytes, ) -> dict[str, Any]: + """One ConfigMap, one key (``code.tar.gz``) holding the whole archive. + """ return { "apiVersion": "v1", "kind": "ConfigMap", @@ -159,8 +210,7 @@ def _build_code_configmap( "labels": {"eirel.dev/submission-id": submission_id}, }, "binaryData": { - path: base64.b64encode(content).decode() - for path, content in files.items() + _CODE_ARCHIVE_KEY: base64.b64encode(archive_bytes).decode(), }, } diff --git a/infra/miner_runtime/runtime_manager.py b/infra/miner_runtime/runtime_manager.py index f3f9fac..ee373ca 100644 --- a/infra/miner_runtime/runtime_manager.py +++ b/infra/miner_runtime/runtime_manager.py @@ -11,13 +11,19 @@ from ._k8s_helpers import ( DeploymentStatus, DeploymentStatusCode, + _CODE_ARCHIVE_KEY, + _CODE_ARCHIVE_MOUNT, + _CODE_ARCHIVE_VOLUME, + _CODE_EXTRACT_MOUNT, + _CODE_EXTRACT_VOLUME, _build_code_configmap, _build_k8s_deployment, _build_k8s_service, _build_network_policy, - _check_configmap_size, + _check_archive_configmap_size, _create_or_replace, _extract_archive_to_dict, + _repack_clean_archive, ) logger = logging.getLogger(__name__) @@ -330,20 +336,60 @@ def _deployment_manifest_common( }, ] ) - if code_configmap_name is not None: - container["volumeMounts"] = [ - {"name": "submission-code", "mountPath": "/submission"}, - ] pod_spec: dict[str, Any] = { "containers": [container], "restartPolicy": "Always", } if code_configmap_name is not None: + # The runtime container imports code from an emptyDir + # (``/submission``) that an init container fills by unpacking the + # single-key archive ConfigMap. The runtime container never sees + # the raw ConfigMap, so the legacy one-file-per-key layout (and + # its 422 key-charset failures) is gone for good. + container["volumeMounts"] = [ + {"name": _CODE_EXTRACT_VOLUME, "mountPath": _CODE_EXTRACT_MOUNT}, + ] + # Python 3.12 ``tarfile`` ``data`` filter: blocks path traversal, + # absolute paths, links escaping the tree, and device/special + # files — safe extraction of an untrusted miner archive. owner-api + # also validates the archive at build time (defense in depth). + extract_script = ( + "import tarfile, pathlib; " + f"dest = pathlib.Path({_CODE_EXTRACT_MOUNT!r}); " + "dest.mkdir(parents=True, exist_ok=True); " + "tar = tarfile.open(" + f"{_CODE_ARCHIVE_MOUNT!r} + '/' + {_CODE_ARCHIVE_KEY!r}); " + "tar.extractall(dest, filter='data'); " + "tar.close()" + ) + pod_spec["initContainers"] = [ + { + "name": "extract-submission-code", + "image": artifact_url, + "imagePullPolicy": "Always", + "command": ["python", "-c", extract_script], + "volumeMounts": [ + { + "name": _CODE_ARCHIVE_VOLUME, + "mountPath": _CODE_ARCHIVE_MOUNT, + "readOnly": True, + }, + { + "name": _CODE_EXTRACT_VOLUME, + "mountPath": _CODE_EXTRACT_MOUNT, + }, + ], + } + ] pod_spec["volumes"] = [ { - "name": "submission-code", + "name": _CODE_ARCHIVE_VOLUME, "configMap": {"name": code_configmap_name}, }, + { + "name": _CODE_EXTRACT_VOLUME, + "emptyDir": {}, + }, ] pod_spec["nodeSelector"] = { "eirel.dev/runtime-pool": "true", @@ -1035,13 +1081,18 @@ async def ensure_runtime(self, **kwargs) -> MinerRuntimeHandle: or "/healthz" ) + # Validate + junk-filter (raises on path escape), then repack the + # cleaned tree into one deterministic archive. The pod's init + # container unpacks this single key, so nested paths and odd + # filenames can never break ConfigMap creation again. code_files = _extract_archive_to_dict(archive_bytes) - _check_configmap_size(code_files) + clean_archive = _repack_clean_archive(code_files) + _check_archive_configmap_size(clean_archive) cm_body = _build_code_configmap( name=f"{dep_name}-code", submission_id=submission_id, - files=code_files, + archive_bytes=clean_archive, ) try: await _create_or_replace( diff --git a/shared/common/database.py b/shared/common/database.py index 1c82b7c..d1a12f6 100644 --- a/shared/common/database.py +++ b/shared/common/database.py @@ -1,5 +1,6 @@ from __future__ import annotations +import os from collections.abc import Iterator from typing import Any @@ -10,14 +11,24 @@ from shared.common.models import Base +def _int_env(name: str, default: int) -> int: + raw = os.getenv(name, "").strip() + if not raw: + return default + try: + return max(1, int(raw)) + except ValueError: + return default + + class Database: def __init__(self, url: str): normalized_url = url.replace("sqlite+aiosqlite://", "sqlite://") engine_kwargs: dict[str, Any] = {"future": True} if not normalized_url.startswith("sqlite"): engine_kwargs.update({ - "pool_size": 5, - "max_overflow": 10, + "pool_size": _int_env("EIREL_DB_POOL_SIZE", 20), + "max_overflow": _int_env("EIREL_DB_MAX_OVERFLOW", 30), "pool_pre_ping": True, "pool_recycle": 1800, "connect_args": {"connect_timeout": 10}, diff --git a/shared/common/migrations.py b/shared/common/migrations.py index 7b43e79..3f3b7a5 100644 --- a/shared/common/migrations.py +++ b/shared/common/migrations.py @@ -68,7 +68,15 @@ def _run_migrations_unlocked(engine: Engine) -> list[str]: VALUES (:version, :description) """ ), - {"version": migration.version, "description": migration.description}, + { + "version": migration.version, + # schema_migrations.description is VARCHAR(255); a + # longer string would raise StringDataRightTruncation + # AFTER apply() already ran, wedging startup in a + # crash loop. Truncate defensively — this column is + # human-readable bookkeeping, not load-bearing. + "description": (migration.description or "")[:255], + }, ) executed.append(migration.version) return executed @@ -234,6 +242,32 @@ def _migration_validator_cost_tracking(engine: Engine) -> None: ) +def _migration_oracle_baseline_ttl_cache(engine: Engine) -> None: + """Add ``task_evaluations.baseline_cached_at`` for the 12h oracle + baseline TTL cache used by capacity-aware pool evaluation. + + Skipped on fresh databases — ``Base.metadata.create_all`` runs after + migrations and creates the column from the model. Only ALTERs + pre-existing DBs. + """ + from sqlalchemy import inspect, text + inspector = inspect(engine) + if "task_evaluations" not in inspector.get_table_names(): + return + existing_columns = { + col["name"] for col in inspector.get_columns("task_evaluations") + } + if "baseline_cached_at" in existing_columns: + return + with engine.begin() as conn: + conn.execute( + text( + "ALTER TABLE task_evaluations " + "ADD COLUMN baseline_cached_at TIMESTAMP WITHOUT TIME ZONE" + ) + ) + + MIGRATIONS: tuple[Migration, ...] = ( Migration( version="initial_schema", @@ -273,4 +307,12 @@ def _migration_validator_cost_tracking(engine: Engine) -> None: ), apply=_migration_validator_cost_tracking, ), + Migration( + version="oracle_baseline_ttl_cache", + description=( + "Add task_evaluations.baseline_cached_at for the 12h oracle " + "baseline TTL cache (capacity-aware pool eval)." + ), + apply=_migration_oracle_baseline_ttl_cache, + ), ) diff --git a/shared/common/models.py b/shared/common/models.py index d320e85..7e88611 100644 --- a/shared/common/models.py +++ b/shared/common/models.py @@ -666,6 +666,14 @@ class TaskEvaluation(Base): oracle_cost_usd: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) evaluated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=False), nullable=True) + # When the oracle baseline in ``baseline_response_json`` was first + # computed. Drives the 12h TTL: within the window every pool cycle + # that re-claims this task reuses the cached baseline (flat oracle + # cost + identical baseline → batch-independent scoring); past it + # the next claim recomputes and resets this stamp. + baseline_cached_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=False), nullable=True + ) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=False), default=utcnow, nullable=False) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=False), default=utcnow, nullable=False) diff --git a/tests/owner_api/test_capacity_pool_eval.py b/tests/owner_api/test_capacity_pool_eval.py new file mode 100644 index 0000000..c0d80e8 --- /dev/null +++ b/tests/owner_api/test_capacity_pool_eval.py @@ -0,0 +1,271 @@ +"""Capacity-aware continuous-pool evaluation (Option B). + +Covers the pure/unit-testable core: the oracle-baseline TTL migration, +the reconciled-oracle cache round-trip, the recency-ordered pool +selection with done-detection + K slicing, and build_claim_items' +healthy-deployment + cached-baseline behavior. +""" + +from __future__ import annotations + +import tempfile +from datetime import timedelta +from types import SimpleNamespace + +from sqlalchemy import create_engine, inspect, text + +from shared.common.database import Database +from shared.common.migrations import run_migrations +from shared.common.models import ( + Base, + EpochTargetSnapshot, + ManagedDeployment, + ManagedMinerSubmission, + TaskEvaluation, + TaskMinerResult, + utcnow, +) +from control_plane.owner_api.evaluation.run_manager import RunManager +from control_plane.owner_api.evaluation.evaluation_task_manager import ( + EvaluationTaskManager, + _ORACLE_BASELINE_TTL, +) +from validation.validator.engine import ( + _reconciled_from_payload, + _reconciled_to_payload, +) +from validation.validator.reconciler import ReconciledOracle + + +def _make_db(tmp_path) -> Database: + db = Database(f"sqlite+aiosqlite:///{tmp_path / 'pool.db'}") + db.create_all() + return db + + +# -- migration ------------------------------------------------------------- + + +def test_baseline_cached_at_present_on_fresh_db(): + e = create_engine(f"sqlite:///{tempfile.mktemp(suffix='.db')}") + run_migrations(e) + Base.metadata.create_all(e) + cols = {c["name"] for c in inspect(e).get_columns("task_evaluations")} + assert "baseline_cached_at" in cols + + +def test_migration_alters_preexisting_db_and_is_idempotent(): + path = tempfile.mktemp(suffix=".db") + e = create_engine(f"sqlite:///{path}") + # Simulate a pre-existing DB: table without the new column. + with e.begin() as c: + c.execute(text( + "CREATE TABLE task_evaluations (id TEXT PRIMARY KEY, " + "epoch_id TEXT, family_id TEXT, task_id TEXT, task_index INT, " + "status TEXT, claim_attempt_count INT, oracle_cost_usd REAL, " + "created_at TIMESTAMP, updated_at TIMESTAMP)" + )) + first = run_migrations(e) + assert "oracle_baseline_ttl_cache" in first + cols = {c["name"] for c in inspect(e).get_columns("task_evaluations")} + assert "baseline_cached_at" in cols + # Re-running applies nothing new (idempotent). + second = run_migrations(e) + assert "oracle_baseline_ttl_cache" not in second + + +# -- reconciled-oracle cache round-trip ------------------------------------ + + +def test_reconciled_payload_round_trips(): + original = ReconciledOracle( + expected_claims=["a", "b"], + must_not_claim=["x"], + oracle_status="consensus", + vendor_costs={"openai": 0.0021}, + vendor_citations={"openai": ["http://e.g/1"]}, + ) + restored = _reconciled_from_payload(_reconciled_to_payload(original)) + assert restored.expected_claims == ["a", "b"] + assert restored.must_not_claim == ["x"] + assert restored.oracle_status == "consensus" + assert restored.vendor_costs == {"openai": 0.0021} + + +def test_reconciled_from_payload_drops_unknown_keys(): + # A future validator may add fields; replaying an old cache must not + # explode. + payload = _reconciled_to_payload(ReconciledOracle(expected_claims=["q"])) + payload["some_future_field"] = 123 + restored = _reconciled_from_payload(payload) + assert restored.expected_claims == ["q"] + + +# -- pool_keep_deployment_ids --------------------------------------------- + + +def _snap_member(hk: str, dep_id: str, sub_id: str) -> dict: + return {"hotkey": hk, "metadata": {"deployment_id": dep_id, "submission_id": sub_id}} + + +def _seed_pool(session, *, run_id="run-1", family="general_chat", n=4, n_tasks=3): + """n miners, n_tasks tasks. Submissions created oldest→newest by index.""" + base = utcnow() + members = [] + for i in range(n): + sub = ManagedMinerSubmission( + id=f"sub-{i}", miner_hotkey=f"hk{i}", family_id=family, + archive_sha256=f"sha{i}", submission_block=i, submission_seq=i, + artifact_id=f"art-{i}", manifest_json={}, status="received", + created_at=base + timedelta(minutes=i), # i=n-1 is newest + ) + session.add(sub) + members.append(_snap_member(f"hk{i}", f"dep-{i}", f"sub-{i}")) + session.add(EpochTargetSnapshot( + run_id=run_id, family_id=family, benchmark_version="v1", + rubric_version="v1", judge_model="j", status="open", + frozen_validator_stakes_json={}, members_json=members, + )) + for t in range(n_tasks): + session.add(TaskEvaluation( + run_id=run_id, family_id=family, task_id=f"t{t}", + task_index=t, status="pending", + )) + session.commit() + + +def test_pool_selects_top_k_by_recency(tmp_path): + db = _make_db(tmp_path) + rm = RunManager(SimpleNamespace()) + with db.sessionmaker() as s: + _seed_pool(s, n=4, n_tasks=3) + out = rm.pool_keep_deployment_ids(s, family_id="general_chat", capacity_k=2) + # Newest two submissions (sub-3, sub-2) kept; all 4 eligible. + assert out["keep"] == {"dep-3", "dep-2"} + assert out["eligible_total"] == 4 + assert out["done"] == set() + assert out["snapshot_total"] == 4 + + +def test_pool_marks_fully_evaluated_miner_done(tmp_path): + db = _make_db(tmp_path) + rm = RunManager(SimpleNamespace()) + with db.sessionmaker() as s: + _seed_pool(s, n=3, n_tasks=2) + # hk0 has a row for both tasks → done (even all-error counts). + for t in ("t0", "t1"): + s.add(TaskMinerResult( + task_evaluation_id=f"te-{t}", run_id="run-1", + family_id="general_chat", task_id=t, miner_hotkey="hk0", + miner_response_json={}, miner_citations_json=[], + agreement_verdict="error", agreement_score=0.0, + miner_latency_seconds=0.0, latency_seconds=0.0, + proxy_cost_usd=0.0, judge_cost_usd=0.0, + )) + s.commit() + out = rm.pool_keep_deployment_ids(s, family_id="general_chat", capacity_k=10) + assert "dep-0" in out["done"] + assert out["eligible_total"] == 2 # hk1, hk2 still owe tasks + assert "dep-0" not in out["keep"] + + +def test_pool_empty_when_no_open_snapshot(tmp_path): + db = _make_db(tmp_path) + rm = RunManager(SimpleNamespace()) + with db.sessionmaker() as s: + out = rm.pool_keep_deployment_ids(s, family_id="general_chat", capacity_k=5) + assert out == { + "keep": set(), "done": set(), "eligible_total": 0, "snapshot_total": 0, + } + + +# -- build_claim_items: healthy predicate + cached baseline --------------- + + +def _owner_stub(tasks): + return SimpleNamespace( + run_evaluation_bundle=lambda *a, **k: {"tasks": tasks}, + ) + + +def test_claim_excludes_miners_without_healthy_deployment(tmp_path): + db = _make_db(tmp_path) + with db.sessionmaker() as s: + members = [ + _snap_member("hkA", "depA", "subA"), + _snap_member("hkB", "depB", "subB"), + ] + s.add(EpochTargetSnapshot( + run_id="run-1", family_id="general_chat", benchmark_version="v1", + rubric_version="v1", judge_model="j", status="open", + frozen_validator_stakes_json={}, members_json=members, + )) + s.add(ManagedDeployment( + id="depA", submission_id="subA", miner_hotkey="hkA", + family_id="general_chat", deployment_revision="revA", + image_ref="img", endpoint="http://a", + status="deployed_for_eval", health_status="healthy", + )) + s.add(ManagedDeployment( + id="depB", submission_id="subB", miner_hotkey="hkB", + family_id="general_chat", deployment_revision="revB", + image_ref="img", endpoint="http://b", + status="building", health_status="starting", + )) + te = TaskEvaluation( + run_id="run-1", family_id="general_chat", task_id="t0", + task_index=0, status="claimed", + ) + s.add(te) + s.commit() + mgr = EvaluationTaskManager(_owner_stub([{"task_id": "t0"}])) + items = mgr.build_claim_items( + s, claimed_tasks=[te], run_id="run-1", family_id="general_chat", + ) + assert len(items) == 1 + hks = {m["hotkey"] for m in items[0]["miners"]} + assert hks == {"hkA"} # hkB excluded — pod not healthy yet + + +def test_claim_attaches_cached_baseline_within_ttl(tmp_path): + db = _make_db(tmp_path) + with db.sessionmaker() as s: + s.add(EpochTargetSnapshot( + run_id="run-1", family_id="general_chat", benchmark_version="v1", + rubric_version="v1", judge_model="j", status="open", + frozen_validator_stakes_json={}, + members_json=[_snap_member("hkA", "depA", "subA")], + )) + s.add(ManagedDeployment( + id="depA", submission_id="subA", miner_hotkey="hkA", + family_id="general_chat", deployment_revision="r1", + image_ref="img", endpoint="http://a", + status="deployed_for_eval", health_status="healthy", + )) + fresh = TaskEvaluation( + run_id="run-1", family_id="general_chat", task_id="tf", + task_index=0, status="claimed", + baseline_response_json={"response_text": "cached!"}, + baseline_cached_at=utcnow() - timedelta(hours=1), + ) + stale = TaskEvaluation( + run_id="run-1", family_id="general_chat", task_id="ts", + task_index=1, status="claimed", + baseline_response_json={"response_text": "old"}, + baseline_cached_at=utcnow() - _ORACLE_BASELINE_TTL + - timedelta(minutes=1), + ) + s.add_all([fresh, stale]) + s.commit() + mgr = EvaluationTaskManager( + _owner_stub([{"task_id": "tf"}, {"task_id": "ts"}]) + ) + items = { + it["task_id"]: it + for it in mgr.build_claim_items( + s, claimed_tasks=[fresh, stale], + run_id="run-1", family_id="general_chat", + ) + } + assert items["tf"].get("cached_baseline") == {"response_text": "cached!"} + assert "cached_baseline" not in items["ts"] # past 12h TTL → recompute diff --git a/tests/owner_api/test_eval_feedback_endpoint.py b/tests/owner_api/test_eval_feedback_endpoint.py index 8c628fc..0ad1792 100644 --- a/tests/owner_api/test_eval_feedback_endpoint.py +++ b/tests/owner_api/test_eval_feedback_endpoint.py @@ -16,6 +16,9 @@ from types import SimpleNamespace from typing import Any +import pytest +from fastapi import HTTPException + from shared.common.database import Database from shared.common.models import EvalFeedback from control_plane.owner_api.evaluation.evaluation_task_manager import ( @@ -232,3 +235,27 @@ async def test_read_orders_by_created_at(tmp_path): req, run_id="run-1", hotkey="hk_alpha", ) assert [item.task_id for item in response.items] == ["t-3", "t-1", "t-2"] + + +async def test_read_returns_503_when_disabled(tmp_path, monkeypatch): + """Flipping ``EIREL_EVAL_FEEDBACK_ENABLED=0`` shuts off the miner-facing + GET; existing rows survive but are not served.""" + db = _make_db(tmp_path) + _seed(db, miner_hotkey="hk_alpha", task_id="t-1") + + monkeypatch.setenv("EIREL_EVAL_FEEDBACK_ENABLED", "0") + req = _make_request(db=db) + with pytest.raises(HTTPException) as exc_info: + await read_eval_feedback(req, run_id="run-1", hotkey="hk_alpha") + assert exc_info.value.status_code == 503 + + +def test_upsert_is_noop_when_disabled(tmp_path, monkeypatch): + """Flipping the flag off short-circuits the upsert so the validator + can keep submitting task results without accruing EvalFeedback rows.""" + monkeypatch.setenv("EIREL_EVAL_FEEDBACK_ENABLED", "0") + db = _make_db(tmp_path) + _seed(db) + + with db.sessionmaker() as session: + assert session.query(EvalFeedback).count() == 0 diff --git a/tests/owner_api/test_k8s_backend_mutating.py b/tests/owner_api/test_k8s_backend_mutating.py index 10bb6b6..4259566 100644 --- a/tests/owner_api/test_k8s_backend_mutating.py +++ b/tests/owner_api/test_k8s_backend_mutating.py @@ -327,7 +327,12 @@ async def test_ensure_runtime_rejects_archive_over_900kib(monkeypatch): mgr = _make_manager(cluster) _patch_time(monkeypatch) - large_content = b"x" * (901 * 1024) + # Incompressible payload: the ConfigMap stores the *gzipped* archive, + # so the size gate is on the compressed bytes. Random data won't + # shrink, so this genuinely overflows a ConfigMap. + import os + + large_content = os.urandom(2 * 1024 * 1024) archive = _make_archive({"big_file.bin": large_content}) with pytest.raises(RuntimeManagerError, match="too large"): diff --git a/tests/owner_api/test_k8s_helpers.py b/tests/owner_api/test_k8s_helpers.py index fe82420..c01e1a0 100644 --- a/tests/owner_api/test_k8s_helpers.py +++ b/tests/owner_api/test_k8s_helpers.py @@ -1,13 +1,35 @@ from __future__ import annotations +import io +import tarfile from types import SimpleNamespace +import pytest + from infra.miner_runtime._k8s_helpers import ( + _CODE_ARCHIVE_KEY, + _build_code_configmap, _build_network_policy, + _check_archive_configmap_size, + _extract_archive_to_dict, + _repack_clean_archive, parse_cpu_to_millis, parse_memory_to_bytes, ) -from infra.miner_runtime.runtime_manager import _deployment_manifest_common +from infra.miner_runtime.runtime_manager import ( + RuntimeManagerError, + _deployment_manifest_common, +) + + +def _make_targz(files: dict[str, bytes]) -> bytes: + buf = io.BytesIO() + with tarfile.open(fileobj=buf, mode="w:gz") as tar: + for path, data in files.items(): + info = tarfile.TarInfo(name=path) + info.size = len(data) + tar.addfile(info, io.BytesIO(data)) + return buf.getvalue() def _stub_manifest(port: int = 8080, health_path: str = "/healthz"): @@ -191,15 +213,28 @@ def test_manifest_emits_configmap_volume_when_code_configmap_name_set(): ) deployment, container = _get_deployment_and_container(manifests) pod_spec = deployment["spec"]["template"]["spec"] + + # Runtime container imports from the extracted emptyDir, never the + # raw ConfigMap. assert any( - vm["mountPath"] == "/submission" and vm["name"] == "submission-code" + vm["mountPath"] == "/submission" + and vm["name"] == "submission-code-extracted" for vm in container["volumeMounts"] ) - assert any( - v["name"] == "submission-code" - and v["configMap"]["name"] == "miner-test-123-code" - for v in pod_spec["volumes"] - ) + + # ConfigMap (single archive key) is mounted only into the init + # container, which unpacks it into the shared emptyDir. + init = pod_spec["initContainers"][0] + assert init["image"] == container["image"] + assert init["command"][:2] == ["python", "-c"] + assert "filter='data'" in init["command"][2] + init_mounts = {vm["name"]: vm["mountPath"] for vm in init["volumeMounts"]} + assert init_mounts["submission-code"] == "/submission-archive" + assert init_mounts["submission-code-extracted"] == "/submission" + + vols = {v["name"]: v for v in pod_spec["volumes"]} + assert vols["submission-code"]["configMap"]["name"] == "miner-test-123-code" + assert "emptyDir" in vols["submission-code-extracted"] def test_manifest_emits_network_policy_when_emit_network_policy_true(): @@ -273,3 +308,66 @@ def test_build_network_policy_blocks_direct_https_egress(): "egress rule allows unscoped outbound HTTPS — miners would " "bypass provider-proxy and reach api.openai.com directly" ) + + +# -- single-archive ConfigMap ------ + + +def test_configmap_has_single_archive_key_for_nested_submission(): + """A submission with subdirectories must NOT produce per-path + ConfigMap keys (those fail k8s validation with 422). One opaque + archive key only.""" + archive = _make_targz({ + "app.py": b"x = 1\n", + "bench/tasks.py": b"# nested\n", + "tests/test_middleware.py": b"# deep/nested\n", + "pkg/sub/mod.py": b"# very/deep\n", + }) + files = _extract_archive_to_dict(archive) + clean = _repack_clean_archive(files) + cm = _build_code_configmap( + name="miner-x-code", submission_id="x", archive_bytes=clean, + ) + assert list(cm["binaryData"].keys()) == [_CODE_ARCHIVE_KEY] + # No '/' in any key — the whole class of 422 failures is gone. + assert all("/" not in k for k in cm["binaryData"]) + + +def test_repacked_archive_round_trips_and_drops_junk(): + archive = _make_targz({ + "app.py": b"MAIN\n", + "pkg/util.py": b"UTIL\n", + "__pycache__/app.cpython-312.pyc": b"\x00junk", + ".git/config": b"[core]\n", + }) + files = _extract_archive_to_dict(archive) + clean = _repack_clean_archive(files) + extracted: dict[str, bytes] = {} + with tarfile.open(fileobj=io.BytesIO(clean), mode="r:gz") as tar: + for m in tar.getmembers(): + f = tar.extractfile(m) + if f is not None: + extracted[m.name] = f.read() + assert extracted == {"app.py": b"MAIN\n", "pkg/util.py": b"UTIL\n"} + + +def test_repacked_archive_is_deterministic(): + archive = _make_targz({"b.py": b"B", "a/c.py": b"C", "a.py": b"A"}) + files = _extract_archive_to_dict(archive) + assert _repack_clean_archive(files) == _repack_clean_archive(files) + + +def test_archive_size_check_rejects_oversized(): + big = b"\x00" * (2 * 1024 * 1024) + with pytest.raises(RuntimeManagerError, match="too large for ConfigMap"): + _check_archive_configmap_size(big) + + +def test_archive_size_check_allows_small(): + _check_archive_configmap_size(b"small archive") # no raise + + +def test_extract_archive_rejects_path_escape(): + escape = _make_targz({"../evil.py": b"PWN"}) + with pytest.raises(ValueError, match="escape root"): + _extract_archive_to_dict(escape) diff --git a/tests/owner_api/test_weights_endpoint.py b/tests/owner_api/test_weights_endpoint.py index bdd3317..25baa09 100644 --- a/tests/owner_api/test_weights_endpoint.py +++ b/tests/owner_api/test_weights_endpoint.py @@ -4,16 +4,7 @@ from control_plane.owner_api.app import app from control_plane.owner_api._helpers import utcnow -from shared.common.models import EvaluationRun, RunFamilyResult, ValidatorRecord -from tests.conftest import signed_headers - - -def _register_validator(hotkey: str) -> None: - services = app.state.services - with services.db.sessionmaker() as session: - if session.get(ValidatorRecord, hotkey) is None: - session.add(ValidatorRecord(hotkey=hotkey, uid=1)) - session.commit() +from shared.common.models import EvaluationRun, RunFamilyResult def _make_run(run_id: str, *, sequence: int, status: str) -> None: @@ -62,14 +53,8 @@ def _make_family_result( session.commit() -async def test_weights_endpoint_returns_not_ready_when_no_completed_run( - client, identities -): - signer = identities["validator-1"]["signer"] - _register_validator(signer.hotkey) - - hdrs = signed_headers(signer, method="GET", path="/v1/weights", body=b"") - resp = await client.get("/v1/weights", headers=hdrs) +async def test_weights_endpoint_returns_not_ready_when_no_completed_run(client): + resp = await client.get("/v1/weights") assert resp.status_code == 200 body = resp.json() @@ -86,8 +71,6 @@ async def test_weights_endpoint_returns_not_ready_when_no_completed_run( async def test_weights_endpoint_returns_winner_weights_when_run_completed( client, identities ): - signer = identities["validator-1"]["signer"] - _register_validator(signer.hotkey) miner_hotkey = identities["miner"]["signer"].hotkey _make_run("run-weights-1", sequence=1, status="completed") @@ -99,8 +82,7 @@ async def test_weights_endpoint_returns_winner_weights_when_run_completed( best_raw_score=0.75, ) - hdrs = signed_headers(signer, method="GET", path="/v1/weights", body=b"") - resp = await client.get("/v1/weights", headers=hdrs) + resp = await client.get("/v1/weights") assert resp.status_code == 200 body = resp.json() @@ -115,11 +97,8 @@ async def test_weights_endpoint_returns_winner_weights_when_run_completed( async def test_weights_endpoint_returns_empty_winner_when_family_has_no_winner( - client, identities + client, ): - signer = identities["validator-1"]["signer"] - _register_validator(signer.hotkey) - _make_run("run-weights-2", sequence=2, status="completed") _make_family_result( "run-weights-2", @@ -128,8 +107,7 @@ async def test_weights_endpoint_returns_empty_winner_when_family_has_no_winner( winner_hotkey=None, ) - hdrs = signed_headers(signer, method="GET", path="/v1/weights", body=b"") - resp = await client.get("/v1/weights", headers=hdrs) + resp = await client.get("/v1/weights") assert resp.status_code == 200 body = resp.json() diff --git a/validation/validator/engine.py b/validation/validator/engine.py index f01515e..62b6d9a 100644 --- a/validation/validator/engine.py +++ b/validation/validator/engine.py @@ -4,7 +4,7 @@ import os import secrets import time -from dataclasses import dataclass +from dataclasses import dataclass, fields as _dc_fields from typing import Any from urllib.parse import urlparse, urlunparse @@ -82,6 +82,28 @@ def model_dump(self, *, mode: str = "python") -> dict[str, Any]: } +def _reconciled_to_payload(reconciled: "ReconciledOracle") -> dict[str, Any]: + """Serialize a ReconciledOracle for the oracle-baseline TTL cache. + + Embedded under ``_reconciled`` in the submitted ``baseline_response`` + so a later pool cycle reconstructs the *exact* reconciled oracle — + identical expected_claims/must_not_claim/vendor data — making the + cached judging input byte-identical to the first compute (the + invariant that keeps scoring batch-independent). + """ + return {f.name: getattr(reconciled, f.name) for f in _dc_fields(reconciled)} + + +def _reconciled_from_payload(payload: dict[str, Any]) -> "ReconciledOracle": + """Rebuild a ReconciledOracle from a cached ``_reconciled`` blob. + + Unknown keys are dropped so a future field addition doesn't break + replay of baselines cached by an older validator. + """ + known = {f.name for f in _dc_fields(ReconciledOracle)} + return ReconciledOracle(**{k: v for k, v in payload.items() if k in known}) + + def _select_pairwise_reference( *, reconciled: "ReconciledOracle", preferred_vendor: str, ) -> tuple[str, str]: @@ -1157,6 +1179,16 @@ class _TaskProxy: # carry this field; default to None (= deterministic). task_obj.oracle_source = task_payload.get("oracle_source") + # Oracle baseline 12h TTL cache. owner-api attaches + # ``cached_baseline`` when this task's baseline is still within + # the window; we then skip oracle enrichment entirely (no + # OpenAI/Gemini/Grok/reconciler calls) and replay the exact + # reconciled oracle from the first compute — keeping every pool + # cycle's judging input identical (batch-independent scoring) + # at flat oracle cost. + cached_baseline = task_claim.get("cached_baseline") + using_cached_baseline = bool(cached_baseline) + # Oracle enrichment runs in parallel with miner dispatch. # three_oracle items → 3 frontier-LLM calls + 1 reconciler. # deterministic items → 1 frontier-LLM call (the configured @@ -1169,15 +1201,17 @@ class _TaskProxy: # $10/1k web-search adder on self-contained tasks. # Result is consumed by every ``_judge_miner`` call for this # task. Cache lifetime = this task evaluation. - reconciled_task = asyncio.create_task( - _enrich_task_oracle( - task_obj, - fanout=oracle_fanout, - reconciler=reconciler, - pairwise_vendor=pairwise_vendor, - web_search=web_search_flag, + reconciled_task = None + if not using_cached_baseline: + reconciled_task = asyncio.create_task( + _enrich_task_oracle( + task_obj, + fanout=oracle_fanout, + reconciler=reconciler, + pairwise_vendor=pairwise_vendor, + web_search=web_search_flag, + ) ) - ) miner_tasks = [ asyncio.create_task(_invoke_one_miner(m, task_obj, task_id)) for m in miners @@ -1196,56 +1230,79 @@ class _TaskProxy: t.cancel() return "submit_failed" - # Pairwise comparator: pick the chosen oracle's answer as the - # reference instead of paying for a separate OpenAI baseline - # call. Fallback chain handles vendors that errored or - # ``deterministic`` items where no oracle ran. - try: - reconciled_for_baseline = await reconciled_task - except Exception as enrich_exc: - logger.warning( - "oracle enrichment crashed for task=%s: %s; " - "falling back to disputed (no comparator text available)", - task_id, enrich_exc, + if using_cached_baseline: + # Replay the first compute's reconciled oracle + reference + # text verbatim. No oracle spend; judging input identical to + # the original cycle (batch-independent scoring invariant). + reconciled_for_baseline = _reconciled_from_payload( + (cached_baseline.get("_reconciled") or {}) + if isinstance(cached_baseline, dict) else {} ) - reconciled_for_baseline = ReconciledOracle( - expected_claims=[], - must_not_claim=list( - (task_obj.expected_output or {}).get("must_not_claim") or [] - ), - oracle_status="disputed", - disagreement_note=f"enrichment_crashed: {enrich_exc!r}", + baseline_source = str( + (cached_baseline or {}).get("source_vendor") or "cached" + ) + baseline = _SyntheticBaselineResponse( + response_text=str((cached_baseline or {}).get("response_text") or ""), + citations=list((cached_baseline or {}).get("citations") or []), + cost_usd=0.0, + latency_seconds=0.0, + source_vendor=baseline_source, + ) + logger.info( + "task_eval=%s baseline=ok source=cached(%s)", + task_evaluation_id[:8], baseline_source, + ) + else: + # Pairwise comparator: pick the chosen oracle's answer as the + # reference instead of paying for a separate OpenAI baseline + # call. Fallback chain handles vendors that errored or + # ``deterministic`` items where no oracle ran. + try: + reconciled_for_baseline = await reconciled_task + except Exception as enrich_exc: + logger.warning( + "oracle enrichment crashed for task=%s: %s; " + "falling back to disputed (no comparator text available)", + task_id, enrich_exc, + ) + reconciled_for_baseline = ReconciledOracle( + expected_claims=[], + must_not_claim=list( + (task_obj.expected_output or {}).get("must_not_claim") or [] + ), + oracle_status="disputed", + disagreement_note=f"enrichment_crashed: {enrich_exc!r}", + ) + baseline_text, baseline_source = _select_pairwise_reference( + reconciled=reconciled_for_baseline, + preferred_vendor=pairwise_vendor, + ) + # Carry the source vendor's web-search citations through onto + # the baseline so the dashboard can render the reference + # answer's URLs alongside the miner's. Non-vendor sources + # (deterministic, consensus_claim, none) have none to copy. + vendor_key = baseline_source.removesuffix("_fallback") + baseline_citations = [ + {"url": str(u), "title": ""} + for u in (reconciled_for_baseline.vendor_citations or {}).get(vendor_key, []) + if u + ] + # Synthetic ``baseline`` shim that still satisfies the rest + # of the engine's expected fields (cost, latency, citations). + # ``cost_usd=0.0`` because reusing the cached oracle answer + # adds no incremental spend. + baseline = _SyntheticBaselineResponse( + response_text=baseline_text, + citations=baseline_citations, + cost_usd=0.0, + latency_seconds=0.0, + source_vendor=baseline_source, + ) + logger.info( + "task_eval=%s baseline=ok source=%s", + task_evaluation_id[:8], + baseline_source, ) - baseline_text, baseline_source = _select_pairwise_reference( - reconciled=reconciled_for_baseline, - preferred_vendor=pairwise_vendor, - ) - # Carry the source vendor's web-search citations through onto the - # baseline so the dashboard can render the reference answer's URLs - # alongside the miner's. Non-vendor sources (deterministic, - # consensus_claim, none) have no per-vendor citations to copy. - vendor_key = baseline_source.removesuffix("_fallback") - baseline_citations = [ - {"url": str(u), "title": ""} - for u in (reconciled_for_baseline.vendor_citations or {}).get(vendor_key, []) - if u - ] - # Synthetic ``baseline`` shim that still satisfies the rest of - # the engine's expected fields (cost, latency, citations). - # ``cost_usd=0.0`` because reusing the cached oracle answer - # adds no incremental spend. - baseline = _SyntheticBaselineResponse( - response_text=baseline_text, - citations=baseline_citations, - cost_usd=0.0, - latency_seconds=0.0, - source_vendor=baseline_source, - ) - logger.info( - "task_eval=%s baseline=ok source=%s", - task_evaluation_id[:8], - baseline_source, - ) # Per-miner fan-out outcome. miner_runs is a list of either # ``(hotkey, BenchmarkTaskRun)`` tuples or raised exceptions @@ -1884,10 +1941,17 @@ def _cost_usd(payload: dict[str, Any] | None) -> float: oracle_layer_cost + judge_cost, ) - # Submit the batch + # Submit the batch. Embed the reconciled oracle under + # ``_reconciled`` so owner-api's TTL cache can replay the exact + # judging input on later pool cycles (additive key — older + # readers ignore it; schema stays stable). + baseline_payload = baseline.model_dump(mode="json") + baseline_payload["_reconciled"] = _reconciled_to_payload( + reconciled_for_baseline + ) result_path = f"/v1/families/{family_id}/task-evaluations/{task_evaluation_id}/result" result_body = { - "baseline_response": baseline.model_dump(mode="json"), + "baseline_response": baseline_payload, "miner_results": judge_results, "validator_hotkey": signer.hotkey, "judge_model": judge_model,