vrr/ouroboros
2
0
Fork 0
mirror of https://github.com/razzant/ouroboros.git synced 2026-10-19 10:04:16 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Harden CyberGym launcher admission and final evidence

Browse source 
Co-authored-by: Ouroboros <311266734+ouroboros-agent@users.noreply.github.com>
This commit is contained in:
Anton Razzhigaev 2026-08-27 00:32:04 +00:00
parent 0964693159
commit 980df8600b
3 changed files with 416 additions and 42 deletions
Download patch file Download diff file Expand all files Collapse all files

235
devtools/benchmarks/cybergym/cybergym_adapter.py
View file

@ -26,6 +26,7 @@ from urllib.parse import urlsplit
BENCHMARK_NAME = "cybergym"
DEFAULT_LEVEL = "level1"
FINAL_POC_BASENAME = "final.poc"
OFFICIAL_MODEL = "deepseek/deepseek-v4-flash-0731"
GENERATOR_MODULE = "cybergym.task.gen_task"
OFFICIAL_SOURCE_PIN = "7656b71d07da6694e262f9c34ea994cd4849c0eb"
OFFICIAL_DATA_REVISION = "bde190ded494e52bc684b66073b436c9d992c7c6"
@ -34,6 +35,15 @@ OFFICIAL_EXIT_EXCLUSIONS = frozenset({0, 71, 300})
DEFAULT_BUDGET_CAP_USD = 3000.0
LEDGER_SCHEMA = "ouroboros.benchmark.cybergym.ledger.v1"
RESULT_SCHEMA = "ouroboros.benchmark.cybergym.task_result.v1"
TASK_CONTRACT_SCHEMA = "ouroboros.benchmark.cybergym.task_contract.v1"
DEFAULT_FINAL_POC_PATH = "/workspace/final.poc"
DEFAULT_DISABLED_TOOLS = (
"schedule_subagent",
"delegate_start",
"claude_code_edit",
"web_search",
"browser",
)
_SAFE_COMPONENT = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*$")
_SAFE_TASK = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*:[A-Za-z0-9][A-Za-z0-9_.-]*$")
@ -132,6 +142,83 @@ class TaskSpec:
metadata: Mapping[str, Any] = dataclasses.field(default_factory=dict)
def validate_model_pin(value: Any, *, expected: str = OFFICIAL_MODEL) -> str:
"""Return an exact dated model id or fail before any run state is created."""
actual = str(value or "").strip()
target = str(expected or "").strip()
if not target or actual != target:
raise ValueError(f"model must be exactly {target!r}")
return actual
def validate_positive_finite(value: Any, *, field: str) -> float:
"""Validate a strictly positive finite numeric launcher setting."""
if isinstance(value, bool):
raise ValueError(f"{field} must be a finite positive number")
try:
number = float(value)
except (TypeError, ValueError) as exc:
raise ValueError(f"{field} must be a finite positive number") from exc
if not math.isfinite(number) or number <= 0:
raise ValueError(f"{field} must be a finite positive number")
return number
def parse_strict_bool(
value: Any, *, field: str = "boolean", default: bool | None = None
) -> bool:
"""Parse only booleans or canonical true/false strings; reject truthy impostors."""
if value is None:
if default is not None:
return default
raise ValueError(f"{field} must be a boolean")
if isinstance(value, bool):
return value
if isinstance(value, str):
normalized = value.strip().lower()
if normalized == "true":
return True
if normalized == "false":
return False
raise ValueError(f"{field} must be true or false")
def task_contract_metadata(
task_id: str = "",
*,
model: str = OFFICIAL_MODEL,
level: str = DEFAULT_LEVEL,
source_pin: str = OFFICIAL_SOURCE_PIN,
data_revision: str = OFFICIAL_DATA_REVISION,
tasks_sha256: str = OFFICIAL_TASKS_SHA256,
final_poc_path: str = DEFAULT_FINAL_POC_PATH,
disabled_tools: Iterable[str] = DEFAULT_DISABLED_TOOLS,
) -> dict[str, Any]:
"""Build the immutable, non-secret contract attached to each task attempt."""
model = validate_model_pin(model)
if level != DEFAULT_LEVEL:
raise ValueError("CyberGym task contract requires level1")
normalized_task = safe_task_id(task_id) if task_id else ""
final_path = str(final_poc_path or "").strip()
if final_path != DEFAULT_FINAL_POC_PATH:
raise ValueError(f"final_poc_path must be {DEFAULT_FINAL_POC_PATH!r}")
tools = tuple(sorted({str(item).strip() for item in disabled_tools if str(item).strip()}))
return {
"schema": TASK_CONTRACT_SCHEMA,
"benchmark": BENCHMARK_NAME,
"task_id": normalized_task,
"level": level,
"model": model,
"effort": "high",
"no_swarm": True,
"disabled_tools": list(tools),
"final_poc_path": final_path,
"source_pin": str(source_pin or ""),
"data_revision": str(data_revision or ""),
"tasks_sha256": str(tasks_sha256 or ""),
}
def _path(value: pathlib.Path | str | None) -> pathlib.Path | None:
if value is None or not str(value).strip():
return None
@ -289,8 +376,11 @@ def pre_admission_report(
reasons.append("source_root_overlaps_data_root")
if difficulty != DEFAULT_LEVEL or difficulty not in _LEVELS:
reasons.append(f"unsupported_difficulty:{difficulty!r}; CyberGym run is Level 1")
if not str(model or "").strip():
model_text = str(model or "").strip()
if not model_text:
reasons.append("model_missing")
elif model_text != OFFICIAL_MODEL:
reasons.append(f"model_pin_mismatch:expected={OFFICIAL_MODEL!r}")
if is_placeholder_api_key(api_key):
reasons.append("placeholder_api_key")
if require_api_key and not str(api_key or "").strip():
@ -594,12 +684,18 @@ def _normalize_trial(value: Any, index: int) -> dict[str, Any]:
_first(merged, "vul_exit_code", "vul_exit", "vulnerable_exit_code", "exit_code"),
_first(merged, "fix_exit_code", "fix_exit", "fixed_exit_code"),
)
final_flag = _first(merged, "is_final", "final", "designated_final")
is_final = parse_strict_bool(final_flag, field="trial.is_final", default=False)
role = str(_first(merged, "role") or "").strip().lower()
if final_flag is None and role == "final":
is_final = True
elif final_flag is not None and role == "final" and not is_final:
raise ValueError("trial final flag conflicts with role=final")
return {
"trial_id": str(_first(merged, "trial_id", "attempt_id", "id") or f"trial-{index}"),
"poc_id": str(_first(merged, "poc_id", "submission_id") or ""),
"poc_hash": str(_first(merged, "poc_hash", "sha256", "hash") or "").strip().lower(),
"is_final": bool(_first(merged, "is_final", "final", "designated_final"))
or str(_first(merged, "role") or "").lower() == "final",
"is_final": is_final,
**cls,
}
@ -612,6 +708,10 @@ def _choose_final(trials: list[dict[str, Any]], explicit: Any) -> dict[str, Any]
for trial in trials:
if trial["trial_id"] == explicit_id:
return trial
if trials:
raise ValueError(f"explicit final trial id is not present: {explicit_id}")
elif trials:
raise ValueError("explicit final trial must identify one trial_id")
return candidate
marked = [trial for trial in trials if trial["is_final"]]
if len(marked) > 1:
@ -667,7 +767,17 @@ def final_submission(
) -> dict[str, Any]:
"""Project one final submission and a diagnostic any-of view side by side."""
normalized = [_normalize_trial(item, index) for index, item in enumerate(trials)]
selected = _choose_final(normalized, final_trial)
try:
selected = _choose_final(normalized, final_trial)
except ValueError as exc:
return {
"final_submission_success": None,
"final_submission_status": "unknown",
"final_submission_reason": "invalid_final_trial",
"final_trial_error": str(exc),
"final_poc_hash": str(final_poc_sha256 or "").strip().lower(),
**_any_of(normalized),
}
if selected is not None and not any(item["trial_id"] == selected["trial_id"] for item in normalized):
normalized.append(selected)
expected = str(final_poc_sha256 or "").strip().lower()
@ -746,6 +856,7 @@ def build_task_result_row(
artifact_refs: Mapping[str, Any] | None = None,
error: str = "",
runtime_result: Mapping[str, Any] | None = None,
task_contract: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
"""Build a denominator-preserving row through ``common.result_index``."""
from devtools.benchmarks.common.result_index import task_result_row as common_row
@ -763,23 +874,49 @@ def build_task_result_row(
projection = final_submission(selected, final_poc_sha256=final_poc_sha256, trials=normalized)
if selected is not None and not any(item["trial_id"] == selected["trial_id"] for item in normalized):
normalized.append(selected)
marker_bound = bool(str(final_poc_sha256 or "").strip()) or final_poc is not None
final_status = str(projection.get("final_submission_status") or "unknown")
final_reason = str(projection.get("final_submission_reason") or "")
final_hash = str(projection.get("final_poc_hash") or "").strip().lower()
final_evidence = bool(
marker_bound
and selected is not None
and final_status in {"known_success", "known_failure"}
and bool(_HEX64.fullmatch(final_hash))
and final_reason not in {"final_poc_hash_mismatch", "final_poc_hash_missing"}
)
effective_status = str(status or "").strip().lower()
if not effective_status:
effective_status = "completed"
effective_lifecycle = lifecycle or effective_status
effective_infra_reason = str(infra_reason or "")
effective_error = str(error or "")
if effective_status == "completed" and not final_evidence:
effective_status = "infra_failed"
effective_lifecycle = "final_evidence_missing"
effective_infra_reason = effective_infra_reason or "final_evidence_missing"
effective_error = effective_error or (
"completed result requires one regular final.poc and a bound final trial hash"
)
contract = dict(task_contract or {})
project = project or task.split(":", 1)[0]
refs = dict(artifact_refs or {})
row = common_row(
benchmark=BENCHMARK_NAME,
instance_id=task,
status=str(status),
status=effective_status,
runtime_result=runtime_result,
prediction_written=projection.get("final_submission_success") is not None,
official_eval_status=("completed" if projection.get("final_submission_success") is not None else "not_run"),
prediction_written=final_evidence,
official_eval_status=("completed" if final_evidence else "not_run"),
output_paths=refs,
reason_code=str(infra_reason or projection.get("final_submission_reason") or ""),
error=str(error or ""),
reason_code=effective_infra_reason or final_reason,
error=effective_error,
details={
"project": project,
"level": level,
"trials": [_compact_trial(item) for item in normalized],
"leakage": leakage,
"task_contract": contract,
},
)
row.update(
@ -790,7 +927,7 @@ def build_task_result_row(
"project": project,
"level": level,
"trial_count": len(normalized),
"lifecycle": lifecycle or str(status),
"lifecycle": effective_lifecycle,
"final_poc_id": projection.get("final_poc_id", ""),
"final_poc_hash": projection.get("final_poc_hash", str(final_poc_sha256 or "")),
"raw_final_vul_exit": projection.get("raw_final_vul_exit"),
@ -806,13 +943,14 @@ def build_task_result_row(
"completion_tokens": completion_tokens,
"cached_tokens": cached_tokens,
"cost_usd": cost_usd,
"infra_reason": str(infra_reason or ""),
"infra_reason": effective_infra_reason,
"leakage": leakage,
"artifact_refs": refs,
"final_submission_status": projection.get("final_submission_status", "unknown"),
"final_submission_reason": projection.get("final_submission_reason", ""),
"any_of_status": projection.get("any_of_status", "unknown"),
"any_of_reason": projection.get("any_of_reason", ""),
"task_contract": contract,
}
)
return row
@ -1047,6 +1185,29 @@ def append_cybergym_result(run_root: pathlib.Path | str, row: Mapping[str, Any])
append_result_index(safe_task_path(root, task), value)
def _task_spec(value: TaskSpec | Mapping[str, Any] | str) -> TaskSpec:
"""Normalize one injected task value without touching the filesystem."""
if isinstance(value, TaskSpec):
task_id = safe_task_id(value.task_id)
if value.level != DEFAULT_LEVEL:
raise ValueError("CyberGym task contract requires level1")
return dataclasses.replace(value, task_id=task_id)
if isinstance(value, Mapping):
task_id = safe_task_id(str(value.get("task_id", value.get("id", ""))))
level = str(value.get("level") or DEFAULT_LEVEL)
if level != DEFAULT_LEVEL:
raise ValueError("CyberGym task contract requires level1")
metadata = dict(value)
return TaskSpec(
task_id,
str(value.get("project") or task_id.split(":", 1)[0]),
level,
metadata,
)
task_id = safe_task_id(str(value))
return TaskSpec(task_id, task_id.split(":", 1)[0])
def run_campaign(
tasks: Sequence[TaskSpec | Mapping[str, Any] | str],
*,
@ -1062,24 +1223,20 @@ def run_campaign(
seam never falls back to Docker, a shell, or a host network.
"""
root = pathlib.Path(run_root).expanduser().resolve(strict=False)
normalized_tasks: list[TaskSpec] = []
seen_task_ids: set[str] = set()
for item in tasks:
task = _task_spec(item)
if task.task_id in seen_task_ids:
raise ValueError(f"duplicate task id: {task.task_id}")
seen_task_ids.add(task.task_id)
normalized_tasks.append(task)
ledger = BudgetLedger(root / "claims.jsonl", cap_usd=budget_cap_usd)
rows: list[dict[str, Any]] = []
for item in tasks:
if isinstance(item, TaskSpec):
task = item
elif isinstance(item, Mapping):
task_id = safe_task_id(str(item.get("task_id", item.get("id", ""))))
task = TaskSpec(
task_id,
str(item.get("project") or task_id.split(":", 1)[0]),
str(item.get("level") or DEFAULT_LEVEL),
item,
)
else:
task_id = safe_task_id(str(item))
task = TaskSpec(task_id, task_id.split(":", 1)[0])
for task in normalized_tasks:
task_dir = safe_task_path(root, task.task_id)
task_dir.mkdir(parents=True, exist_ok=True)
contract = task.metadata.get("task_contract") if isinstance(task.metadata, Mapping) else None
if executor is None:
row = build_task_result_row(
task.task_id,
@ -1089,6 +1246,7 @@ def run_campaign(
infra_reason="executor_not_injected",
artifact_refs={"task_dir": str(task_dir)},
error="CyberGym executor is not configured",
task_contract=contract if isinstance(contract, Mapping) else None,
)
append_cybergym_result(root, row)
rows.append(row)
@ -1101,16 +1259,33 @@ def run_campaign(
if not isinstance(result, Mapping):
raise CyberGymIntegrationUnavailable("CyberGym executor must return a mapping")
outcome = dict(result)
requested_status = str(outcome.get("status") or "completed").strip().lower()
final_poc = outcome.get("final_poc")
marker = task_dir / FINAL_POC_BASENAME
if final_poc is None and marker.exists():
final_poc = marker
marker_record: FinalPoc | None = None
if requested_status == "completed":
marker_record = final_poc_record(task_dir)
declared_hash = str(outcome.get("final_poc_sha256") or "").strip().lower()
if declared_hash and declared_hash != marker_record.sha256:
raise FinalPocRefused("executor final_poc_sha256 does not match final.poc")
if final_poc is not None:
if isinstance(final_poc, FinalPoc):
supplied_hash = final_poc.sha256
elif isinstance(final_poc, Mapping):
supplied_hash = str(final_poc.get("sha256", final_poc.get("poc_hash", "")))
else:
supplied_hash = final_poc_hash(final_poc)
if supplied_hash and supplied_hash.strip().lower() != marker_record.sha256:
raise FinalPocRefused("executor final_poc does not match final.poc")
final_poc = marker_record
elif final_poc is None and (task_dir / FINAL_POC_BASENAME).exists():
marker_record = final_poc_record(task_dir)
final_poc = marker_record
row = build_task_result_row(
task.task_id,
trials=outcome.get("trials") or (),
final_trial=outcome.get("final_trial"),
final_poc=final_poc,
status=str(outcome.get("status") or "completed"),
status=requested_status,
lifecycle=str(outcome.get("lifecycle") or "completed"),
level=task.level,
observed_provider=str(outcome.get("observed_provider") or ""),
@ -1125,6 +1300,7 @@ def run_campaign(
artifact_refs=outcome.get("artifact_refs") or {"task_dir": str(task_dir)},
error=str(outcome.get("error") or ""),
runtime_result=outcome.get("runtime_result"),
task_contract=contract if isinstance(contract, Mapping) else None,
)
if outcome.get("cost_usd") is None:
ledger.mark_unresolved(str(claim["attempt_id"]), outcome.get("cost_upper_bound_usd"))
@ -1144,6 +1320,7 @@ def run_campaign(
infra_reason=type(exc).__name__,
artifact_refs={"task_dir": str(task_dir), "claims": str(ledger.path)},
error=str(exc),
task_contract=contract if isinstance(contract, Mapping) else None,
)
append_cybergym_result(root, row)
rows.append(row)

107
devtools/benchmarks/cybergym/run_cybergym.py
View file

@ -34,8 +34,11 @@ from devtools.benchmarks.common.run_roots import (
from devtools.benchmarks.cybergym.cybergym_adapter import (
BENCHMARK_NAME,
DEFAULT_BUDGET_CAP_USD,
DEFAULT_DISABLED_TOOLS,
DEFAULT_FINAL_POC_PATH,
DEFAULT_LEVEL,
OFFICIAL_DATA_REVISION,
OFFICIAL_MODEL,
OFFICIAL_SOURCE_PIN,
OFFICIAL_TASKS_SHA256,
BudgetLedger,
@ -50,6 +53,9 @@ from devtools.benchmarks.cybergym.cybergym_adapter import (
pre_admission_report,
run_campaign,
safe_task_id,
task_contract_metadata,
validate_model_pin,
validate_positive_finite,
)
REPO_ROOT = pathlib.Path(__file__).resolve().parents[3]
@ -134,11 +140,36 @@ def _load_executor(spec: str) -> Callable[[TaskSpec, pathlib.Path], Mapping[str,
return callback
def _task_specs(task_ids: Sequence[str], *, level: str = DEFAULT_LEVEL) -> list[TaskSpec]:
return [TaskSpec(task_id, task_id.split(":", 1)[0], level) for task_id in task_ids]
def _task_specs(
task_ids: Sequence[str],
*,
level: str = DEFAULT_LEVEL,
contract: Mapping[str, Any] | None = None,
) -> list[TaskSpec]:
"""Build task values carrying the immutable contract into the executor seam."""
base_contract = dict(contract or task_contract_metadata(level=level))
specs: list[TaskSpec] = []
for raw_task_id in task_ids:
task_id = safe_task_id(raw_task_id)
task_contract = {**base_contract, "task_id": task_id}
specs.append(
TaskSpec(
task_id,
task_id.split(":", 1)[0],
level,
{"task_contract": task_contract},
)
)
return specs
def _write_planned_rows(out_root: pathlib.Path, task_ids: Sequence[str], *, level: str) -> list[dict[str, Any]]:
def _write_planned_rows(
out_root: pathlib.Path,
task_ids: Sequence[str],
*,
level: str,
contract: Mapping[str, Any] | None = None,
) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
for task_id in task_ids:
row = build_task_result_row(
@ -149,6 +180,7 @@ def _write_planned_rows(out_root: pathlib.Path, task_ids: Sequence[str], *, leve
masked_id=mask_task_id(task_id),
infra_reason="dry_run",
artifact_refs={"run_root": str(out_root)},
task_contract=contract,
)
append_cybergym_result(out_root, row)
rows.append(row)
@ -174,10 +206,12 @@ def _prepare_applied_settings(
) from exc
if not isinstance(template, dict):
raise CyberGymIntegrationUnavailable("settings template must contain a JSON object")
from devtools.benchmarks.common.manifests import write_json
from devtools.benchmarks.common.manifests import model_slot_snapshot, write_json
from devtools.benchmarks.common.server_runner import build_isolated_settings
model = str(args.model or "").strip()
model = validate_model_pin(args.model, expected=OFFICIAL_MODEL)
budget_usd = validate_positive_finite(args.budget_usd, field="budget_usd")
timeout_sec = validate_positive_finite(args.timeout_sec, field="timeout_sec")
overrides: dict[str, Any] = {
"OUROBOROS_MODEL": model,
"OUROBOROS_MODEL_LIGHT": model,
@ -185,12 +219,13 @@ def _prepare_applied_settings(
"OUROBOROS_MODEL_CONSCIOUSNESS": model,
"OUROBOROS_MODEL_FALLBACKS": model,
"OUROBOROS_MODEL_DEEP_SELF_REVIEW": model,
"OUROBOROS_WEBSEARCH_MODEL": model,
"OUROBOROS_SCOPE_REVIEW_MODELS": model,
"OUROBOROS_SCOPE_REVIEW_MODEL": model,
"OUROBOROS_REVIEW_MODELS": ",".join([model] * 3),
"OUROBOROS_MAX_SUBAGENT_DEPTH": 0,
"TOTAL_BUDGET": float(args.budget_usd),
"OUROBOROS_TASK_ABS_CEILING_SEC": int(float(args.timeout_sec)),
"TOTAL_BUDGET": budget_usd,
"OUROBOROS_TASK_ABS_CEILING_SEC": max(1, int(timeout_sec)),
}
# The provider pool is selected by a live probe in the sidecar lane. Keep
# the template's safe fallback-ready shape until that callback supplies an
@ -200,12 +235,32 @@ def _prepare_applied_settings(
applied = build_isolated_settings(template, **overrides)
output_path = out_root / "settings_applied.json"
write_json(output_path, applied)
model_slots = model_slot_snapshot(output_path, env_overrides=False)
applied_model = str(model_slots.get("OUROBOROS_MODEL") or "").strip()
if applied_model != model:
raise CyberGymIntegrationUnavailable(
"applied settings changed the pinned model: "
f"expected {model!r}, got {applied_model!r}"
)
model_mismatches = []
for key, value in model_slots.items():
if "MODEL" not in key or not value:
continue
configured = [item.strip() for item in str(value).split(",") if item.strip()]
if any(item != model for item in configured):
model_mismatches.append(key)
if model_mismatches:
raise CyberGymIntegrationUnavailable(
"applied settings contain non-pinned model slots: " + ", ".join(model_mismatches)
)
return output_path, {
"path": str(output_path),
"template_path": str(template_path),
"model": model,
"budget_usd": float(args.budget_usd),
"task_abs_ceiling_sec": int(float(args.timeout_sec)),
"requested_model": model,
"model": applied_model,
"model_slots": model_slots,
"budget_usd": budget_usd,
"task_abs_ceiling_sec": max(1, int(timeout_sec)),
"provider_policy": provider,
"provider_probe_required": True,
"keys": sorted(str(key) for key in applied if isinstance(key, str)),
@ -218,6 +273,17 @@ def main(argv: Sequence[str] | None = None) -> int:
# Everything through this point is argument/path arithmetic. In particular,
# do not read tasks.json, inspect Docker, import the optional upstream package,
# or create a directory before the shared admission manifest exists.
try:
args.model = validate_model_pin(args.model, expected=OFFICIAL_MODEL)
args.budget_usd = validate_positive_finite(args.budget_usd, field="budget_usd")
args.timeout_sec = validate_positive_finite(args.timeout_sec, field="timeout_sec")
if args.per_task_estimate_usd is not None:
args.per_task_estimate_usd = validate_positive_finite(
args.per_task_estimate_usd, field="per_task_estimate_usd"
)
except ValueError as exc:
print(f"[cybergym] pre-admission refusal: {exc}", file=sys.stderr)
return 2
try:
declared_ids = _declared_task_ids(args)
except ValueError as exc:
@ -335,12 +401,29 @@ def main(argv: Sequence[str] | None = None) -> int:
manifest["requested_task_ids"] = task_ids
manifest["requested_count"] = len(task_ids)
contract = task_contract_metadata(
model=args.model,
level=DEFAULT_LEVEL,
source_pin=OFFICIAL_SOURCE_PIN,
data_revision=OFFICIAL_DATA_REVISION,
tasks_sha256=str(args.expected_tasks_sha256 or OFFICIAL_TASKS_SHA256),
final_poc_path=DEFAULT_FINAL_POC_PATH,
disabled_tools=DEFAULT_DISABLED_TOOLS,
)
manifest.setdefault("extra", {})["task_contract"] = contract
applied_path, applied_metadata = _prepare_applied_settings(settings_path, out_root, args)
manifest.setdefault("extra", {})["settings_snapshot"] = applied_metadata
manifest.setdefault("output_paths", {})["settings_applied"] = str(applied_path)
manifest["model_slots"] = dict(applied_metadata.get("model_slots") or {})
manifest.setdefault("harness", {})["applied_model"] = str(
applied_metadata.get("model") or ""
)
if args.dry_run:
rows = _write_planned_rows(out_root, task_ids, level=DEFAULT_LEVEL)
rows = _write_planned_rows(
out_root, task_ids, level=DEFAULT_LEVEL, contract=contract
)
else:
if args.per_task_estimate_usd is None:
raise CyberGymIntegrationUnavailable(
@ -348,7 +431,7 @@ def main(argv: Sequence[str] | None = None) -> int:
)
executor = _load_executor(args.executor)
rows = run_campaign(
_task_specs(task_ids),
_task_specs(task_ids, contract=contract),
run_root=out_root,
executor=executor,
estimated_cost_usd=float(args.per_task_estimate_usd),

116
tests/test_cybergym_protocol.py
View file

@ -7,6 +7,8 @@ import hashlib
import pytest
from devtools.benchmarks.cybergym.cybergym_adapter import (
DEFAULT_FINAL_POC_PATH,
OFFICIAL_MODEL,
BudgetLedger,
BudgetRefused,
ClaimRefused,
@ -17,10 +19,15 @@ from devtools.benchmarks.cybergym.cybergym_adapter import (
classify_official_exit,
final_poc_record,
final_submission,
parse_strict_bool,
pre_admission_report,
project_budget,
run_campaign,
safe_task_id,
safe_task_path,
task_contract_metadata,
validate_model_pin,
validate_positive_finite,
)
@ -91,7 +98,10 @@ def test_result_row_preserves_final_and_any_of_columns():
digest = "a" * 64
row = build_task_result_row(
"arvo:1",
trials=[{"trial_id": "old", "poc_hash": digest, "vul_exit_code": 1, "fix_exit_code": 0}],
trials=[
{"trial_id": "old", "poc_hash": digest, "vul_exit_code": 1, "fix_exit_code": 0},
{"trial_id": "final", "poc_hash": "b" * 64, "vul_exit_code": 71, "fix_exit_code": 0},
],
final_trial={"trial_id": "final", "poc_hash": "b" * 64, "vul_exit_code": 71, "fix_exit_code": 0},
final_poc_sha256="b" * 64,
status="completed",
@ -125,3 +135,107 @@ def test_budget_projection_replays_terminal_states():
assert projection.can_dispatch
with pytest.raises(LedgerError):
project_budget([{"event": "settle", "attempt_id": "orphan", "cost_usd": 1}])
def test_exact_model_and_positive_launcher_values_are_strict():
assert validate_model_pin(OFFICIAL_MODEL) == OFFICIAL_MODEL
with pytest.raises(ValueError):
validate_model_pin("deepseek/deepseek-v4-flash")
assert validate_positive_finite("1.5", field="budget") == 1.5
for value in (0, -1, float("nan"), float("inf"), True, ""):
with pytest.raises(ValueError):
validate_positive_finite(value, field="timeout")
def test_strict_trial_bool_rejects_truthy_strings_and_contract_is_pinned():
assert parse_strict_bool("false") is False
assert parse_strict_bool("TRUE") is True
with pytest.raises(ValueError):
parse_strict_bool("yes")
contract = task_contract_metadata("arvo:1")
assert contract["model"] == OFFICIAL_MODEL
assert contract["final_poc_path"] == DEFAULT_FINAL_POC_PATH
assert contract["no_swarm"] is True
assert "schedule_subagent" in contract["disabled_tools"]
def test_completed_row_requires_marker_bound_final_evidence():
row = build_task_result_row(
"arvo:1",
trials=[{"trial_id": "final", "poc_hash": "a" * 64, "vul_exit_code": 1, "fix_exit_code": 0}],
status="completed",
)
assert row["status"] == "infra_failed"
assert row["infra_reason"] == "final_evidence_missing"
def test_run_campaign_rejects_duplicate_ids_before_creating_output(tmp_path):
with pytest.raises(ValueError, match="duplicate task id"):
run_campaign(
["arvo:1", "arvo:1"],
run_root=tmp_path / "run",
executor=None,
estimated_cost_usd=1,
)
assert not (tmp_path / "run").exists()
def test_run_campaign_requires_regular_marker_and_binds_hash(tmp_path):
def no_marker(_task, _task_dir):
return {
"status": "completed",
"trials": [{"trial_id": "final", "poc_hash": "a" * 64, "vul_exit_code": 1, "fix_exit_code": 0}],
"cost_usd": 0.5,
}
rows = run_campaign(
["arvo:1"],
run_root=tmp_path / "missing",
executor=no_marker,
estimated_cost_usd=1,
budget_cap_usd=2,
)
assert rows[0]["status"] == "infra_failed"
assert rows[0]["infra_reason"] == "FinalPocRefused"
def good_marker(_task, task_dir):
marker = task_dir / "final.poc"
marker.write_bytes(b"poc")
digest = hashlib.sha256(b"poc").hexdigest()
return {
"status": "completed",
"trials": [{"trial_id": "final", "is_final": True, "poc_hash": digest, "vul_exit_code": 1, "fix_exit_code": 0}],
"cost_usd": 0.5,
}
rows = run_campaign(
["arvo:2"],
run_root=tmp_path / "good",
executor=good_marker,
estimated_cost_usd=1,
budget_cap_usd=2,
)
assert rows[0]["status"] == "completed"
assert rows[0]["final_submission_success"] is True
def test_applied_settings_metadata_is_read_back_from_written_snapshot(tmp_path):
from types import SimpleNamespace
from devtools.benchmarks.cybergym.run_cybergym import _prepare_applied_settings
template = tmp_path / "settings.json"
template.write_text(
'{"OUROBOROS_MODEL": "wrong", "OUROBOROS_MODEL_LIGHT": "wrong"}',
encoding="utf-8",
)
output_root = tmp_path / "run"
output_root.mkdir()
path, metadata = _prepare_applied_settings(
template,
output_root,
SimpleNamespace(model=OFFICIAL_MODEL, budget_usd=3, timeout_sec=4),
)
assert path.exists()
assert metadata["model"] == OFFICIAL_MODEL
assert metadata["model_slots"]["OUROBOROS_MODEL"] == OFFICIAL_MODEL