mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-17 20:19:29 +00:00
Marius CiepluchandClaude Opus 5
84b1ba7a72
`publish` keyed baselines by model alone, so the vibe-py run silently replaced the agent-vfc run and the table showed hand-written teaching code as if it were the corpus. That is precisely the failure the constrained-detector requirements exist to prevent, live in a published artifact. Third time today that a key was fixed in one place and not another -- `compare_baselines` got the model+slice label and `_baselines` did not, the same way the exception path got checkpointing and the timeout path did not. The table now carries a slice column and one row per slice, and the artifact is keyed by what identifies a run. Republished: agent-vfc's six families are back, including access_control at 0.000 recall on four pairs, which is the number that was missing. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG
264 lines
12 KiB
Python
264 lines
12 KiB
Python
"""Every published number, regenerated from the artifacts that produced it (learning-harness, Req 11.1, 11.2, 4.5, 5.3).
|
|
|
|
A number in a roadmap that nobody can regenerate is a claim, not a measurement. This writes them all
|
|
from one entry point, out of the artifacts committed beside them, and puts the denominator next to each:
|
|
how many pairs could be scored, why the rest could not, which taxonomy version scored them, the noise
|
|
the family shows against itself, and what the run cost per pair it got right.
|
|
|
|
Running it twice without new artifacts changes nothing, and the roadmap section is replaced in place
|
|
rather than appended, so the file cannot accumulate stale numbers beside fresh ones.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
BEGIN = "<!-- learning-harness:begin -->"
|
|
END = "<!-- learning-harness:end -->"
|
|
COMMAND = "uv run ousast learning publish"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Baseline:
|
|
"""One model's round-zero artifact, read back."""
|
|
|
|
model: str
|
|
slices: tuple[str, ...] = ()
|
|
metrics: dict[str, Mapping[str, object]] = field(default_factory=dict)
|
|
floors: dict[str, Mapping[str, object]] = field(default_factory=dict)
|
|
cost_usd: float = 0.0
|
|
k_runs: int = 0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PublishReport:
|
|
command: str
|
|
table: str
|
|
families: tuple[str, ...] = ()
|
|
models: tuple[str, ...] = ()
|
|
artifacts: tuple[Path, ...] = field(default_factory=tuple)
|
|
|
|
|
|
def publish(*, learning_dir: Path, measurements_dir: Path, roadmap: Path, date: str | None = None) -> PublishReport:
|
|
"""Regenerate the family numbers, write the artifact, and replace the roadmap section in place."""
|
|
stamp = date or datetime.now(UTC).strftime("%Y-%m-%d")
|
|
models = _baselines(learning_dir)
|
|
families = tuple(sorted({family for baseline in models.values() for family in baseline.metrics}))
|
|
lever = _lever(measurements_dir)
|
|
table = _table(models, families, lever)
|
|
artifacts: list[Path] = []
|
|
if models:
|
|
measurements_dir.mkdir(parents=True, exist_ok=True)
|
|
artifact = measurements_dir / f"{stamp}-learning-families.json"
|
|
artifact.write_text(
|
|
json.dumps(
|
|
{
|
|
"generated": stamp,
|
|
"command": COMMAND,
|
|
"models": {name: dict(baseline.metrics) for name, baseline in sorted(models.items())},
|
|
"floors": {name: dict(baseline.floors) for name, baseline in sorted(models.items())},
|
|
},
|
|
indent=2,
|
|
sort_keys=True,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
artifacts.append(artifact)
|
|
_replace_section(roadmap, table)
|
|
_supersede(roadmap)
|
|
return PublishReport(command=COMMAND, table=table, families=families, models=tuple(sorted(models)), artifacts=tuple(artifacts))
|
|
|
|
|
|
def _label(model: str, slices: tuple[str, ...]) -> str:
|
|
"""What identifies a run in the published table: the model, and the slice when it is not the default pair."""
|
|
return f"{model} ({', '.join(slices)})" if slices else model
|
|
|
|
|
|
def _baselines(learning_dir: Path) -> dict[str, Baseline]:
|
|
root = learning_dir / "baseline"
|
|
found: dict[str, Baseline] = {}
|
|
for directory in sorted(path for path in root.iterdir() if path.is_dir()) if root.is_dir() else []:
|
|
report = directory / "report.json"
|
|
if not report.is_file():
|
|
continue
|
|
payload = json.loads(report.read_text(encoding="utf-8"))
|
|
# Keyed by model *and* slice. Keyed by model alone, a second slice silently replaced the first and the
|
|
# published table showed one of them as if it were the corpus (Req 3.1, 3.2).
|
|
model = str(payload.get("model", directory.name))
|
|
slices = tuple(str(item) for item in (payload.get("slices") or ()))
|
|
found[_label(model, slices)] = Baseline(
|
|
model=model,
|
|
slices=slices,
|
|
metrics={str(name): block for name, block in (payload.get("metrics") or {}).items() if isinstance(block, Mapping)},
|
|
floors={str(name): block for name, block in (payload.get("floors") or {}).items() if isinstance(block, Mapping)},
|
|
cost_usd=float(payload.get("cost_usd", 0.0) or 0.0),
|
|
k_runs=int(payload.get("k_runs", 0) or 0),
|
|
)
|
|
return found
|
|
|
|
|
|
def _lever(measurements_dir: Path) -> Mapping[str, object] | None:
|
|
path = measurements_dir / "2026-09-06-mechanism-lever-split.json"
|
|
if not path.is_file():
|
|
return None
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
return payload if isinstance(payload, dict) else None
|
|
|
|
|
|
def _table(models: Mapping[str, Baseline], families: tuple[str, ...], lever: Mapping[str, object] | None) -> str:
|
|
if not models:
|
|
return (
|
|
f"Generated by `{COMMAND}`.\n\nNo baseline has been recorded yet, so there is nothing to publish. "
|
|
"Run round zero first; a number without the run that produced it is a claim, not a measurement.\n"
|
|
)
|
|
lines = [
|
|
f"Generated by `{COMMAND}` from the artifacts committed beside these numbers.",
|
|
"",
|
|
(
|
|
"| model | slice | family | scorable | unscorable | recall | silence | Youden "
|
|
"| at fixed FPR | noise floor | cost per correct pair |"
|
|
),
|
|
"|---|---|---|---|---|---|---|---|---|---|---|",
|
|
]
|
|
for _label_key, baseline in sorted(models.items()):
|
|
model = baseline.model
|
|
slice_name = ", ".join(baseline.slices) or "all"
|
|
for family in families:
|
|
row = baseline.metrics.get(family)
|
|
if row is None:
|
|
continue
|
|
outcomes = row.get("outcomes")
|
|
raw_correct = outcomes.get("pair_correct") if isinstance(outcomes, Mapping) else None
|
|
correct = int(raw_correct) if isinstance(raw_correct, int | float) else 0
|
|
floor = baseline.floors.get(family)
|
|
raw_budget = floor.get("budget_flips") if floor is not None else None
|
|
budget = int(raw_budget) if isinstance(raw_budget, int | float) else 0
|
|
cost = baseline.cost_usd
|
|
lines.append(
|
|
f"| {model} | {slice_name} | {family} | {row.get('scorable', 0)} (taxonomy {row.get('taxonomy_version', '')}) "
|
|
f"| {_reasons(row)} | {_num(row.get('recall'))} | {_num(row.get('silence'))} | {_num(row.get('youden'))} "
|
|
f"| {_num(row.get('fixed_fpr_recall'))} | budget {budget} | {_cost(cost, correct)} |"
|
|
)
|
|
lines += ["", _credit_note(models), ""]
|
|
cutoff = _cutoff_note(models, families)
|
|
if cutoff:
|
|
lines += [cutoff, ""]
|
|
if lever is not None:
|
|
lines += [_lever_note(lever), ""]
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _cutoff_note(models: Mapping[str, Baseline], families: tuple[str, ...]) -> str:
|
|
"""The post-cutoff slice, its cutoff and the undated count (Req 10.5).
|
|
|
|
A model that may have read the fix cannot be measured on it the same way as one that cannot have, so the rows
|
|
fixed after the detector's training cutoff are reported apart. Rows the corpus records no date for are in
|
|
neither slice, and their number is here rather than folded silently into the whole."""
|
|
rows = [
|
|
(model, family, baseline.metrics[family])
|
|
for model, baseline in sorted(models.items())
|
|
for family in families
|
|
if family in baseline.metrics and str(baseline.metrics[family].get("cutoff") or "")
|
|
]
|
|
if not rows:
|
|
return "No detector cutoff is configured, so no post-cutoff slice is reported; every figure above is over the whole corpus."
|
|
cutoff = str(rows[0][2].get("cutoff"))
|
|
lines = [
|
|
f"Post-cutoff slice, fixed after {cutoff} and therefore beyond the detector's training data. "
|
|
"A row counts only when its earliest possible date is after the cutoff, so a year-precision date can "
|
|
"under-claim this slice but never over-claim it.",
|
|
"",
|
|
"| model | family | post-cutoff scorable | post-cutoff recall | post-cutoff Youden | undated |",
|
|
"|---|---|---|---|---|---|",
|
|
]
|
|
for model, family, row in rows:
|
|
block = row.get("post_cutoff")
|
|
block = block if isinstance(block, Mapping) else {}
|
|
lines.append(
|
|
f"| {model} | {family} | {block.get('scorable', 0)} | {_num(block.get('recall'))} "
|
|
f"| {_num(block.get('youden'))} | {row.get('undated', 0)} |"
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _credit_note(models: Mapping[str, Baseline]) -> str:
|
|
hierarchical = any(bool(row.get("hierarchical_credit")) for baseline in models.values() for row in baseline.metrics.values())
|
|
if hierarchical:
|
|
return "A parent-child answer earned partial credit in these figures."
|
|
return "No family in this taxonomy declares a parent, so hierarchical partial credit could not apply to any figure above."
|
|
|
|
|
|
def _lever_note(lever: Mapping[str, object]) -> str:
|
|
before = lever.get("before")
|
|
after = lever.get("after")
|
|
before_map = before if isinstance(before, Mapping) else {}
|
|
after_map = after if isinstance(after, Mapping) else {}
|
|
pairs = lever.get("holdout_pairs", 0)
|
|
return (
|
|
f"Mechanism lever on the {lever.get('slice', '')} holdout: detected {after_map.get('detected')} of {pairs} at Youden "
|
|
f"{after_map.get('youden')} once only train-split pairs teach, against detected {before_map.get('detected')} at Youden "
|
|
f"{before_map.get('youden')} when every pair taught. The difference is what holdout pairs stopped teaching."
|
|
)
|
|
|
|
|
|
def _reasons(row: Mapping[str, object]) -> str:
|
|
reasons = row.get("unscorable") or {}
|
|
if not isinstance(reasons, Mapping) or not reasons:
|
|
return "none"
|
|
return ", ".join(f"{name}: {count}" for name, count in sorted(reasons.items()))
|
|
|
|
|
|
def _num(value: object) -> str:
|
|
return f"{float(value):.3f}" if isinstance(value, int | float) else "-"
|
|
|
|
|
|
def _cost(cost: float, correct: int) -> str:
|
|
return f"{cost / correct:.2f}" if cost and correct else "-"
|
|
|
|
|
|
# Claims the split correction supersedes, as they were written before holdout pairs stopped teaching. Req 5.3 says
|
|
# the corrected number is published *in place of* the earlier one; a document that carries both asserts two numbers
|
|
# for one measurement and the reader has no way to tell which run is current.
|
|
_SUPERSEDED = (
|
|
re.compile(r"human holdout profile pair_correct \d+ -> \d+ of \d+, Youden [-+]?\d+ -> [-+]?[\d.]+(?:, round \d+ `[^`]+`)?\."),
|
|
re.compile(r"The lever recovered \d+ of \d+ holdout pairs at Youden [-+][\d.]+\."),
|
|
)
|
|
_SUPERSEDED_NOTE = (
|
|
"superseded by the learning-harness section below: that number was measured with holdout pairs teaching "
|
|
"(see `benchmarks/measurements/2026-09-06-mechanism-lever-split.json`)."
|
|
)
|
|
|
|
|
|
def _supersede(roadmap: Path) -> None:
|
|
"""Rewrite every superseded lever claim to point at the regenerated section. Idempotent: the pattern is gone
|
|
afterwards, so a second run changes nothing."""
|
|
if not roadmap.is_file():
|
|
return
|
|
text = roadmap.read_text(encoding="utf-8")
|
|
updated = text
|
|
for pattern in _SUPERSEDED:
|
|
updated = pattern.sub(_SUPERSEDED_NOTE, updated)
|
|
if updated != text:
|
|
roadmap.write_text(updated, encoding="utf-8")
|
|
|
|
|
|
def _replace_section(roadmap: Path, table: str) -> None:
|
|
body = f"{BEGIN}\n\n## Classified detector numbers\n\n{table.rstrip()}\n\n{END}\n"
|
|
text = roadmap.read_text(encoding="utf-8") if roadmap.is_file() else ""
|
|
if BEGIN in text and END in text:
|
|
head, _, rest = text.partition(BEGIN)
|
|
_, _, tail = rest.partition(END)
|
|
roadmap.write_text(head + body + tail.lstrip("\n"), encoding="utf-8")
|
|
return
|
|
roadmap.parent.mkdir(parents=True, exist_ok=True)
|
|
roadmap.write_text((text.rstrip("\n") + "\n\n" if text else "") + body, encoding="utf-8")
|
|
|
|
|
|
__all__ = ["BEGIN", "COMMAND", "END", "Baseline", "PublishReport", "publish"]
|