Skip to content

Capacity-aware pool evaluation #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kyron1112567 merged 2 commits into from
May 15, 2026
Merged

Capacity-aware pool evaluation #5

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e7ca11e
feat(eval): capacity-aware continuous-pool evaluation + permanent Con…
kyron1112567 May 15, 2026
646aae0
feat(dashboard): per-task redaction + runtime tool-token secrets
kyron1112567 May 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 32 additions & 14 deletions control_plane/owner_api/dashboard/queries.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {}

Expand Down Expand Up @@ -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
Expand All @@ -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)),
Expand All @@ -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,
Expand Down Expand Up @@ -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
]
Expand Down
159 changes: 125 additions & 34 deletions control_plane/owner_api/deployment/deployment_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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)
Expand All @@ -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:
Expand Down
Loading
Loading