feat(results): report coverage, judge agreement, and run limits

assert_ai/analysis/stats.py

@@ -33,6 +33,83 @@ def _wilson_ci(k: int, n: int, alpha: float = 0.10) -> tuple[float, float]:
     return (max(0.0, center - spread), min(1.0, center + spread))
 
 
+# Landis & Koch (1977) benchmarks. Below this, agreement is weak enough that a
+# consensus verdict should not be read as a reliable one.
+KAPPA_WARN_THRESHOLD = 0.60
+
+
+def fleiss_kappa(ratings: list[list[Any]]) -> float | None:
+    """Chance-corrected inter-rater agreement across a fixed number of raters.
+
+    ``ratings`` is one list of votes per item, each containing one vote per
+    judge. Votes may be any hashable label; ``None`` is a category like any
+    other, so a judge marking a dimension not-applicable is a real position
+    rather than a missing value.
+
+    Raw percent agreement is not a substitute for this. With a skewed base rate
+    - and violation rates usually are skewed - two judges voting independently
+    agree most of the time by chance alone, so a high raw figure can describe
+    almost no real reliability. Kappa subtracts that expected agreement.
+
+    Returns ``None`` when kappa is undefined: fewer than two items, fewer than
+    two raters, or a ragged number of raters across items. Returns ``1.0`` when
+    every rater agrees on every item, including the degenerate case where only
+    one category was ever used and expected agreement is also 1.
+    """
+    if len(ratings) < 1:
+        return None
+    n_raters = len(ratings[0])
+    if n_raters < 2:
+        return None
+    if any(len(item) != n_raters for item in ratings):
+        return None
+
+    categories = sorted({_kappa_label(vote) for item in ratings for vote in item})
+    if not categories:
+        return None
+
+    n_items = len(ratings)
+    counts: list[list[int]] = []
+    for item in ratings:
+        row = {category: 0 for category in categories}
+        for vote in item:
+            row[_kappa_label(vote)] += 1
+        counts.append([row[category] for category in categories])
+
+    # Observed agreement: mean over items of the proportion of rater pairs that
+    # agree.
+    p_item = [
+        (sum(count * count for count in row) - n_raters) / (n_raters * (n_raters - 1))
+        for row in counts
+    ]
+    p_observed = sum(p_item) / n_items
+
+    # Expected agreement from the marginal distribution of categories.
+    total_ratings = n_items * n_raters
+    p_category = [
+        sum(row[index] for row in counts) / total_ratings
+        for index in range(len(categories))
+    ]
+    p_expected = sum(p * p for p in p_category)
+
+    denominator = 1.0 - p_expected
+    if denominator <= 1e-12:
+        # Only one category was used anywhere, so chance agreement is already
+        # total and kappa is 0/0. Every rater did agree on every item, which is
+        # the sense in which this is 1.0 rather than undefined.
+        return 1.0
+    return (p_observed - p_expected) / denominator
+
+
+def _kappa_label(vote: Any) -> str:
+    """Map a vote to a stable category label, keeping None a real category."""
+    if vote is None:
+        return "\x00none"
+    if isinstance(vote, bool):
+        return f"bool:{vote}"
+    return f"{type(vote).__name__}:{vote}"
+
+
 def binary_rate_ci(
     outcomes: list[bool],
     *,

assert_ai/config.py

@@ -29,6 +29,7 @@
     ModelConfig,
     PipelineConfig,
     InferenceConfig,
+    RunLimits,
     TargetConfig,
     ToolsConfig,
     TraceConfig,
@@ -184,6 +185,14 @@ def load_runtime_context(
             "artifacts_root",
             "results_dir",
             "pipeline",
+            # DO NOT REMOVE "limits" from this set, even if the enforcement in
+            # UsageAccumulator is reverted. Unknown top-level keys are rejected
+            # outright, so once a user has written a limits: block, dropping the
+            # key here makes their config fail to load rather than degrade to
+            # the previous unlimited behaviour. If the ceiling logic needs to go,
+            # leave this entry and let parse_run_limits return an inactive
+            # RunLimits.
+            "limits",
         },
     )
     default_model_raw = _get_default_model_mapping(raw)
@@ -302,6 +311,7 @@ def load_runtime_context(
         "stages": stages,
         "target": target,
         "evaluation": pipeline.evaluation if pipeline else None,
+        "limits": parse_run_limits(raw.get("limits")),
     }
 
 
@@ -406,6 +416,52 @@ def reject_unknown_keys(raw: dict[str, Any], *, field_name: str, allowed: set[st
         raise ValueError(f"{field_name} has unsupported field(s): {', '.join(unknown)}")
 
 
+def parse_run_limits(raw: Any, *, field_name: str = "limits") -> RunLimits:
+    """Parse the optional top-level ``limits:`` block.
+
+    A missing or empty block yields an inactive :class:`RunLimits`, so configs
+    written before this existed behave exactly as they did.
+    """
+    if raw is None:
+        return RunLimits()
+    if not isinstance(raw, dict):
+        raise ValueError(f"{field_name} must be a mapping")
+    reject_unknown_keys(
+        raw,
+        field_name=field_name,
+        allowed={"max_total_calls", "max_total_tokens", "max_wall_time_s", "on_exceed"},
+    )
+
+    def _positive_int(key: str) -> int | None:
+        value = raw.get(key)
+        if value is None:
+            return None
+        if isinstance(value, bool) or not isinstance(value, int):
+            raise ValueError(f"{field_name}.{key} must be a positive integer")
+        if value <= 0:
+            raise ValueError(f"{field_name}.{key} must be a positive integer")
+        return value
+
+    max_wall_time_s = raw.get("max_wall_time_s")
+    if max_wall_time_s is not None:
+        if isinstance(max_wall_time_s, bool) or not isinstance(max_wall_time_s, (int, float)):
+            raise ValueError(f"{field_name}.max_wall_time_s must be a positive number")
+        if max_wall_time_s <= 0:
+            raise ValueError(f"{field_name}.max_wall_time_s must be a positive number")
+        max_wall_time_s = float(max_wall_time_s)
+
+    on_exceed = raw.get("on_exceed", "stop")
+    if on_exceed not in ("stop", "warn"):
+        raise ValueError(f"{field_name}.on_exceed must be 'stop' or 'warn'")
+
+    return RunLimits(
+        max_total_calls=_positive_int("max_total_calls"),
+        max_total_tokens=_positive_int("max_total_tokens"),
+        max_wall_time_s=max_wall_time_s,
+        on_exceed=on_exceed,
+    )
+
+
 def parse_model_config(
     raw: Any,
     *,

assert_ai/core/config_model.py

@@ -253,6 +253,39 @@ class PipelineConfig:
     evaluation: EvaluationConfig | None = None
 
 
+@dataclass
+class RunLimits:
+    """Whole-run consumption ceilings.
+
+    Every other limit in ASSERT is per-call or per-task - max_tool_calls,
+    max_turns, model timeout - so nothing bounds a run as a whole. The realistic
+    failure is a typo: ``sample_size: 5000`` against an expensive judge, with
+    nothing able to stop it once it starts.
+
+    All fields default to None, meaning unlimited, so a config without a
+    ``limits:`` block behaves exactly as it did before.
+
+    There is deliberately no cost ceiling. ASSERT carries no pricing table, and
+    a cost limit computed from an invented one would be wrong in whichever
+    direction the operator could least afford. Token and call ceilings are
+    directly measurable and are what is offered instead.
+    """
+
+    max_total_calls: int | None = None
+    max_total_tokens: int | None = None
+    max_wall_time_s: float | None = None
+    on_exceed: str = "stop"
+
+    def is_active(self) -> bool:
+        return any(
+            value is not None
+            for value in (self.max_total_calls, self.max_total_tokens, self.max_wall_time_s)
+        )
+
+    def to_dict(self) -> dict[str, Any]:
+        return {k: v for k, v in asdict(self).items() if v is not None}
+
+
 @dataclass
 class SuiteMetadata:
     created_at: str
@@ -278,6 +311,14 @@ class RunManifest:
     progress: dict[str, Any] | None = None
     artifact_versions: dict[str, dict[str, Any]] = field(default_factory=dict)
     stage_timings: dict[str, dict[str, Any]] = field(default_factory=dict)
+    # Set when metrics.json could not be written. Without this a failed write
+    # loses the entire cost record while the run still reports success, so the
+    # gap has to be recorded somewhere durable rather than only in the log.
+    metrics_write_failed: bool | None = None
+    metrics_write_error: str | None = None
+    # Set when a configured run limit stopped the pipeline, so the run is
+    # distinguishable from one that failed on an error.
+    stopped_by_limit: str | None = None
 
     def to_dict(self) -> dict[str, Any]:
         empty_collection_keys = {"artifact_versions", "stage_timings"}

assert_ai/core/model_client.py

@@ -161,6 +161,16 @@ class UsageAccumulator:
     cached_input_tokens: int = 0
     cache_creation_input_tokens: int = 0
     per_model: dict[str, dict[str, int]] = field(default_factory=dict)
+    # Whole-run ceilings, checked after each call. None means unlimited, which
+    # is the behaviour when no limits: block is configured.
+    limits: Any = None
+    # Consumption from earlier stages. track_usage() is entered once per stage,
+    # so without a baseline a run-level ceiling would reset at every stage
+    # boundary and never bind.
+    baseline_calls: int = 0
+    baseline_tokens: int = 0
+    started_at: float = field(default_factory=time.monotonic)
+    _limit_warned: bool = False
 
     def add(self, usage: UsageStats | None, *, model: str | None = None) -> None:
         """Fold one call's normalized usage into this accumulator."""
@@ -191,6 +201,52 @@ def add(self, usage: UsageStats | None, *, model: str | None = None) -> None:
         bucket["output_tokens"] += opt
         bucket["cached_input_tokens"] += cit
         bucket["cache_creation_input_tokens"] += cct
+        self.check_limits()
+
+    def total_tokens(self) -> int:
+        return self.input_tokens + self.output_tokens
+
+    def run_calls(self) -> int:
+        """Calls made across the whole run, including earlier stages."""
+        return self.baseline_calls + self.calls
+
+    def run_tokens(self) -> int:
+        """Tokens used across the whole run, including earlier stages."""
+        return self.baseline_tokens + self.total_tokens()
+
+    def elapsed_s(self) -> float:
+        return time.monotonic() - self.started_at
+
+    def exceeded(self) -> str | None:
+        """Return a description of the first breached limit, or None."""
+        limits = self.limits
+        if limits is None or not getattr(limits, "is_active", lambda: False)():
+            return None
+        max_calls = getattr(limits, "max_total_calls", None)
+        if max_calls is not None and self.run_calls() > max_calls:
+            return f"max_total_calls ({self.run_calls()} > {max_calls})"
+        max_tokens = getattr(limits, "max_total_tokens", None)
+        if max_tokens is not None and self.run_tokens() > max_tokens:
+            return f"max_total_tokens ({self.run_tokens()} > {max_tokens})"
+        max_wall = getattr(limits, "max_wall_time_s", None)
+        if max_wall is not None and self.elapsed_s() > max_wall:
+            return f"max_wall_time_s ({self.elapsed_s():.0f}s > {max_wall:.0f}s)"
+        return None
+
+    def check_limits(self) -> None:
+        """Raise or warn when a configured whole-run ceiling has been passed."""
+        breach = self.exceeded()
+        if breach is None:
+            return
+        if getattr(self.limits, "on_exceed", "stop") == "warn":
+            if not self._limit_warned:
+                self._limit_warned = True
+                log.warning(
+                    "Run limit exceeded: %s. Continuing because on_exceed is 'warn'.",
+                    breach,
+                )
+            return
+        raise BudgetExceededError(f"Run limit exceeded: {breach}")
 
     def cache_hit_rate(self) -> float:
         """Return cached_input_tokens / input_tokens, or 0.0 when no input tokens."""
@@ -218,16 +274,31 @@ def to_dict(self) -> dict[str, Any]:
 
 
 @contextlib.contextmanager
-def track_usage() -> Iterator[UsageAccumulator]:
+def track_usage(
+    limits: Any = None,
+    *,
+    baseline_calls: int = 0,
+    baseline_tokens: int = 0,
+    started_at: float | None = None,
+) -> Iterator[UsageAccumulator]:
     """Capture token usage from every ``generate*`` call within the block.
 
     Uses a ``ContextVar`` so that ``asyncio.run(...)`` blocks invoked inside the
     ``with`` statement inherit the accumulator and concurrent tasks all add into
     the same object. The accumulator only sees calls made on the same ``async``
     stack (or the same thread) — independent threads or coroutines that are
     started in a fresh context will not contribute.
+
+    ``limits`` is an optional :class:`~assert_ai.core.config_model.RunLimits`.
+    When supplied and active, each recorded call is checked against it and
+    :class:`BudgetExceededError` is raised once a ceiling is passed.
     """
-    accumulator = UsageAccumulator()
+    accumulator = UsageAccumulator(
+        limits=limits,
+        baseline_calls=baseline_calls,
+        baseline_tokens=baseline_tokens,
+        started_at=started_at if started_at is not None else time.monotonic(),
+    )
     token = _USAGE_ACCUMULATOR.set(accumulator)
     try:
         yield accumulator
@@ -751,6 +822,15 @@ class LLMProviderError(Exception):
     """Provider-side error (5xx) — may be retryable."""
 
 
+class BudgetExceededError(Exception):
+    """A configured whole-run consumption ceiling was passed.
+
+    Deliberately not an ``LLM*Error``: the provider call succeeded, and this is
+    the harness stopping on the operator's own instruction. The retry and
+    fallback paths must not treat it as a transient provider failure.
+    """
+
+
 class _ResponsesApiNotAvailableError(LLMProviderError):
     """Region does not support Azure Responses API — triggers automatic
     fallback to Chat Completions for the remainder of the run.