from __future__ import annotations
from datetime import datetime
from pathlib import Path
from typing import Annotated, Literal
from pydantic import BaseModel, ConfigDict, Field, StringConstraints, model_validator
SHA = Annotated[str, StringConstraints(pattern=r"^[0-9a-f]{40}$")]
Repository = Annotated[str, StringConstraints(pattern=r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")]
Metric = Literal[
"cold",
"warm",
"install",
"bundle",
"disk",
"rss",
"pss",
"kernel_start",
"kernel_exec",
"bash",
"git_status",
"output",
"mixed",
"interrupt",
"snapshot",
"restore",
"kernel_rss",
"loaded_rss",
"resume_large",
"resume_large_cpu",
"switch_large",
"switch_large_cpu",
"agents_view",
"agents_view_cpu",
"agents_roster",
"agents_roster_cpu",
"agents_open",
"agents_open_cpu",
"agents_reopen",
"agents_reopen_cpu",
"subagent_open",
"subagent_open_cpu",
"parent_open",
"parent_open_cpu",
"ui_rss",
"scheduled_catalog",
"scheduled_catalog_cpu",
"scheduled_catalog_warm",
"scheduled_catalog_warm_cpu",
"cold_open_catalog",
"cold_open_catalog_cpu",
]
UI_METRIC_KEYS = frozenset(
{
"resume_large",
"resume_large_cpu",
"switch_large",
"switch_large_cpu",
"agents_view",
"agents_view_cpu",
"agents_roster",
"agents_roster_cpu",
"agents_open",
"agents_open_cpu",
"agents_reopen",
"agents_reopen_cpu",
"subagent_open",
"subagent_open_cpu",
"parent_open",
"parent_open_cpu",
"ui_rss",
"scheduled_catalog",
"scheduled_catalog_cpu",
"scheduled_catalog_warm",
"scheduled_catalog_warm_cpu",
"cold_open_catalog",
"cold_open_catalog_cpu",
}
)
NonNegative = Annotated[float, Field(ge=0, le=1e15, allow_inf_nan=False)]
ROOT = Path(__file__).resolve().parent
UV_VERSION = "0.12.9"
class StrictModel(BaseModel):
model_config = ConfigDict(extra="forbid", allow_inf_nan=False, strict=True)
class Config(StrictModel):
image: Annotated[str, StringConstraints(pattern=r"^.+@sha256:[0-9a-f]{64}$")]
region: str
cpu_cores: Annotated[float, Field(gt=0, le=16)]
memory_gb: Annotated[float, Field(gt=0, le=64)]
disk_gb: Annotated[float, Field(gt=0, le=1000)]
trials: Annotated[int, Field(ge=3, le=50)]
install_trials: Annotated[int, Field(ge=1, le=10)]
failure_limit: Annotated[int, Field(ge=1, le=10)] = 2
ui_trials: Annotated[int, Field(ge=1, le=50)] = 3
debounce_seconds: Annotated[int, Field(ge=0, le=300)]
timeout_seconds: Annotated[int, Field(ge=60, le=3600)]
ttl_minutes: Annotated[int, Field(ge=1, le=120)]
budget_usd: Annotated[float, Field(gt=0, le=20)]
cpu_usd_per_hour: NonNegative
memory_usd_per_gb_hour: NonNegative
disk_usd_per_gb_hour: NonNegative
@model_validator(mode="after")
def validate_limits(self) -> Config:
if self.ttl_minutes * 60 <= self.timeout_seconds:
raise ValueError("Sandbox TTL must exceed the run timeout")
return self
@classmethod
def load(cls, path: Path = ROOT / "config.json") -> Config:
return cls.model_validate_json(path.read_text())
def hourly_cost(self) -> float:
return (
self.cpu_cores * self.cpu_usd_per_hour
+ self.memory_gb * self.memory_usd_per_gb_hour
+ self.disk_gb * self.disk_usd_per_gb_hour
)
class Observation(StrictModel):
trial: Annotated[int, Field(ge=0, le=100)]
value: NonNegative | None = None
error: Annotated[str, Field(max_length=500)] | None = None
@model_validator(mode="after")
def validate_outcome(self) -> Observation:
if (self.value is None) == (self.error is None):
raise ValueError("An observation must contain either a value or an error")
return self
class ProcessMemory(StrictModel):
pid: Annotated[int, Field(gt=0)]
name: Annotated[str, Field(max_length=80)]
rss: NonNegative
pss: NonNegative | None = None
cpu: NonNegative = 0.0
class Side(StrictModel):
sha: SHA
metrics: dict[Metric, list[Observation]] = Field(default_factory=dict)
processes: list[ProcessMemory] = Field(default_factory=list, max_length=128)
runtime: dict[str, str] = Field(default_factory=dict)
timings: dict[Literal["provision", "setup", "build"], NonNegative] = Field(default_factory=dict)
artifacts: dict[str, NonNegative] = Field(default_factory=dict)
error: Annotated[str, Field(max_length=500)] | None = None
class Request(StrictModel):
repository: Repository
source_repository: Repository
sha: SHA
harness_sha: SHA
pr: Annotated[int, Field(ge=0)]
run_id: Annotated[int, Field(ge=1)]
attempt: Annotated[int, Field(ge=1)]
role: Literal["main", "pr"]
config: Config
class Result(StrictModel):
request: Request
side: Side
class SandboxUsage(StrictModel):
id: Annotated[str, StringConstraints(pattern=r"^[a-zA-Z0-9_-]+$")]
role: Literal["main", "pr"]
seconds: NonNegative
estimated_usd: NonNegative
deleted: bool
class Report(StrictModel):
schema_version: Literal[1] = 1
repository: Repository
head_repository: Repository
pr: Annotated[int, Field(ge=0)]
run_id: Annotated[int, Field(ge=1)]
attempt: Annotated[int, Field(ge=1)] = 1
harness_sha: SHA
base_sha: SHA
head_sha: SHA
started_at: datetime
finished_at: datetime | None = None
config: Config
status: Literal["pending-trust", "running", "completed", "partial", "failed", "canceled"] = "running"
main: Side
pr_head: Side
sandboxes: list[SandboxUsage] = Field(default_factory=list, max_length=2)
errors: list[Annotated[str, Field(max_length=500)]] = Field(default_factory=list, max_length=30)
warnings: list[Annotated[str, Field(max_length=500)]] = Field(default_factory=list, max_length=30)
@model_validator(mode="after")
def validate_revisions(self) -> Report:
if self.main.sha != self.base_sha or self.pr_head.sha != self.head_sha:
raise ValueError("Result revisions do not match the request")
for side in (self.main, self.pr_head):
for metric, samples in side.metrics.items():
expected = self.config.install_trials if metric == "install" else self.config.trials
if metric in ("bundle", "disk"):
expected = 1
if metric in UI_METRIC_KEYS:
expected = self.config.ui_trials
if (
len(samples) > expected
or len({s.trial for s in samples}) != len(samples)
or any(sample.trial >= expected for sample in samples)
):
raise ValueError("Duplicate or excessive observations")
return self
PHASE_METRICS: dict[str, tuple[Metric, ...]] = {
"prepare": ("bundle",),
"install": ("install",),
"measure": ("cold", "warm", "rss", "disk"),
"runtime": (
"kernel_start",
"kernel_exec",
"bash",
"git_status",
"output",
"mixed",
"interrupt",
"snapshot",
"restore",
"kernel_rss",
"loaded_rss",
),
"ui": (
"resume_large",
"resume_large_cpu",
"switch_large",
"switch_large_cpu",
"agents_view",
"agents_view_cpu",
"agents_roster",
"agents_roster_cpu",
"agents_open",
"agents_open_cpu",
"agents_reopen",
"agents_reopen_cpu",
"subagent_open",
"subagent_open_cpu",
"parent_open",
"parent_open_cpu",
"ui_rss",
"scheduled_catalog",
"scheduled_catalog_cpu",
"scheduled_catalog_warm",
"scheduled_catalog_warm_cpu",
"cold_open_catalog",
"cold_open_catalog_cpu",
),
}
def trial_errors(side: Side, phase: str, trial: int) -> list[str]:
metrics = [metric for metric in PHASE_METRICS[phase] if metric != "disk" or trial == 0]
samples = {metric: [s for s in side.metrics.get(metric, []) if s.trial == trial] for metric in metrics}
errors = [f"{metric}: {sample.error}" for metric in metrics for sample in samples[metric] if sample.error]
if side.error:
errors.insert(0, side.error)
return errors or [f"{metric}: measurement missing" for metric in metrics if not samples[metric]]
def side_complete(side: Side, trials: int, installs: int, ui_trials: int) -> bool:
for metrics in PHASE_METRICS.values():
for metric in metrics:
count = (
1
if metric in ("bundle", "disk")
else installs
if metric == "install"
else ui_trials
if metric in UI_METRIC_KEYS
else trials
)
samples = side.metrics.get(metric, [])
if (
{sample.trial for sample in samples} != set(range(count))
or any(sample.value is None for sample in samples)
or len(samples) != count
):
return False
return not side.error
def report_complete(report: Report) -> bool:
return not report.errors and all(
side_complete(side, report.config.trials, report.config.install_trials, report.config.ui_trials)
for side in (report.main, report.pr_head)
)
def write_json(path: Path, value: BaseModel) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_suffix(".tmp")
temporary.write_text(value.model_dump_json(indent=2) + "\n")
temporary.replace(path)
def load_report(path: Path) -> Report:
if path.stat().st_size > 2_000_000:
raise ValueError("Report exceeds the size limit")
return Report.model_validate_json(path.read_text())