From 5440bdf5b321071197d1ae42bfd1dd5db948e50d Mon Sep 17 00:00:00 2001 From: ernestprovo23 Date: Fri, 17 Jul 2026 16:41:00 -0400 Subject: [PATCH 1/7] docs: plan H1 budget-matched evaluation --- ...-17-h1-budget-matched-evaluation-design.md | 49 +++++++++++++++++ ...2026-07-17-h1-budget-matched-evaluation.md | 55 +++++++++++++++++++ 2 files changed, 104 insertions(+) create mode 100644 docs/plans/2026-07-17-h1-budget-matched-evaluation-design.md create mode 100644 docs/plans/2026-07-17-h1-budget-matched-evaluation.md diff --git a/docs/plans/2026-07-17-h1-budget-matched-evaluation-design.md b/docs/plans/2026-07-17-h1-budget-matched-evaluation-design.md new file mode 100644 index 0000000..a69a071 --- /dev/null +++ b/docs/plans/2026-07-17-h1-budget-matched-evaluation-design.md @@ -0,0 +1,49 @@ +# H1 Budget-Matched Evaluation Design + +**Linear:** DSE-708 +**Status:** Approved direction; implementation begins offline +**Base:** Elite H0 commit `23e8fce437172c040084d31dcea4ed5737ebae2a` + +## Decision + +Build a narrow, experimental `conclave.evals` package that can plan, replay, blind, score, and report a fixed six-condition study. It is an evidence instrument for Elite, not a general evaluation platform and not a new public council mode. + +## Study contract + +The six frozen conditions are: + +1. `single_frontier` +2. `self_refine` +3. `independent_synthesis` +4. `critique_only` +5. `revision_only` +6. `elite_full` + +Every task-condition-replicate cell is declared before execution. Missing, timed-out, malformed, abstained, and incomplete cells remain in the denominator. Conditions receive identical public task material and reference packets. Grader-only keys live in a separate file that the runner never loads. + +## Architecture + +- `models.py`: versioned Pydantic task, condition, run, score, and study-manifest contracts. +- `dataset.py`: load and hash public tasks; separately load grader keys only for scoring. +- `protocols.py`: immutable six-condition registry and token-budget allocation. +- `replay.py`: record/replay at `conclave.transport.post_json`; match sanitized request identity plus occurrence index; reject missing, extra, incompatible, or secret-bearing artifacts. +- `runner.py`: create a seeded task x condition x replicate matrix, enforce budgets, preserve failures, and emit atomic artifacts. +- `blinding.py`: deterministic seeded opaque output IDs and a separate restricted blind map. +- `scoring.py`: atomic judgments, adjudication without overwriting raw scores, Wilson intervals, paired bootstrap differences, and Cohen's kappa. +- `eval_cli.py`: `plan`, `run`, `blind`, and `report` commands. The main CLI exposes these under `conclave eval` only after the offline substrate is stable. + +## Budget matching + +The study manifest declares one output-token ceiling per task-condition cell. Provider adapters receive `max_output_tokens` end to end. Planned ceilings must be within 5% across conditions; actual tokens, latency, failures, and unused budget are reported. Token ceilings are a reproducible provider-spend proxy; a later live-study manifest may add frozen price metadata without placing mutable pricing in library code. + +## Replay and security + +Replay performs zero network calls. Request identities include provider-safe URL components, model, normalized non-secret body, and occurrence index; authorization headers, API keys, raw endpoint credentials, and exception chains are never serialized. Replay fails closed on schema/version/hash mismatch or unmatched calls. + +## Analysis + +The primary endpoint is failure-inclusive critical-error-free decision rate. Reports retain task-level paired outcomes and show distributions and uncertainty: Wilson intervals for rates, seeded paired bootstrap intervals for condition differences, and kappa/adjudication rates for graders. Pilot results are exploratory; held-out results require a frozen preregistration. + +## Delivery boundaries + +This increment excludes provider spend, a hosted dashboard, retrieval, embeddings, routing, mutable pricing tables, LLM-as-primary-grader, and public product claims. H0 must be merged and pinned before a confirmatory live run. Any paid pilot requires an explicit spend ceiling. diff --git a/docs/plans/2026-07-17-h1-budget-matched-evaluation.md b/docs/plans/2026-07-17-h1-budget-matched-evaluation.md new file mode 100644 index 0000000..eb2bd6a --- /dev/null +++ b/docs/plans/2026-07-17-h1-budget-matched-evaluation.md @@ -0,0 +1,55 @@ +# H1 Budget-Matched Evaluation Implementation Plan + +> **Linear:** DSE-708. Execute with test-driven development on stacked branch `feat/h1-budget-matched-eval`; make no live provider calls in this plan. + +**Goal:** Deliver an offline, reproducible substrate that can fairly compare Elite with five fixed budget-matched alternatives. + +**Architecture:** Add an experimental `conclave.evals` package around versioned artifacts. Keep provider I/O at the existing transport seam, thread output-token caps through adapters, and keep study execution separate from the public council API. + +## Phase 1 — Contracts and deterministic planning + +1. Add failing tests in `tests/evals/test_models.py` and `tests/evals/test_protocols.py` for schema versions, the exact six-condition registry, ±5% planned-budget enforcement, stable hashes, deterministic seeded order, and complete task x condition x replicate matrices. +2. Run `pytest tests/evals/test_models.py tests/evals/test_protocols.py -q` and confirm RED. +3. Implement `src/conclave/evals/__init__.py`, `models.py`, `dataset.py`, and `protocols.py` minimally. +4. Re-run focused tests, Ruff, and commit. + +## Phase 2 — Output caps and strict replay + +1. Add failing adapter/provider tests proving `max_output_tokens` reaches OpenAI-compatible, Anthropic, and Gemini request bodies without changing default requests. +2. Add failing `tests/evals/test_replay.py` cases for record/replay identity, repeated-call occurrence indexes, zero-network replay, secret exclusion, and fail-closed missing/extra/version-mismatched artifacts. +3. Thread optional `max_output_tokens` through `ProviderAdapter`, concrete adapters, `call_model`, and streaming paths; implement `src/conclave/evals/replay.py` at `transport.post_json`. +4. Run focused tests, existing adapter/provider/transport tests, Ruff, and commit. + +## Phase 3 — Runner and blinding + +1. Add failing `tests/evals/test_runner.py` and `test_blinding.py` for all six conditions, per-cell caps, immutable failures in denominators, receipts/totals, seeded opaque IDs, and a separate blind map. +2. Implement `src/conclave/evals/runner.py`, `protocols.py`, and `blinding.py`. Protocol executors may use recorded fixtures only in this increment. +3. Re-run focused tests and commit. + +## Phase 4 — Atomic scoring and reports + +1. Add failing `tests/evals/test_scoring.py` for raw grader preservation, adjudication, critical-error-free rates, Wilson intervals, seeded paired bootstrap intervals, and Cohen's kappa. +2. Implement `src/conclave/evals/scoring.py` and machine-readable plus Markdown report writers. +3. Add a small synthetic fixture set under `tests/fixtures/evals/`; label every result exploratory/synthetic. +4. Re-run focused tests and commit. + +## Phase 5 — CLI and documentation + +1. Add failing `tests/evals/test_cli.py` for `conclave eval plan`, `run --replay`, `blind`, and `report`; reject live execution unless explicitly enabled and configured. +2. Implement `src/conclave/eval_cli.py` and mount its Typer app in `src/conclave/cli.py`. +3. Update `README.md`, `DOCUMENTATION_INDEX.md`, and `SYSTEM_CONTEXT_DIAGRAM.md` with the experimental boundary and reproducible commands. +4. Run CLI tests and commit. + +## Final verification and handoff + +Run: + +```bash +python -m pytest -q +python -m ruff check . +python -m ruff format --check . +git diff --check +gitleaks git --redact +``` + +Then push the stacked branch, open a PR linked to DSE-708 and PR #51, request independent review, and attach test/scan evidence to Linear. Do not merge, publish, or run paid models without the required approval. From 35b6969b8f5e6a58cf3ad3cebcff2064550e051d Mon Sep 17 00:00:00 2001 From: ernestprovo23 Date: Fri, 17 Jul 2026 16:45:50 -0400 Subject: [PATCH 2/7] feat(evals): add versioned study planning contracts --- src/conclave/evals/__init__.py | 27 +++++++ src/conclave/evals/dataset.py | 61 ++++++++++++++ src/conclave/evals/models.py | 139 ++++++++++++++++++++++++++++++++ src/conclave/evals/protocols.py | 129 +++++++++++++++++++++++++++++ tests/evals/test_models.py | 113 ++++++++++++++++++++++++++ tests/evals/test_protocols.py | 115 ++++++++++++++++++++++++++ 6 files changed, 584 insertions(+) create mode 100644 src/conclave/evals/__init__.py create mode 100644 src/conclave/evals/dataset.py create mode 100644 src/conclave/evals/models.py create mode 100644 src/conclave/evals/protocols.py create mode 100644 tests/evals/test_models.py create mode 100644 tests/evals/test_protocols.py diff --git a/src/conclave/evals/__init__.py b/src/conclave/evals/__init__.py new file mode 100644 index 0000000..2a019f6 --- /dev/null +++ b/src/conclave/evals/__init__.py @@ -0,0 +1,27 @@ +"""Experimental, offline contracts for budget-matched Conclave studies. + +This package is intentionally not part of the public council API. Its schemas +are versioned so recorded studies can be rejected when their contract changes. +""" + +from .models import ( + EVAL_SCHEMA_VERSION, + ConditionSpec, + GraderKey, + PlannedRun, + PublicTask, + RunRecord, + ScoreRecord, + StudyManifest, +) + +__all__ = [ + "EVAL_SCHEMA_VERSION", + "ConditionSpec", + "GraderKey", + "PlannedRun", + "PublicTask", + "RunRecord", + "ScoreRecord", + "StudyManifest", +] diff --git a/src/conclave/evals/dataset.py b/src/conclave/evals/dataset.py new file mode 100644 index 0000000..a773a2c --- /dev/null +++ b/src/conclave/evals/dataset.py @@ -0,0 +1,61 @@ +"""Separate loading and canonical hashing for public tasks and grader keys.""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Iterable +from pathlib import Path + +from .models import GraderKey, GraderKeyDataset, PublicTask, PublicTaskDataset + + +def _canonical_hash(namespace: str, records: Iterable[PublicTask | GraderKey]) -> str: + ordered = sorted(records, key=lambda record: record.task_id) + canonical = json.dumps( + { + "namespace": namespace, + "records": [record.model_dump(mode="json") for record in ordered], + }, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + return f"sha256:{hashlib.sha256(canonical).hexdigest()}" + + +def hash_public_tasks(tasks: Iterable[PublicTask]) -> str: + """Return the stable, order-independent digest of public task records.""" + + return _canonical_hash("public_tasks", tasks) + + +def hash_grader_keys(keys: Iterable[GraderKey]) -> str: + """Return the stable, order-independent digest of grader-only records.""" + + return _canonical_hash("grader_keys", keys) + + +def _load_json(path: str | Path) -> object: + with Path(path).open(encoding="utf-8") as handle: + return json.load(handle) + + +def load_public_tasks(path: str | Path) -> list[PublicTask]: + """Load only a public-task envelope; grader fields are rejected as extras.""" + + dataset = PublicTaskDataset.model_validate(_load_json(path)) + task_ids = [task.task_id for task in dataset.tasks] + if len(set(task_ids)) != len(task_ids): + raise ValueError("public task_id values must be unique") + return list(dataset.tasks) + + +def load_grader_keys(path: str | Path) -> list[GraderKey]: + """Load grader keys independently from the public execution dataset.""" + + dataset = GraderKeyDataset.model_validate(_load_json(path)) + task_ids = [key.task_id for key in dataset.grader_keys] + if len(set(task_ids)) != len(task_ids): + raise ValueError("grader key task_id values must be unique") + return list(dataset.grader_keys) diff --git a/src/conclave/evals/models.py b/src/conclave/evals/models.py new file mode 100644 index 0000000..7d49186 --- /dev/null +++ b/src/conclave/evals/models.py @@ -0,0 +1,139 @@ +"""Versioned, immutable data contracts for the experimental eval harness.""" + +from __future__ import annotations + +from typing import Annotated, Literal + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +EVAL_SCHEMA_VERSION = "conclave_eval_v1" +SchemaVersion = Literal["conclave_eval_v1"] + +ConditionId = Literal[ + "single_frontier", + "self_refine", + "independent_synthesis", + "critique_only", + "revision_only", + "elite_full", +] + +EVAL_CONDITION_IDS: tuple[ConditionId, ...] = ( + "single_frontier", + "self_refine", + "independent_synthesis", + "critique_only", + "revision_only", + "elite_full", +) + +Sha256Digest = Annotated[str, Field(pattern=r"^sha256:[0-9a-f]{64}$")] + + +class EvalModel(BaseModel): + """Base contract that rejects drift and mutation.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + schema_version: SchemaVersion = EVAL_SCHEMA_VERSION + + +class PublicTask(EvalModel): + """Material visible to every condition during study execution.""" + + task_id: str = Field(min_length=1) + prompt: str = Field(min_length=1) + reference_packets: tuple[str, ...] = () + metadata: dict[str, str] = Field(default_factory=dict) + + +class GraderKey(EvalModel): + """Private grading material loaded only by the scoring process.""" + + task_id: str = Field(min_length=1) + required_facts: tuple[str, ...] = () + critical_errors: tuple[str, ...] = () + rubric: dict[str, str] = Field(default_factory=dict) + + +class ConditionSpec(EvalModel): + """One frozen experimental comparison condition.""" + + condition_id: ConditionId + description: str = Field(min_length=1) + + +class PlannedRun(EvalModel): + """An immutable cell declared before any model execution.""" + + planned_run_id: str = Field(pattern=r"^run_[0-9a-f]{24}$") + study_id: str = Field(min_length=1) + task_id: str = Field(min_length=1) + condition_id: ConditionId + replicate: int = Field(ge=1) + max_output_tokens: int = Field(gt=0) + + +class RunRecord(EvalModel): + """Failure-inclusive result for one predeclared cell.""" + + planned_run_id: str = Field(pattern=r"^run_[0-9a-f]{24}$") + outcome: Literal["success", "failed", "timed_out", "malformed", "abstained", "incomplete"] + output: str | None = None + completion_tokens: int | None = Field(default=None, ge=0) + latency_ms: float | None = Field(default=None, ge=0) + error_category: str | None = None + + +class ScoreRecord(EvalModel): + """One atomic grader judgment without execution-only task material.""" + + planned_run_id: str = Field(pattern=r"^run_[A-Za-z0-9_-]+$") + grader_id: str = Field(min_length=1) + critical_error_free: bool + dimensions: dict[str, int] = Field(default_factory=dict) + notes: str | None = None + + +class StudyManifest(EvalModel): + """Frozen preregistration of the complete execution matrix.""" + + study_id: str = Field(min_length=1) + seed: int + replicates: int = Field(ge=1) + task_ids: tuple[str, ...] = Field(min_length=1) + public_tasks_hash: Sha256Digest + planned_runs: tuple[PlannedRun, ...] = Field(min_length=1) + + @model_validator(mode="after") + def validate_complete_matrix(self) -> StudyManifest: + if len(set(self.task_ids)) != len(self.task_ids): + raise ValueError("task_ids must be unique") + + expected = { + (task_id, condition_id, replicate) + for task_id in self.task_ids + for condition_id in EVAL_CONDITION_IDS + for replicate in range(1, self.replicates + 1) + } + actual = {(run.task_id, run.condition_id, run.replicate) for run in self.planned_runs} + if actual != expected or len(self.planned_runs) != len(expected): + raise ValueError( + "planned_runs must contain the complete task x condition x replicate matrix" + ) + if any(run.study_id != self.study_id for run in self.planned_runs): + raise ValueError("every planned run must belong to this study") + if len({run.planned_run_id for run in self.planned_runs}) != len(self.planned_runs): + raise ValueError("planned_run_id values must be unique") + return self + + +class PublicTaskDataset(EvalModel): + """On-disk public dataset envelope.""" + + tasks: tuple[PublicTask, ...] + + +class GraderKeyDataset(EvalModel): + """On-disk grader-only dataset envelope.""" + + grader_keys: tuple[GraderKey, ...] diff --git a/src/conclave/evals/protocols.py b/src/conclave/evals/protocols.py new file mode 100644 index 0000000..221ec1f --- /dev/null +++ b/src/conclave/evals/protocols.py @@ -0,0 +1,129 @@ +"""Frozen study conditions and deterministic budget-matched planning.""" + +from __future__ import annotations + +import hashlib +import json +import random +from collections.abc import Mapping, Sequence + +from .dataset import hash_public_tasks +from .models import ( + EVAL_CONDITION_IDS, + EVAL_SCHEMA_VERSION, + ConditionId, + ConditionSpec, + PlannedRun, + PublicTask, + StudyManifest, +) + +CONDITIONS: tuple[ConditionSpec, ...] = ( + ConditionSpec(condition_id="single_frontier", description="One frontier-model answer."), + ConditionSpec( + condition_id="self_refine", description="One answer followed by self-refinement." + ), + ConditionSpec( + condition_id="independent_synthesis", + description="Independent answers combined by a synthesizer.", + ), + ConditionSpec(condition_id="critique_only", description="Independent answers plus critique."), + ConditionSpec(condition_id="revision_only", description="Independent answers plus revision."), + ConditionSpec(condition_id="elite_full", description="The complete Elite decision protocol."), +) +CONDITION_IDS: tuple[ConditionId, ...] = EVAL_CONDITION_IDS + + +def condition_order(seed: int) -> tuple[ConditionId, ...]: + """Return all frozen conditions in a deterministic seeded order.""" + + ordered = list(CONDITION_IDS) + random.Random(seed).shuffle(ordered) + return tuple(ordered) + + +def _validate_budgets(output_token_budgets: Mapping[str, int]) -> None: + if set(output_token_budgets) != set(CONDITION_IDS): + raise ValueError("output_token_budgets must contain exactly the six frozen conditions") + if any( + isinstance(value, bool) or not isinstance(value, int) or value <= 0 + for value in output_token_budgets.values() + ): + raise ValueError("output token budgets must be positive integers") + minimum = min(output_token_budgets.values()) + maximum = max(output_token_budgets.values()) + if maximum > minimum * 1.05: + raise ValueError("planned output-token ceilings must remain within 5% across conditions") + + +def _planned_run_id( + *, + study_id: str, + task_id: str, + condition_id: ConditionId, + replicate: int, + max_output_tokens: int, +) -> str: + identity = json.dumps( + { + "schema_version": EVAL_SCHEMA_VERSION, + "study_id": study_id, + "task_id": task_id, + "condition_id": condition_id, + "replicate": replicate, + "max_output_tokens": max_output_tokens, + }, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return f"run_{hashlib.sha256(identity).hexdigest()[:24]}" + + +def build_study_manifest( + *, + study_id: str, + tasks: Sequence[PublicTask], + replicates: int, + seed: int, + output_token_budgets: Mapping[str, int], +) -> StudyManifest: + """Predeclare every task-condition-replicate cell with stable identities.""" + + if replicates < 1: + raise ValueError("replicates must be at least one") + if not tasks: + raise ValueError("at least one public task is required") + task_ids = [task.task_id for task in tasks] + if len(set(task_ids)) != len(task_ids): + raise ValueError("public task_id values must be unique") + _validate_budgets(output_token_budgets) + + sorted_tasks = sorted(tasks, key=lambda task: task.task_id) + ordered_conditions = condition_order(seed) + planned_runs = tuple( + PlannedRun( + planned_run_id=_planned_run_id( + study_id=study_id, + task_id=task.task_id, + condition_id=condition_id, + replicate=replicate, + max_output_tokens=output_token_budgets[condition_id], + ), + study_id=study_id, + task_id=task.task_id, + condition_id=condition_id, + replicate=replicate, + max_output_tokens=output_token_budgets[condition_id], + ) + for task in sorted_tasks + for replicate in range(1, replicates + 1) + for condition_id in ordered_conditions + ) + return StudyManifest( + study_id=study_id, + seed=seed, + replicates=replicates, + task_ids=tuple(task.task_id for task in sorted_tasks), + public_tasks_hash=hash_public_tasks(sorted_tasks), + planned_runs=planned_runs, + ) diff --git a/tests/evals/test_models.py b/tests/evals/test_models.py new file mode 100644 index 0000000..312dc5a --- /dev/null +++ b/tests/evals/test_models.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import json + +import pytest +from pydantic import ValidationError + +from conclave.evals.dataset import ( + hash_grader_keys, + hash_public_tasks, + load_grader_keys, + load_public_tasks, +) +from conclave.evals.models import ( + EVAL_SCHEMA_VERSION, + GraderKey, + PublicTask, + ScoreRecord, + StudyManifest, +) + + +def test_eval_contracts_are_versioned_and_immutable() -> None: + task = PublicTask(task_id="task-1", prompt="Choose the safer migration plan.") + score = ScoreRecord( + planned_run_id="run_abc", + grader_id="grader-1", + critical_error_free=True, + dimensions={"correctness": 4}, + ) + + assert task.schema_version == EVAL_SCHEMA_VERSION + assert score.schema_version == EVAL_SCHEMA_VERSION + with pytest.raises(ValidationError): + task.prompt = "changed" + + +def test_public_tasks_and_grader_keys_load_from_separate_files(tmp_path) -> None: + public_path = tmp_path / "tasks.json" + key_path = tmp_path / "grader-keys.json" + public_path.write_text( + json.dumps( + { + "schema_version": EVAL_SCHEMA_VERSION, + "tasks": [ + { + "task_id": "task-1", + "prompt": "Choose a plan.", + "reference_packets": ["Public constraint A"], + } + ], + } + ) + ) + key_path.write_text( + json.dumps( + { + "schema_version": EVAL_SCHEMA_VERSION, + "grader_keys": [ + { + "task_id": "task-1", + "required_facts": ["Private expected fact"], + "critical_errors": ["Unsafe recommendation"], + } + ], + } + ) + ) + + tasks = load_public_tasks(public_path) + keys = load_grader_keys(key_path) + + assert tasks == [ + PublicTask( + task_id="task-1", + prompt="Choose a plan.", + reference_packets=("Public constraint A",), + ) + ] + assert keys == [ + GraderKey( + task_id="task-1", + required_facts=("Private expected fact",), + critical_errors=("Unsafe recommendation",), + ) + ] + assert "required_facts" not in tasks[0].model_dump() + + +def test_dataset_hashes_are_stable_across_input_order_and_formatting() -> None: + first = [ + PublicTask(task_id="b", prompt="Second"), + PublicTask(task_id="a", prompt="First", reference_packets=("R1",)), + ] + second = list(reversed(first)) + keys = [GraderKey(task_id="a", required_facts=("fact",))] + + assert hash_public_tasks(first) == hash_public_tasks(second) + assert hash_public_tasks(first).startswith("sha256:") + assert hash_grader_keys(keys) == hash_grader_keys(list(keys)) + assert hash_public_tasks(first) != hash_grader_keys(keys) + + +def test_study_manifest_rejects_an_unversioned_or_incomplete_run_matrix() -> None: + with pytest.raises(ValidationError): + StudyManifest( + study_id="study-1", + seed=7, + replicates=1, + task_ids=("task-1",), + public_tasks_hash="sha256:" + "a" * 64, + planned_runs=(), + ) diff --git a/tests/evals/test_protocols.py b/tests/evals/test_protocols.py new file mode 100644 index 0000000..04eabe8 --- /dev/null +++ b/tests/evals/test_protocols.py @@ -0,0 +1,115 @@ +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from conclave.evals.models import PublicTask +from conclave.evals.protocols import ( + CONDITION_IDS, + build_study_manifest, + condition_order, +) + +EXPECTED_CONDITIONS = ( + "single_frontier", + "self_refine", + "independent_synthesis", + "critique_only", + "revision_only", + "elite_full", +) + + +def test_registry_contains_exactly_the_six_frozen_conditions() -> None: + assert CONDITION_IDS == EXPECTED_CONDITIONS + + +def test_condition_order_is_seeded_deterministic_and_complete() -> None: + first = condition_order(seed=20260717) + second = condition_order(seed=20260717) + other = condition_order(seed=20260718) + + assert first == second + assert set(first) == set(EXPECTED_CONDITIONS) + assert len(first) == len(EXPECTED_CONDITIONS) + assert first != other + + +def test_budget_plan_accepts_exact_boundary_and_rejects_over_five_percent() -> None: + tasks = [PublicTask(task_id="task-1", prompt="Decide")] + accepted = { + condition_id: (1000 if condition_id != "elite_full" else 1050) + for condition_id in EXPECTED_CONDITIONS + } + manifest = build_study_manifest( + study_id="study-1", + tasks=tasks, + replicates=1, + seed=17, + output_token_budgets=accepted, + ) + assert {run.max_output_tokens for run in manifest.planned_runs} == {1000, 1050} + + rejected = dict(accepted) + rejected["elite_full"] = 1051 + with pytest.raises(ValueError, match="within 5%"): + build_study_manifest( + study_id="study-1", + tasks=tasks, + replicates=1, + seed=17, + output_token_budgets=rejected, + ) + + +def test_plan_declares_complete_matrix_with_stable_immutable_ids() -> None: + tasks = [ + PublicTask(task_id="task-b", prompt="B"), + PublicTask(task_id="task-a", prompt="A"), + ] + budgets = dict.fromkeys(EXPECTED_CONDITIONS, 1200) + + first = build_study_manifest( + study_id="elite-pilot", + tasks=tasks, + replicates=2, + seed=99, + output_token_budgets=budgets, + ) + second = build_study_manifest( + study_id="elite-pilot", + tasks=list(reversed(tasks)), + replicates=2, + seed=99, + output_token_budgets=budgets, + ) + + assert len(first.planned_runs) == 2 * 6 * 2 + assert len({run.planned_run_id for run in first.planned_runs}) == 24 + assert {run.task_id for run in first.planned_runs} == {"task-a", "task-b"} + assert {run.condition_id for run in first.planned_runs} == set(EXPECTED_CONDITIONS) + assert {run.replicate for run in first.planned_runs} == {1, 2} + assert first == second + assert all(run.planned_run_id.startswith("run_") for run in first.planned_runs) + with pytest.raises(ValidationError): + first.planned_runs[0].max_output_tokens = 1 + + +def test_plan_requires_every_condition_budget_and_valid_replicates() -> None: + tasks = [PublicTask(task_id="task-1", prompt="Decide")] + with pytest.raises(ValueError, match="exactly"): + build_study_manifest( + study_id="study-1", + tasks=tasks, + replicates=1, + seed=17, + output_token_budgets={"elite_full": 1000}, + ) + with pytest.raises(ValueError, match="replicates"): + build_study_manifest( + study_id="study-1", + tasks=tasks, + replicates=0, + seed=17, + output_token_budgets=dict.fromkeys(EXPECTED_CONDITIONS, 1000), + ) From 27cf1a40268775f83082d075d73ad8a5940ba978 Mon Sep 17 00:00:00 2001 From: ernestprovo23 Date: Fri, 17 Jul 2026 16:51:54 -0400 Subject: [PATCH 3/7] feat(evals): add budget caps and strict replay --- src/conclave/adapters/anthropic.py | 12 +- src/conclave/adapters/base.py | 4 + src/conclave/adapters/gemini.py | 16 ++- src/conclave/adapters/openai_compat.py | 15 ++- src/conclave/evals/replay.py | 175 +++++++++++++++++++++++++ src/conclave/providers.py | 21 ++- tests/evals/test_replay.py | 101 ++++++++++++++ tests/test_output_budget_plumbing.py | 111 ++++++++++++++++ 8 files changed, 446 insertions(+), 9 deletions(-) create mode 100644 src/conclave/evals/replay.py create mode 100644 tests/evals/test_replay.py create mode 100644 tests/test_output_budget_plumbing.py diff --git a/src/conclave/adapters/anthropic.py b/src/conclave/adapters/anthropic.py index 2a28c62..df6afaf 100644 --- a/src/conclave/adapters/anthropic.py +++ b/src/conclave/adapters/anthropic.py @@ -120,6 +120,7 @@ def build_request( timeout: float, api_key: str, output_contract: OutputContract | None = None, + max_output_tokens: int | None = None, ) -> tuple[str, dict[str, str], dict]: """Build the Messages POST, hoisting system out of the message array. @@ -157,7 +158,7 @@ def build_request( body: dict = { "model": self._bare_model(model_id), - "max_tokens": self.max_tokens, + "max_tokens": max_output_tokens if max_output_tokens is not None else self.max_tokens, "messages": turns, } if temperature is not None: @@ -261,6 +262,7 @@ def stream_request( timeout: float, api_key: str, output_contract: OutputContract | None = None, + max_output_tokens: int | None = None, ) -> tuple[str, dict[str, str], dict]: """Build the streaming POST: ``build_request`` + ``stream: true``. @@ -271,7 +273,13 @@ def stream_request( fragments. See :meth:`ProviderAdapter.stream_request`. """ url, headers, body = self.build_request( - model_id, messages, temperature, timeout, api_key, output_contract + model_id, + messages, + temperature, + timeout, + api_key, + output_contract=output_contract, + max_output_tokens=max_output_tokens, ) body["stream"] = True return url, headers, body diff --git a/src/conclave/adapters/base.py b/src/conclave/adapters/base.py index 3ae1a74..35bb952 100644 --- a/src/conclave/adapters/base.py +++ b/src/conclave/adapters/base.py @@ -274,6 +274,7 @@ def build_request( timeout: float, api_key: str, output_contract: OutputContract | None = None, + max_output_tokens: int | None = None, ) -> tuple[str, dict[str, str], dict]: """Build ``(url, headers, json_body)`` for this provider. @@ -292,6 +293,8 @@ def build_request( ``responseSchema`` / Anthropic tool ``input_schema``) is deferred to the CAC-02-OAI/ANT/GEM tickets; today every adapter accepts and ignores it. + max_output_tokens: Optional per-call output ceiling. ``None`` keeps + the adapter's existing provider default unchanged. Returns: A ``(url, headers, json_body)`` tuple ready for ``post_json``. @@ -322,6 +325,7 @@ def stream_request( timeout: float, api_key: str, output_contract: OutputContract | None = None, + max_output_tokens: int | None = None, ) -> tuple[str, dict[str, str], dict]: """Build ``(url, headers, json_body)`` for a STREAMING request (issue #7). diff --git a/src/conclave/adapters/gemini.py b/src/conclave/adapters/gemini.py index 801a7bb..35d0b39 100644 --- a/src/conclave/adapters/gemini.py +++ b/src/conclave/adapters/gemini.py @@ -290,6 +290,7 @@ def build_request( timeout: float, api_key: str, output_contract: OutputContract | None = None, + max_output_tokens: int | None = None, ) -> tuple[str, dict[str, str], dict]: """Build the generateContent POST. @@ -320,7 +321,11 @@ def build_request( gemini_role = _ROLE_MAP.get(role, "user") contents.append({"role": gemini_role, "parts": [{"text": content}]}) - generation_config: dict = {"maxOutputTokens": self.max_output_tokens} + generation_config: dict = { + "maxOutputTokens": ( + max_output_tokens if max_output_tokens is not None else self.max_output_tokens + ) + } if temperature is not None: generation_config["temperature"] = temperature # Conditional structured-output injection (no-op when contract is None, @@ -367,6 +372,7 @@ def stream_request( timeout: float, api_key: str, output_contract: OutputContract | None = None, + max_output_tokens: int | None = None, ) -> tuple[str, dict[str, str], dict]: """Build the streaming POST against ``streamGenerateContent?alt=sse``. @@ -380,7 +386,13 @@ def stream_request( # output_contract flows into build_request, which performs the # capability-gated responseMimeType/responseSchema injection. _url, headers, body = self.build_request( - model_id, messages, temperature, timeout, api_key, output_contract + model_id, + messages, + temperature, + timeout, + api_key, + output_contract=output_contract, + max_output_tokens=max_output_tokens, ) model = self._bare_model(model_id) url = f"{GEMINI_BASE}/{model}:streamGenerateContent?alt=sse" diff --git a/src/conclave/adapters/openai_compat.py b/src/conclave/adapters/openai_compat.py index a993786..6ec9c04 100644 --- a/src/conclave/adapters/openai_compat.py +++ b/src/conclave/adapters/openai_compat.py @@ -178,6 +178,7 @@ def build_request( timeout: float, api_key: str, output_contract: OutputContract | None = None, + max_output_tokens: int | None = None, ) -> tuple[str, dict[str, str], dict]: """Build the OpenAI-style POST. @@ -196,8 +197,9 @@ def build_request( } if temperature is not None: body["temperature"] = temperature - if self.max_tokens is not None: - body["max_tokens"] = self.max_tokens + token_cap = max_output_tokens if max_output_tokens is not None else self.max_tokens + if token_cap is not None: + body["max_tokens"] = token_cap # Capability-gated ``response_format`` injection. Passes the FULL # provider-prefixed ``model_id`` (NOT the bare wire id already stored in # body["model"]) because the catalog is keyed on the full id. No-op when @@ -240,6 +242,7 @@ def stream_request( timeout: float, api_key: str, output_contract: OutputContract | None = None, + max_output_tokens: int | None = None, ) -> tuple[str, dict[str, str], dict]: """Build the streaming POST: ``build_request`` + ``stream`` flags. @@ -253,7 +256,13 @@ def stream_request( # capability-gated ``response_format`` shaping (compatible with # stream:true). Stream flags are layered on top of the shaped body. url, headers, body = self.build_request( - model_id, messages, temperature, timeout, api_key, output_contract + model_id, + messages, + temperature, + timeout, + api_key, + output_contract=output_contract, + max_output_tokens=max_output_tokens, ) body["stream"] = True body["stream_options"] = {"include_usage": True} diff --git a/src/conclave/evals/replay.py b/src/conclave/evals/replay.py new file mode 100644 index 0000000..c7e0764 --- /dev/null +++ b/src/conclave/evals/replay.py @@ -0,0 +1,175 @@ +"""Strict buffered transport record/replay for offline evaluation runs.""" + +from __future__ import annotations + +import hashlib +import json +from collections import Counter +from collections.abc import Awaitable, Callable, Mapping +from typing import Any, Literal +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + +from pydantic import BaseModel, ConfigDict, Field + +from ..adapters.base import redact +from .models import Sha256Digest + +REPLAY_SCHEMA_VERSION = "conclave_replay_v1" +_SENSITIVE_NAMES = { + "api_key", + "apikey", + "authorization", + "key", + "password", + "secret", + "signature", + "token", + "x-api-key", + "x-goog-api-key", +} + +PostJson = Callable[[str, dict[str, str], dict, float], Awaitable[tuple[int, object]]] + + +class ReplayError(RuntimeError): + """Base class for fail-closed replay errors.""" + + +class ReplayCompatibilityError(ReplayError): + """The artifact cannot be used with the requested study manifest.""" + + +class ReplayMismatchError(ReplayError): + """Recorded and attempted transport calls do not match exactly.""" + + +class ReplayRecord(BaseModel): + """One sanitized request/response exchange at a deterministic occurrence.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + request_hash: Sha256Digest + occurrence_index: int = Field(ge=0) + request: dict[str, Any] + status: int + response: Any + + +class ReplayArtifact(BaseModel): + """Versioned recording bound to the exact base study manifest.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + schema_version: Literal["conclave_replay_v1"] = REPLAY_SCHEMA_VERSION + base_manifest_hash: Sha256Digest + records: tuple[ReplayRecord, ...] + + +def _is_sensitive_name(name: str) -> bool: + lowered = name.lower().replace("-", "_") + return lowered in {item.replace("-", "_") for item in _SENSITIVE_NAMES} or any( + marker in lowered for marker in ("secret", "password", "authorization") + ) + + +def _sanitize(value: Any) -> Any: + if isinstance(value, Mapping): + return { + str(key): _sanitize(item) + for key, item in sorted(value.items(), key=lambda pair: str(pair[0])) + if not _is_sensitive_name(str(key)) + } + if isinstance(value, (list, tuple)): + return [_sanitize(item) for item in value] + if isinstance(value, str): + return redact(value) + if value is None or isinstance(value, (bool, int, float)): + return value + return redact(str(value)) + + +def _sanitize_url(url: str) -> str: + parts = urlsplit(url) + safe_query = [ + (name, redact(value)) + for name, value in parse_qsl(parts.query, keep_blank_values=True) + if not _is_sensitive_name(name) + ] + return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(safe_query), "")) + + +def _request(url: str, body: dict) -> tuple[dict[str, Any], str]: + safe = {"url": _sanitize_url(url), "body": _sanitize(body)} + canonical = json.dumps(safe, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + digest = f"sha256:{hashlib.sha256(canonical.encode('utf-8')).hexdigest()}" + return safe, digest + + +class RecordingPostJson: + """Callable drop-in for :func:`transport.post_json` that records safe artifacts.""" + + def __init__(self, delegate: PostJson, *, base_manifest_hash: str) -> None: + self._delegate = delegate + self._base_manifest_hash = base_manifest_hash + self._counts: Counter[str] = Counter() + self._records: list[ReplayRecord] = [] + + async def __call__( + self, url: str, headers: dict[str, str], json_body: dict, timeout: float + ) -> tuple[int, object]: + safe_request, request_hash = _request(url, json_body) + occurrence = self._counts[request_hash] + self._counts[request_hash] += 1 + status, response = await self._delegate(url, headers, json_body, timeout) + self._records.append( + ReplayRecord( + request_hash=request_hash, + occurrence_index=occurrence, + request=safe_request, + status=status, + response=_sanitize(response), + ) + ) + return status, response + + def artifact(self) -> ReplayArtifact: + return ReplayArtifact( + base_manifest_hash=self._base_manifest_hash, + records=tuple(self._records), + ) + + +class ReplayingPostJson: + """Callable zero-network replay that requires an exact complete call set.""" + + def __init__(self, artifact: ReplayArtifact, *, base_manifest_hash: str) -> None: + if artifact.schema_version != REPLAY_SCHEMA_VERSION: + raise ReplayCompatibilityError( + f"replay schema version mismatch: {artifact.schema_version!r}" + ) + if artifact.base_manifest_hash != base_manifest_hash: + raise ReplayCompatibilityError("replay base manifest hash mismatch") + self._records = { + (record.request_hash, record.occurrence_index): record for record in artifact.records + } + if len(self._records) != len(artifact.records): + raise ReplayCompatibilityError("replay contains duplicate request occurrences") + self._counts: Counter[str] = Counter() + self._consumed: set[tuple[str, int]] = set() + + async def __call__( + self, url: str, headers: dict[str, str], json_body: dict, timeout: float + ) -> tuple[int, object]: + del headers, timeout + _safe_request, request_hash = _request(url, json_body) + occurrence = self._counts[request_hash] + self._counts[request_hash] += 1 + key = (request_hash, occurrence) + record = self._records.get(key) + if record is None: + raise ReplayMismatchError(f"unmatched request {request_hash} occurrence {occurrence}") + self._consumed.add(key) + return record.status, record.response + + def assert_consumed(self) -> None: + remaining = set(self._records) - self._consumed + if remaining: + raise ReplayMismatchError(f"{len(remaining)} unconsumed record(s) remain") diff --git a/src/conclave/providers.py b/src/conclave/providers.py index 1ae0fe9..7bb52a9 100644 --- a/src/conclave/providers.py +++ b/src/conclave/providers.py @@ -132,6 +132,7 @@ async def call_model( timeout: float = 120.0, config: ConclaveConfig | None = None, output_contract: OutputContract | None = None, + max_output_tokens: int | None = None, ) -> ModelAnswer: """Call a single model and return a structured :class:`ModelAnswer`. @@ -158,6 +159,8 @@ async def call_model( ``input_schema`` / Gemini ``responseSchema``) and degrades gracefully on providers that cannot enforce it. ``None`` (the default) leaves the request body byte-for-byte unchanged -- the current free-prose behavior. + max_output_tokens: Optional per-call output-token ceiling. ``None`` keeps + the adapter's existing request body and provider default unchanged. Returns: A ``ModelAnswer`` with either ``answer`` populated or ``error`` set. @@ -185,7 +188,13 @@ async def call_model( try: url, headers, body = adapter.build_request( - model_id, messages, temperature, timeout, api_key, output_contract=output_contract + model_id, + messages, + temperature, + timeout, + api_key, + output_contract=output_contract, + max_output_tokens=max_output_tokens, ) status, payload = await transport.post_json(url, headers, body, timeout) text, usage = adapter.parse_response(status, payload) @@ -252,6 +261,7 @@ async def call_model_stream( timeout: float = 120.0, config: ConclaveConfig | None = None, output_contract: OutputContract | None = None, + max_output_tokens: int | None = None, ) -> AsyncIterator[str | ModelAnswer]: """Stream a single model's answer, yielding text deltas then a final answer. @@ -313,6 +323,7 @@ async def call_model_stream( timeout=timeout, config=resolved_config, output_contract=output_contract, + max_output_tokens=max_output_tokens, ) if answer.answer: yield answer.answer @@ -332,7 +343,13 @@ async def call_model_stream( usage: TokenUsage | None = None try: url, headers, body = adapter.stream_request( - model_id, messages, temperature, timeout, api_key, output_contract=output_contract + model_id, + messages, + temperature, + timeout, + api_key, + output_contract=output_contract, + max_output_tokens=max_output_tokens, ) async for event, data in transport.stream_sse(url, headers, body, timeout): delta = adapter.parse_sse_event(event, data) diff --git a/tests/evals/test_replay.py b/tests/evals/test_replay.py new file mode 100644 index 0000000..cb150a5 --- /dev/null +++ b/tests/evals/test_replay.py @@ -0,0 +1,101 @@ +"""Strict record/replay tests at the buffered transport seam.""" + +from __future__ import annotations + +import json + +import pytest + +from conclave.evals.replay import ( + REPLAY_SCHEMA_VERSION, + RecordingPostJson, + ReplayArtifact, + ReplayCompatibilityError, + ReplayingPostJson, + ReplayMismatchError, +) + +BASE_HASH = "sha256:" + "a" * 64 + + +async def test_record_replay_uses_occurrence_indexes_and_never_calls_network(): + calls = 0 + + async def network(url, headers, body, timeout): + nonlocal calls + calls += 1 + return 200, {"answer": calls} + + recorder = RecordingPostJson(network, base_manifest_hash=BASE_HASH) + args = ( + "https://api.example.test/v1/chat?alt=json", + {"Authorization": "Bearer sk-secret-value"}, + {"model": "m", "messages": [{"role": "user", "content": "same"}]}, + 30.0, + ) + assert await recorder(*args) == (200, {"answer": 1}) + assert await recorder(*args) == (200, {"answer": 2}) + artifact = recorder.artifact() + assert [record.occurrence_index for record in artifact.records] == [0, 1] + + async def forbidden_network(*args, **kwargs): + raise AssertionError("replay performed network I/O") + + replay = ReplayingPostJson(artifact, base_manifest_hash=BASE_HASH) + assert await replay(*args) == (200, {"answer": 1}) + assert await replay(*args) == (200, {"answer": 2}) + replay.assert_consumed() + assert calls == 2 + + +async def test_artifact_excludes_headers_keys_and_secret_url_parameters(): + async def network(url, headers, body, timeout): + return 200, {"ok": True} + + recorder = RecordingPostJson(network, base_manifest_hash=BASE_HASH) + await recorder( + "https://example.test/generate?key=AIzaSecretValue123&alt=json", + {"x-goog-api-key": "AIzaSecretValue123", "Authorization": "Bearer sk-secret-value"}, + {"model": "m", "api_key": "sk-secret-value", "prompt": "safe"}, + 10, + ) + encoded = json.dumps(recorder.artifact().model_dump(mode="json"), sort_keys=True) + assert "AIzaSecretValue123" not in encoded + assert "sk-secret-value" not in encoded + assert "Authorization" not in encoded + assert "x-goog-api-key" not in encoded + assert "api_key" not in encoded + assert "alt=json" in encoded + + +def test_replay_rejects_schema_or_base_manifest_hash_drift(): + artifact = ReplayArtifact(base_manifest_hash=BASE_HASH, records=()) + with pytest.raises(ReplayCompatibilityError, match="base manifest hash"): + ReplayingPostJson(artifact, base_manifest_hash="sha256:" + "b" * 64) + + drifted = artifact.model_copy(update={"schema_version": "future"}) + with pytest.raises(ReplayCompatibilityError, match="schema version"): + ReplayingPostJson(drifted, base_manifest_hash=BASE_HASH) + assert artifact.schema_version == REPLAY_SCHEMA_VERSION + + +async def test_replay_fails_closed_on_missing_mismatch_and_extra_records(): + async def network(url, headers, body, timeout): + return 200, {"ok": True} + + recorder = RecordingPostJson(network, base_manifest_hash=BASE_HASH) + await recorder("https://example.test/v1", {}, {"model": "m", "prompt": "one"}, 10) + artifact = recorder.artifact() + + replay = ReplayingPostJson(artifact, base_manifest_hash=BASE_HASH) + with pytest.raises(ReplayMismatchError, match="unmatched request"): + await replay("https://example.test/v1", {}, {"model": "m", "prompt": "two"}, 10) + + replay = ReplayingPostJson(artifact, base_manifest_hash=BASE_HASH) + with pytest.raises(ReplayMismatchError, match="unconsumed record"): + replay.assert_consumed() + + replay = ReplayingPostJson(artifact, base_manifest_hash=BASE_HASH) + await replay("https://example.test/v1", {}, {"model": "m", "prompt": "one"}, 10) + with pytest.raises(ReplayMismatchError, match="unmatched request"): + await replay("https://example.test/v1", {}, {"model": "m", "prompt": "one"}, 10) diff --git a/tests/test_output_budget_plumbing.py b/tests/test_output_budget_plumbing.py new file mode 100644 index 0000000..a6a0aa9 --- /dev/null +++ b/tests/test_output_budget_plumbing.py @@ -0,0 +1,111 @@ +"""Output-token ceilings are optional, provider-native, and end-to-end.""" + +from __future__ import annotations + +from conclave.adapters.anthropic import AnthropicAdapter +from conclave.adapters.gemini import GeminiAdapter +from conclave.adapters.openai_compat import OpenAICompatAdapter +from conclave.models import ModelAnswer +from conclave.providers import call_model, call_model_stream + +MESSAGES = [{"role": "user", "content": "hi"}] + + +def test_adapter_caps_override_native_defaults_without_changing_default_bodies(): + openai = OpenAICompatAdapter("openai", "https://example.test/v1/chat", ("OPENAI_API_KEY",)) + anthropic = AnthropicAdapter() + gemini = GeminiAdapter() + + _, _, openai_default = openai.build_request("openai/gpt-4.1", MESSAGES, 0.2, 30, "key") + _, _, anthropic_default = anthropic.build_request( + "anthropic/claude-sonnet-4-20250514", MESSAGES, 0.2, 30, "key" + ) + _, _, gemini_default = gemini.build_request("gemini/gemini-2.5-pro", MESSAGES, 0.2, 30, "key") + + _, _, openai_capped = openai.build_request( + "openai/gpt-4.1", MESSAGES, 0.2, 30, "key", max_output_tokens=321 + ) + _, _, anthropic_capped = anthropic.build_request( + "anthropic/claude-sonnet-4-20250514", + MESSAGES, + 0.2, + 30, + "key", + max_output_tokens=321, + ) + _, _, gemini_capped = gemini.build_request( + "gemini/gemini-2.5-pro", MESSAGES, 0.2, 30, "key", max_output_tokens=321 + ) + + assert "max_tokens" not in openai_default + assert anthropic_default["max_tokens"] == anthropic.max_tokens + assert gemini_default["generationConfig"]["maxOutputTokens"] == gemini.max_output_tokens + assert openai_capped["max_tokens"] == 321 + assert anthropic_capped["max_tokens"] == 321 + assert gemini_capped["generationConfig"]["maxOutputTokens"] == 321 + + +def test_stream_requests_receive_the_same_optional_cap(): + adapters_and_models = ( + ( + OpenAICompatAdapter("openai", "https://example.test/v1/chat", ("OPENAI_API_KEY",)), + "openai/gpt-4.1", + lambda body: body["max_tokens"], + ), + (AnthropicAdapter(), "anthropic/claude-sonnet-4-20250514", lambda body: body["max_tokens"]), + ( + GeminiAdapter(), + "gemini/gemini-2.5-pro", + lambda body: body["generationConfig"]["maxOutputTokens"], + ), + ) + for adapter, model_id, extract in adapters_and_models: + _, _, body = adapter.stream_request( + model_id, MESSAGES, 0.2, 30, "key", max_output_tokens=654 + ) + assert extract(body) == 654 + + +async def test_call_model_threads_max_output_tokens(monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "sk-test-value") + monkeypatch.setenv("CONCLAVE_CONFIG", "/nonexistent/conclave.yml") + captured = {} + + async def fake_post_json(url, headers, json_body, timeout): + captured["body"] = json_body + return 200, {"choices": [{"message": {"content": "ok"}}]} + + monkeypatch.setattr("conclave.transport.post_json", fake_post_json) + answer = await call_model( + "openai", + "openai/gpt-4.1", + MESSAGES, + max_output_tokens=777, + ) + assert answer.ok + assert captured["body"]["max_tokens"] == 777 + + +async def test_call_model_stream_threads_max_output_tokens(monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "sk-test-value") + monkeypatch.setenv("CONCLAVE_CONFIG", "/nonexistent/conclave.yml") + captured = {} + + async def fake_stream_sse(url, headers, json_body, timeout): + captured["body"] = json_body + yield "", '{"choices":[{"delta":{"content":"ok"}}]}' + yield "", "[DONE]" + + monkeypatch.setattr("conclave.transport.stream_sse", fake_stream_sse) + items = [ + item + async for item in call_model_stream( + "openai", + "openai/gpt-4.1", + MESSAGES, + max_output_tokens=888, + ) + ] + assert isinstance(items[-1], ModelAnswer) + assert items[-1].ok + assert captured["body"]["max_tokens"] == 888 From 969019548a7e4e0fb74a4521e9133a6367f2a2fc Mon Sep 17 00:00:00 2001 From: ernestprovo23 Date: Fri, 17 Jul 2026 16:58:33 -0400 Subject: [PATCH 4/7] feat(evals): add failure-inclusive runner and blinding --- src/conclave/evals/__init__.py | 4 + src/conclave/evals/blinding.py | 78 +++++++++++++ src/conclave/evals/models.py | 24 +++- src/conclave/evals/runner.py | 159 ++++++++++++++++++++++++++ tests/evals/test_blinding.py | 44 ++++++++ tests/evals/test_runner.py | 199 +++++++++++++++++++++++++++++++++ 6 files changed, 507 insertions(+), 1 deletion(-) create mode 100644 src/conclave/evals/blinding.py create mode 100644 src/conclave/evals/runner.py create mode 100644 tests/evals/test_blinding.py create mode 100644 tests/evals/test_runner.py diff --git a/src/conclave/evals/__init__.py b/src/conclave/evals/__init__.py index 2a019f6..e152834 100644 --- a/src/conclave/evals/__init__.py +++ b/src/conclave/evals/__init__.py @@ -9,10 +9,12 @@ ConditionSpec, GraderKey, PlannedRun, + ProtocolExecution, PublicTask, RunRecord, ScoreRecord, StudyManifest, + StudyRun, ) __all__ = [ @@ -20,8 +22,10 @@ "ConditionSpec", "GraderKey", "PlannedRun", + "ProtocolExecution", "PublicTask", "RunRecord", "ScoreRecord", "StudyManifest", + "StudyRun", ] diff --git a/src/conclave/evals/blinding.py b/src/conclave/evals/blinding.py new file mode 100644 index 0000000..42a0163 --- /dev/null +++ b/src/conclave/evals/blinding.py @@ -0,0 +1,78 @@ +"""Seeded opaque output blinding with a physically separate identity map.""" + +from __future__ import annotations + +import hashlib +import random +from collections.abc import Sequence + +from pydantic import BaseModel, ConfigDict, Field + +from .models import EVAL_SCHEMA_VERSION, RunOutcome, RunRecord, SchemaVersion + + +class BlindModel(BaseModel): + """Immutable and drift-resistant blinding artifact contract.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + schema_version: SchemaVersion = EVAL_SCHEMA_VERSION + + +class BlindedOutput(BlindModel): + """Grader-visible output without execution identity or labels.""" + + opaque_output_id: str = Field(pattern=r"^output_[0-9a-f]{24}$") + outcome: RunOutcome + output: str | None = None + + +class BlindedOutputSet(BlindModel): + """Artifact safe to provide to graders.""" + + outputs: tuple[BlindedOutput, ...] + + +class BlindMapEntry(BlindModel): + """Restricted mapping from an opaque output to its execution identity.""" + + opaque_output_id: str = Field(pattern=r"^output_[0-9a-f]{24}$") + planned_run_id: str = Field(pattern=r"^run_[0-9a-f]{24}$") + + +class BlindMap(BlindModel): + """Separate restricted identity artifact, never included in grader output.""" + + entries: tuple[BlindMapEntry, ...] + + +def _opaque_id(*, planned_run_id: str, seed: int) -> str: + identity = f"{seed}:{planned_run_id}".encode() + return f"output_{hashlib.sha256(identity).hexdigest()[:24]}" + + +def blind_run_records( + records: Sequence[RunRecord], *, seed: int +) -> tuple[BlindedOutputSet, BlindMap]: + """Blind and deterministically shuffle records, returning the map separately.""" + + if len({record.planned_run_id for record in records}) != len(records): + raise ValueError("run records must have unique planned_run_id values") + blinded = [ + ( + BlindedOutput( + opaque_output_id=_opaque_id(planned_run_id=record.planned_run_id, seed=seed), + outcome=record.outcome, + output=record.output, + ), + BlindMapEntry( + opaque_output_id=_opaque_id(planned_run_id=record.planned_run_id, seed=seed), + planned_run_id=record.planned_run_id, + ), + ) + for record in records + ] + random.Random(seed).shuffle(blinded) + return ( + BlindedOutputSet(outputs=tuple(output for output, _ in blinded)), + BlindMap(entries=tuple(entry for _, entry in blinded)), + ) diff --git a/src/conclave/evals/models.py b/src/conclave/evals/models.py index 7d49186..81bd18a 100644 --- a/src/conclave/evals/models.py +++ b/src/conclave/evals/models.py @@ -28,6 +28,7 @@ ) Sha256Digest = Annotated[str, Field(pattern=r"^sha256:[0-9a-f]{64}$")] +RunOutcome = Literal["success", "failed", "timed_out", "malformed", "abstained", "incomplete"] class EvalModel(BaseModel): @@ -73,17 +74,38 @@ class PlannedRun(EvalModel): max_output_tokens: int = Field(gt=0) +class ProtocolExecution(EvalModel): + """Typed output returned by an injected offline protocol executor.""" + + outcome: RunOutcome + output: str | None = None + completion_tokens: int | None = Field(default=None, ge=0) + latency_ms: float | None = Field(default=None, ge=0) + error_category: str | None = None + + class RunRecord(EvalModel): """Failure-inclusive result for one predeclared cell.""" planned_run_id: str = Field(pattern=r"^run_[0-9a-f]{24}$") - outcome: Literal["success", "failed", "timed_out", "malformed", "abstained", "incomplete"] + outcome: RunOutcome output: str | None = None completion_tokens: int | None = Field(default=None, ge=0) latency_ms: float | None = Field(default=None, ge=0) error_category: str | None = None +class StudyRun(EvalModel): + """Complete failure-inclusive execution result for a frozen manifest.""" + + study_id: str = Field(min_length=1) + records: tuple[RunRecord, ...] + total_planned_runs: int = Field(ge=1) + outcome_counts: dict[RunOutcome, int] + total_completion_tokens: int = Field(ge=0) + total_latency_ms: float = Field(ge=0) + + class ScoreRecord(EvalModel): """One atomic grader judgment without execution-only task material.""" diff --git a/src/conclave/evals/runner.py b/src/conclave/evals/runner.py new file mode 100644 index 0000000..ffadd44 --- /dev/null +++ b/src/conclave/evals/runner.py @@ -0,0 +1,159 @@ +"""Deterministic, executor-injected runner for offline evaluation studies.""" + +from __future__ import annotations + +import asyncio +import time +from collections import Counter +from collections.abc import Awaitable, Callable, Mapping, Sequence + +from .dataset import hash_public_tasks +from .models import ( + EVAL_CONDITION_IDS, + ConditionId, + PlannedRun, + ProtocolExecution, + PublicTask, + RunRecord, + StudyManifest, + StudyRun, +) + +ProtocolExecutor = Callable[ + ..., + Awaitable[ProtocolExecution], +] + + +class RunValidationError(ValueError): + """Execution inputs or outputs do not exactly match the frozen manifest.""" + + +def validate_run_records( + manifest: StudyManifest, records: Sequence[RunRecord] +) -> tuple[RunRecord, ...]: + """Require exactly one result for every planned cell and no other results.""" + + planned_ids = [planned.planned_run_id for planned in manifest.planned_runs] + actual_ids = [record.planned_run_id for record in records] + if Counter(actual_ids) != Counter(planned_ids): + raise RunValidationError("run records must cover every planned_run_id exactly once") + return tuple(records) + + +def _validate_inputs( + manifest: StudyManifest, + tasks: Sequence[PublicTask], + executors: Mapping[ConditionId, ProtocolExecutor], +) -> dict[str, PublicTask]: + task_by_id = {task.task_id: task for task in tasks} + if len(task_by_id) != len(tasks) or set(task_by_id) != set(manifest.task_ids): + raise RunValidationError("public tasks must exactly match the manifest task_ids") + if hash_public_tasks(tasks) != manifest.public_tasks_hash: + raise RunValidationError("public task content does not match the manifest hash") + if set(executors) != set(EVAL_CONDITION_IDS): + raise RunValidationError("executors must contain exactly the six frozen conditions") + return task_by_id + + +async def _execute_cell( + *, + planned_run: PlannedRun, + task: PublicTask, + executor: ProtocolExecutor, + timeout_seconds: float | None, +) -> RunRecord: + started = time.perf_counter() + try: + invocation = executor( + task=task, + planned_run=planned_run, + max_output_tokens=planned_run.max_output_tokens, + ) + execution = ( + await asyncio.wait_for(invocation, timeout=timeout_seconds) + if timeout_seconds is not None + else await invocation + ) + if not isinstance(execution, ProtocolExecution): + return RunRecord( + planned_run_id=planned_run.planned_run_id, + outcome="malformed", + latency_ms=(time.perf_counter() - started) * 1000, + error_category="invalid_executor_result", + ) + if execution.completion_tokens is not None and ( + execution.completion_tokens > planned_run.max_output_tokens + ): + return RunRecord( + planned_run_id=planned_run.planned_run_id, + outcome="malformed", + completion_tokens=execution.completion_tokens, + latency_ms=execution.latency_ms, + error_category="output_budget_exceeded", + ) + if execution.outcome == "success" and execution.output is None: + return RunRecord( + planned_run_id=planned_run.planned_run_id, + outcome="malformed", + completion_tokens=execution.completion_tokens, + latency_ms=execution.latency_ms, + error_category="missing_success_output", + ) + return RunRecord( + planned_run_id=planned_run.planned_run_id, + outcome=execution.outcome, + output=execution.output, + completion_tokens=execution.completion_tokens, + latency_ms=execution.latency_ms, + error_category=execution.error_category, + ) + except TimeoutError: + return RunRecord( + planned_run_id=planned_run.planned_run_id, + outcome="timed_out", + latency_ms=(time.perf_counter() - started) * 1000, + error_category="timeout", + ) + except Exception: + return RunRecord( + planned_run_id=planned_run.planned_run_id, + outcome="failed", + latency_ms=(time.perf_counter() - started) * 1000, + error_category="executor_error", + ) + + +async def run_study( + *, + manifest: StudyManifest, + tasks: Sequence[PublicTask], + executors: Mapping[ConditionId, ProtocolExecutor], + timeout_seconds: float | None = None, +) -> StudyRun: + """Execute every predeclared cell sequentially with no provider dependency.""" + + if timeout_seconds is not None and timeout_seconds <= 0: + raise ValueError("timeout_seconds must be positive") + task_by_id = _validate_inputs(manifest, tasks, executors) + records = tuple( + [ + await _execute_cell( + planned_run=planned_run, + task=task_by_id[planned_run.task_id], + executor=executors[planned_run.condition_id], + timeout_seconds=timeout_seconds, + ) + for planned_run in manifest.planned_runs + ] + ) + records = validate_run_records(manifest, records) + outcome_counts = dict(sorted(Counter(record.outcome for record in records).items())) + return StudyRun( + study_id=manifest.study_id, + records=records, + total_planned_runs=len(manifest.planned_runs), + outcome_counts=outcome_counts, + total_completion_tokens=sum(record.completion_tokens or 0 for record in records), + total_latency_ms=sum(record.latency_ms or 0.0 for record in records), + ) diff --git a/tests/evals/test_blinding.py b/tests/evals/test_blinding.py new file mode 100644 index 0000000..a80e431 --- /dev/null +++ b/tests/evals/test_blinding.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from conclave.evals.blinding import blind_run_records +from conclave.evals.models import RunRecord + + +def _records() -> tuple[RunRecord, ...]: + return ( + RunRecord(planned_run_id="run_" + "a" * 24, outcome="success", output="Alpha"), + RunRecord(planned_run_id="run_" + "b" * 24, outcome="failed", error_category="error"), + RunRecord(planned_run_id="run_" + "c" * 24, outcome="abstained"), + ) + + +def test_blinding_is_seeded_deterministic_opaque_and_separates_the_map() -> None: + first_outputs, first_map = blind_run_records(_records(), seed=20260717) + second_outputs, second_map = blind_run_records(_records(), seed=20260717) + other_outputs, other_map = blind_run_records(_records(), seed=20260718) + + assert first_outputs == second_outputs + assert first_map == second_map + assert first_outputs != other_outputs + assert first_map != other_map + assert len({item.opaque_output_id for item in first_outputs.outputs}) == 3 + assert all(item.opaque_output_id.startswith("output_") for item in first_outputs.outputs) + assert {entry.opaque_output_id for entry in first_map.entries} == { + output.opaque_output_id for output in first_outputs.outputs + } + assert {entry.planned_run_id for entry in first_map.entries} == { + record.planned_run_id for record in _records() + } + + +def test_blinded_outputs_exclude_run_condition_provider_and_model_labels() -> None: + blinded_outputs, blind_map = blind_run_records(_records(), seed=9) + serialized_outputs = blinded_outputs.model_dump_json() + + assert "planned_run_id" not in serialized_outputs + assert "condition" not in serialized_outputs + assert "provider" not in serialized_outputs + assert "model" not in serialized_outputs + assert "run_aaaaaaaaaaaaaaaaaaaaaaaa" not in serialized_outputs + assert "Alpha" in serialized_outputs + assert "planned_run_id" in blind_map.model_dump_json() diff --git a/tests/evals/test_runner.py b/tests/evals/test_runner.py new file mode 100644 index 0000000..cf58f93 --- /dev/null +++ b/tests/evals/test_runner.py @@ -0,0 +1,199 @@ +from __future__ import annotations + +import asyncio + +import pytest +from pydantic import ValidationError + +from conclave.evals.models import ProtocolExecution, PublicTask, RunRecord +from conclave.evals.protocols import CONDITION_IDS, build_study_manifest +from conclave.evals.runner import RunValidationError, run_study, validate_run_records + + +def _study(): + tasks = [ + PublicTask(task_id="task-a", prompt="Choose A."), + PublicTask(task_id="task-b", prompt="Choose B."), + ] + manifest = build_study_manifest( + study_id="offline-study", + tasks=tasks, + replicates=2, + seed=17, + output_token_budgets={condition_id: 400 for condition_id in CONDITION_IDS}, + ) + return tasks, manifest + + +@pytest.mark.asyncio +async def test_runner_executes_complete_matrix_with_cell_budget_and_immutable_records() -> None: + tasks, manifest = _study() + calls: list[tuple[str, str, int]] = [] + + async def executor(*, task, planned_run, max_output_tokens): + calls.append((task.task_id, planned_run.condition_id, max_output_tokens)) + return ProtocolExecution( + outcome="success", + output=f"{task.task_id}:{planned_run.condition_id}", + completion_tokens=11, + latency_ms=2.5, + ) + + study_run = await run_study( + manifest=manifest, + tasks=tasks, + executors={condition_id: executor for condition_id in CONDITION_IDS}, + ) + + assert len(calls) == len(manifest.planned_runs) == 24 + assert {condition_id for _, condition_id, _ in calls} == set(CONDITION_IDS) + assert {budget for _, _, budget in calls} == {400} + assert tuple(record.planned_run_id for record in study_run.records) == tuple( + run.planned_run_id for run in manifest.planned_runs + ) + assert study_run.total_planned_runs == 24 + assert study_run.outcome_counts == {"success": 24} + assert study_run.total_completion_tokens == 264 + assert study_run.total_latency_ms == 60.0 + with pytest.raises(ValidationError): + study_run.records[0].output = "changed" + + +@pytest.mark.asyncio +async def test_runner_preserves_every_non_success_outcome_in_denominator() -> None: + tasks = [PublicTask(task_id="task-a", prompt="Choose A.")] + manifest = build_study_manifest( + study_id="failure-study", + tasks=tasks, + replicates=1, + seed=2, + output_token_budgets={condition_id: 100 for condition_id in CONDITION_IDS}, + ) + outcomes = dict( + zip( + CONDITION_IDS, + ("failed", "timed_out", "abstained", "malformed", "incomplete", "success"), + strict=True, + ) + ) + + async def executor(*, task, planned_run, max_output_tokens): + del task, max_output_tokens + outcome = outcomes[planned_run.condition_id] + return ProtocolExecution( + outcome=outcome, + output="answer" if outcome == "success" else None, + completion_tokens=3 if outcome in {"success", "incomplete"} else None, + latency_ms=4.0, + error_category=None if outcome in {"success", "abstained"} else outcome, + ) + + study_run = await run_study( + manifest=manifest, + tasks=tasks, + executors={condition_id: executor for condition_id in CONDITION_IDS}, + ) + + assert study_run.total_planned_runs == 6 + assert sum(study_run.outcome_counts.values()) == 6 + assert study_run.outcome_counts == { + "abstained": 1, + "failed": 1, + "incomplete": 1, + "malformed": 1, + "success": 1, + "timed_out": 1, + } + assert study_run.total_completion_tokens == 6 + assert study_run.total_latency_ms == 24.0 + + +@pytest.mark.asyncio +async def test_runner_fails_closed_when_executor_exceeds_budget_or_returns_empty_success() -> None: + tasks = [PublicTask(task_id="task-a", prompt="Choose A.")] + manifest = build_study_manifest( + study_id="budget-study", + tasks=tasks, + replicates=1, + seed=4, + output_token_budgets={condition_id: 100 for condition_id in CONDITION_IDS}, + ) + + async def executor(*, task, planned_run, max_output_tokens): + del task, max_output_tokens + if planned_run.condition_id == "single_frontier": + return ProtocolExecution(outcome="success", output="too long", completion_tokens=101) + if planned_run.condition_id == "self_refine": + return ProtocolExecution(outcome="success", output=None, completion_tokens=4) + return ProtocolExecution(outcome="success", output="answer", completion_tokens=4) + + study_run = await run_study( + manifest=manifest, + tasks=tasks, + executors={condition_id: executor for condition_id in CONDITION_IDS}, + ) + by_condition = { + planned.condition_id: record + for planned, record in zip(manifest.planned_runs, study_run.records, strict=True) + } + + assert by_condition["single_frontier"].outcome == "malformed" + assert by_condition["single_frontier"].error_category == "output_budget_exceeded" + assert by_condition["self_refine"].outcome == "malformed" + assert by_condition["self_refine"].error_category == "missing_success_output" + assert study_run.total_planned_runs == 6 + assert sum(study_run.outcome_counts.values()) == 6 + + +@pytest.mark.asyncio +async def test_runner_converts_executor_errors_and_timeouts_to_records() -> None: + tasks = [PublicTask(task_id="task-a", prompt="Choose A.")] + manifest = build_study_manifest( + study_id="exception-study", + tasks=tasks, + replicates=1, + seed=3, + output_token_budgets={condition_id: 100 for condition_id in CONDITION_IDS}, + ) + + async def executor(*, task, planned_run, max_output_tokens): + del task, max_output_tokens + if planned_run.condition_id == "single_frontier": + raise RuntimeError("provider detail must not escape") + if planned_run.condition_id == "self_refine": + await asyncio.sleep(0.05) + return ProtocolExecution(outcome="success", output="answer") + + study_run = await run_study( + manifest=manifest, + tasks=tasks, + executors={condition_id: executor for condition_id in CONDITION_IDS}, + timeout_seconds=0.001, + ) + + by_condition = { + planned.condition_id: record + for planned, record in zip(manifest.planned_runs, study_run.records, strict=True) + } + assert by_condition["single_frontier"].outcome == "failed" + assert by_condition["single_frontier"].error_category == "executor_error" + assert by_condition["self_refine"].outcome == "timed_out" + assert by_condition["self_refine"].error_category == "timeout" + assert "provider detail" not in study_run.model_dump_json() + + +def test_record_validation_fails_closed_on_missing_duplicate_or_unplanned_output() -> None: + _, manifest = _study() + valid = tuple( + RunRecord(planned_run_id=run.planned_run_id, outcome="success", output="answer") + for run in manifest.planned_runs + ) + + assert validate_run_records(manifest, valid) == valid + with pytest.raises(RunValidationError, match="exactly once"): + validate_run_records(manifest, valid[:-1]) + with pytest.raises(RunValidationError, match="exactly once"): + validate_run_records(manifest, (*valid, valid[0])) + unplanned = RunRecord(planned_run_id="run_" + "f" * 24, outcome="success", output="x") + with pytest.raises(RunValidationError, match="exactly once"): + validate_run_records(manifest, (*valid[:-1], unplanned)) From a63218537c47f04cb4b35b4d37044e9f680c7dde Mon Sep 17 00:00:00 2001 From: ernestprovo23 Date: Fri, 17 Jul 2026 17:12:17 -0400 Subject: [PATCH 5/7] feat(evals): add failure-inclusive scoring reports --- src/conclave/evals/__init__.py | 4 + src/conclave/evals/reporting.py | 108 +++++++++ src/conclave/evals/scoring.py | 406 ++++++++++++++++++++++++++++++++ tests/evals/test_reporting.py | 77 ++++++ tests/evals/test_scoring.py | 242 +++++++++++++++++++ 5 files changed, 837 insertions(+) create mode 100644 src/conclave/evals/reporting.py create mode 100644 src/conclave/evals/scoring.py create mode 100644 tests/evals/test_reporting.py create mode 100644 tests/evals/test_scoring.py diff --git a/src/conclave/evals/__init__.py b/src/conclave/evals/__init__.py index e152834..ad9260e 100644 --- a/src/conclave/evals/__init__.py +++ b/src/conclave/evals/__init__.py @@ -16,6 +16,7 @@ StudyManifest, StudyRun, ) +from .scoring import AdjudicationRecord, GraderJudgment, StudyScoreReport __all__ = [ "EVAL_SCHEMA_VERSION", @@ -28,4 +29,7 @@ "ScoreRecord", "StudyManifest", "StudyRun", + "AdjudicationRecord", + "GraderJudgment", + "StudyScoreReport", ] diff --git a/src/conclave/evals/reporting.py b/src/conclave/evals/reporting.py new file mode 100644 index 0000000..f3ad1e0 --- /dev/null +++ b/src/conclave/evals/reporting.py @@ -0,0 +1,108 @@ +"""Deterministic JSON and Markdown rendering for exploratory eval reports.""" + +from __future__ import annotations + +import os +import tempfile +from pathlib import Path + +from .scoring import StudyScoreReport + + +def render_markdown_report(report: StudyScoreReport) -> str: + """Render a compact human report without promoting exploratory evidence.""" + + lines = [ + "# Conclave Elite Evaluation — SYNTHETIC / EXPLORATORY", + "", + f"Study: `{report.study_id}`", + "", + "**Decision status: not yet eligible.** Go / redesign / kill decisions require " + "confirmatory held-out data.", + "", + "## Outcome distribution", + "", + "| Outcome | Planned runs |", + "|---|---:|", + ] + lines.extend( + f"| {outcome} | {count} |" + for outcome, count in sorted(report.run_outcome_distribution.items()) + ) + lines.extend( + [ + "", + "## Failure-inclusive critical-error-free rates", + "", + "| Condition | Error-free / planned | Rate | 95% Wilson CI |", + "|---|---:|---:|---:|", + ] + ) + for metric in report.condition_metrics: + interval = metric.wilson_95_interval + lines.append( + f"| {metric.condition_id} | {metric.critical_error_free_count} / " + f"{metric.planned_runs} | {metric.critical_error_free_rate:.3f} | " + f"[{interval.lower:.3f}, {interval.upper:.3f}] |" + ) + lines.extend( + [ + "", + "## Paired bootstrap comparisons", + "", + "Elite-minus-baseline task-paired differences; intervals are exploratory.", + "", + "| Baseline | Difference | 95% paired bootstrap CI | Tasks |", + "|---|---:|---:|---:|", + ] + ) + lines.extend( + f"| {item.baseline_condition_id} | {item.estimate:.3f} | " + f"[{item.lower:.3f}, {item.upper:.3f}] | {item.task_count} |" + for item in report.paired_differences + ) + reliability = report.reliability + kappa = "n/a" if reliability.cohen_kappa is None else f"{reliability.cohen_kappa:.3f}" + lines.extend( + [ + "", + "## Grader reliability", + "", + f"- Cohen kappa: {kappa} ({reliability.paired_judgments} paired judgments)", + f"- Adjudication rate: {reliability.adjudication_rate:.3f} " + f"({reliability.adjudicated_disagreements}/{reliability.disagreements} disagreements)", + "", + "## Raw grader provenance", + "", + f"- Raw judgments preserved: {len(report.raw_judgments)}", + f"- Separate adjudications preserved: {len(report.adjudications)}", + "- Complete atomic records and resolved-cell provenance are available in the JSON report.", + "", + "## Go / redesign / kill", + "", + "- Go: not yet eligible", + "- Redesign: not yet eligible", + "- Kill: not yet eligible", + "", + ] + ) + return "\n".join(lines) + + +def _atomic_write(path: Path, content: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with tempfile.NamedTemporaryFile( + "w", dir=path.parent, delete=False, encoding="utf-8" + ) as handle: + handle.write(content) + temporary_path = Path(handle.name) + os.replace(temporary_path, path) + + +def write_report_bundle( + report: StudyScoreReport, *, json_path: str | Path, markdown_path: str | Path +) -> None: + """Atomically write matching machine-readable and human-readable reports.""" + + _atomic_write(Path(json_path), report.model_dump_json(indent=2) + "\n") + _atomic_write(Path(markdown_path), render_markdown_report(report)) diff --git a/src/conclave/evals/scoring.py b/src/conclave/evals/scoring.py new file mode 100644 index 0000000..297ba8f --- /dev/null +++ b/src/conclave/evals/scoring.py @@ -0,0 +1,406 @@ +"""Failure-inclusive scoring and reliability statistics for frozen eval studies.""" + +from __future__ import annotations + +import math +import random +from collections import Counter, defaultdict +from collections.abc import Mapping, Sequence + +from pydantic import Field, model_validator + +from .blinding import BlindMap +from .models import ( + ConditionId, + EvalModel, + RunOutcome, + StudyManifest, + StudyRun, +) + + +class GraderJudgment(EvalModel): + """One immutable raw grader judgment tied to one blinded or planned output.""" + + judgment_id: str = Field(min_length=1) + planned_run_id: str | None = Field(default=None, pattern=r"^run_[0-9a-f]{24}$") + opaque_output_id: str | None = Field(default=None, pattern=r"^output_[0-9a-f]{24}$") + grader_id: str = Field(min_length=1) + critical_error_free: bool + dimensions: dict[str, int] = Field(default_factory=dict) + notes: str | None = None + + @model_validator(mode="after") + def validate_one_target(self) -> GraderJudgment: + if (self.planned_run_id is None) == (self.opaque_output_id is None): + raise ValueError("exactly one of planned_run_id or opaque_output_id is required") + return self + + +class AdjudicationRecord(EvalModel): + """A separate resolution record; source judgments remain unchanged.""" + + adjudication_id: str = Field(min_length=1) + planned_run_id: str | None = Field(default=None, pattern=r"^run_[0-9a-f]{24}$") + opaque_output_id: str | None = Field(default=None, pattern=r"^output_[0-9a-f]{24}$") + critical_error_free: bool + source_judgment_ids: tuple[str, ...] = Field(min_length=1) + adjudicator_id: str = Field(min_length=1) + rationale: str = Field(min_length=1) + + @model_validator(mode="after") + def validate_one_target(self) -> AdjudicationRecord: + if (self.planned_run_id is None) == (self.opaque_output_id is None): + raise ValueError("exactly one of planned_run_id or opaque_output_id is required") + if len(set(self.source_judgment_ids)) != len(self.source_judgment_ids): + raise ValueError("source_judgment_ids must be unique") + return self + + +class RateInterval(EvalModel): + """A bounded confidence interval for a proportion.""" + + lower: float = Field(ge=0.0, le=1.0) + upper: float = Field(ge=0.0, le=1.0) + confidence_level: float = Field(gt=0.0, lt=1.0) + + +class PairedDifference(EvalModel): + """Task-paired Elite-minus-baseline bootstrap result.""" + + baseline_condition_id: ConditionId + elite_condition_id: ConditionId = "elite_full" + estimate: float = Field(ge=-1.0, le=1.0) + lower: float = Field(ge=-1.0, le=1.0) + upper: float = Field(ge=-1.0, le=1.0) + confidence_level: float = Field(gt=0.0, lt=1.0) + task_count: int = Field(ge=1) + bootstrap_seed: int + bootstrap_samples: int = Field(ge=1) + + +class ResolvedOutcome(EvalModel): + """Failure-inclusive binary outcome for one predeclared cell.""" + + planned_run_id: str = Field(pattern=r"^run_[0-9a-f]{24}$") + critical_error_free: bool + resolution: str = Field(min_length=1) + + +class ConditionMetric(EvalModel): + """Primary endpoint and execution distribution for one condition.""" + + condition_id: ConditionId + planned_runs: int = Field(ge=1) + critical_error_free_count: int = Field(ge=0) + critical_error_free_rate: float = Field(ge=0.0, le=1.0) + wilson_95_interval: RateInterval + outcome_distribution: dict[RunOutcome, int] + + +class ReliabilityMetrics(EvalModel): + """Inter-grader agreement and separate adjudication workload.""" + + grader_pair: tuple[str, str] | None = None + cohen_kappa: float | None = Field(default=None, ge=-1.0, le=1.0) + paired_judgments: int = Field(ge=0) + disagreements: int = Field(ge=0) + adjudicated_disagreements: int = Field(ge=0) + adjudication_rate: float = Field(ge=0.0, le=1.0) + + +class StudyScoreReport(EvalModel): + """Machine-readable, explicitly non-confirmatory evaluation report.""" + + study_id: str = Field(min_length=1) + evidence_classification: str = "synthetic_exploratory" + decision_eligibility: str = "not_yet_eligible" + decision_gates: dict[str, str] = Field( + default_factory=lambda: { + "go": "not_yet_eligible", + "redesign": "not_yet_eligible", + "kill": "not_yet_eligible", + } + ) + total_planned_runs: int = Field(ge=1) + run_outcome_distribution: dict[RunOutcome, int] + condition_metrics: tuple[ConditionMetric, ...] + paired_differences: tuple[PairedDifference, ...] + reliability: ReliabilityMetrics + raw_judgments: tuple[GraderJudgment, ...] + adjudications: tuple[AdjudicationRecord, ...] + resolved_outcomes: tuple[ResolvedOutcome, ...] + + +def wilson_interval(*, successes: int, trials: int, confidence_level: float = 0.95) -> RateInterval: + """Return a Wilson score interval using the fixed 95% normal quantile.""" + + if trials < 0 or successes < 0 or successes > trials: + raise ValueError("successes and trials must satisfy 0 <= successes <= trials") + if confidence_level != 0.95: + raise ValueError("only the predeclared 95% interval is supported") + if trials == 0: + return RateInterval(lower=0.0, upper=1.0, confidence_level=confidence_level) + z = 1.959963984540054 + proportion = successes / trials + denominator = 1 + z * z / trials + center = (proportion + z * z / (2 * trials)) / denominator + margin = ( + z + * math.sqrt(proportion * (1 - proportion) / trials + z * z / (4 * trials * trials)) + / denominator + ) + return RateInterval( + lower=max(0.0, center - margin), + upper=min(1.0, center + margin), + confidence_level=confidence_level, + ) + + +def _quantile(values: Sequence[float], probability: float) -> float: + index = (len(values) - 1) * probability + lower = math.floor(index) + upper = math.ceil(index) + if lower == upper: + return values[lower] + return values[lower] + (values[upper] - values[lower]) * (index - lower) + + +def paired_bootstrap_difference( + elite_by_task: Mapping[str, float], + baseline_by_task: Mapping[str, float], + *, + seed: int, + samples: int, + confidence_level: float = 0.95, + baseline_condition_id: ConditionId = "single_frontier", +) -> PairedDifference: + """Bootstrap paired task differences with a deterministic local RNG.""" + + if set(elite_by_task) != set(baseline_by_task) or not elite_by_task: + raise ValueError("elite and baseline must contain the same task IDs") + if samples < 1: + raise ValueError("samples must be positive") + if confidence_level != 0.95: + raise ValueError("only the predeclared 95% interval is supported") + differences = [ + elite_by_task[task_id] - baseline_by_task[task_id] for task_id in sorted(elite_by_task) + ] + estimate = sum(differences) / len(differences) + rng = random.Random(seed) + bootstrapped = sorted( + sum(rng.choice(differences) for _ in differences) / len(differences) for _ in range(samples) + ) + return PairedDifference( + baseline_condition_id=baseline_condition_id, + estimate=estimate, + lower=_quantile(bootstrapped, 0.025), + upper=_quantile(bootstrapped, 0.975), + confidence_level=confidence_level, + task_count=len(differences), + bootstrap_seed=seed, + bootstrap_samples=samples, + ) + + +def cohen_kappa(first: Sequence[bool], second: Sequence[bool]) -> float | None: + """Return Cohen's kappa for aligned binary ratings, or ``None`` if undefined.""" + + if not first or len(first) != len(second): + return None + observed = sum(left == right for left, right in zip(first, second, strict=True)) / len(first) + first_true = sum(first) / len(first) + second_true = sum(second) / len(second) + expected = first_true * second_true + (1 - first_true) * (1 - second_true) + if expected == 1.0: + return 1.0 if observed == 1.0 else None + return (observed - expected) / (1 - expected) + + +def _target_map(blind_map: BlindMap | None) -> dict[str, str]: + if blind_map is None: + return {} + mapping = {entry.opaque_output_id: entry.planned_run_id for entry in blind_map.entries} + if len(mapping) != len(blind_map.entries): + raise ValueError("blind map contains duplicate opaque_output_id values") + return mapping + + +def _planned_target(record: GraderJudgment | AdjudicationRecord, mapping: Mapping[str, str]) -> str: + if record.planned_run_id is not None: + return record.planned_run_id + assert record.opaque_output_id is not None + try: + return mapping[record.opaque_output_id] + except KeyError as exc: + raise ValueError("opaque target is missing from the blind map") from exc + + +def _reliability( + judgments_by_run: Mapping[str, Sequence[GraderJudgment]], + adjudicated_runs: set[str], +) -> ReliabilityMetrics: + graders = sorted({item.grader_id for values in judgments_by_run.values() for item in values}) + disagreements = sum( + len({item.critical_error_free for item in values}) > 1 + for values in judgments_by_run.values() + ) + adjudicated_disagreements = sum( + run_id in adjudicated_runs + for run_id, values in judgments_by_run.items() + if len({item.critical_error_free for item in values}) > 1 + ) + rate = adjudicated_disagreements / disagreements if disagreements else 0.0 + if len(graders) != 2: + return ReliabilityMetrics( + paired_judgments=0, + disagreements=disagreements, + adjudicated_disagreements=adjudicated_disagreements, + adjudication_rate=rate, + ) + first, second = graders + pairs = [] + for run_id in sorted(judgments_by_run): + by_grader = {item.grader_id: item.critical_error_free for item in judgments_by_run[run_id]} + if first in by_grader and second in by_grader: + pairs.append((by_grader[first], by_grader[second])) + return ReliabilityMetrics( + grader_pair=(first, second), + cohen_kappa=cohen_kappa([left for left, _ in pairs], [right for _, right in pairs]), + paired_judgments=len(pairs), + disagreements=disagreements, + adjudicated_disagreements=adjudicated_disagreements, + adjudication_rate=rate, + ) + + +def score_study( + *, + manifest: StudyManifest, + study_run: StudyRun, + raw_judgments: Sequence[GraderJudgment], + adjudications: Sequence[AdjudicationRecord] = (), + blind_map: BlindMap | None = None, + bootstrap_seed: int, + bootstrap_samples: int, +) -> StudyScoreReport: + """Score one frozen study without dropping failures or modifying raw judgments.""" + + if study_run.study_id != manifest.study_id: + raise ValueError("study run and manifest study_id values must match") + planned_by_id = {run.planned_run_id: run for run in manifest.planned_runs} + run_by_id = {record.planned_run_id: record for record in study_run.records} + if set(run_by_id) != set(planned_by_id) or len(run_by_id) != len(study_run.records): + raise ValueError("study run must cover every planned_run_id exactly once") + if len({item.judgment_id for item in raw_judgments}) != len(raw_judgments): + raise ValueError("judgment_id values must be unique") + if len({item.adjudication_id for item in adjudications}) != len(adjudications): + raise ValueError("adjudication_id values must be unique") + + mapping = _target_map(blind_map) + judgments_by_run: dict[str, list[GraderJudgment]] = defaultdict(list) + judgment_ids = {item.judgment_id for item in raw_judgments} + seen_grader_targets: set[tuple[str, str]] = set() + for judgment in raw_judgments: + run_id = _planned_target(judgment, mapping) + if run_id not in planned_by_id: + raise ValueError("grader judgment references an unplanned run") + grader_target = (run_id, judgment.grader_id) + if grader_target in seen_grader_targets: + raise ValueError("a grader may provide only one judgment per planned run") + seen_grader_targets.add(grader_target) + judgments_by_run[run_id].append(judgment) + + adjudication_by_run: dict[str, AdjudicationRecord] = {} + for adjudication in adjudications: + run_id = _planned_target(adjudication, mapping) + if run_id not in planned_by_id: + raise ValueError("adjudication references an unplanned run") + if run_id in adjudication_by_run: + raise ValueError("only one adjudication is allowed per planned run") + if not set(adjudication.source_judgment_ids).issubset(judgment_ids): + raise ValueError("adjudication references an unknown source judgment") + expected_source_ids = {item.judgment_id for item in judgments_by_run[run_id]} + if not set(adjudication.source_judgment_ids).issubset(expected_source_ids): + raise ValueError("adjudication source judgments must target the same run") + adjudication_by_run[run_id] = adjudication + + resolved: list[ResolvedOutcome] = [] + for planned in manifest.planned_runs: + run_record = run_by_id[planned.planned_run_id] + values = [item.critical_error_free for item in judgments_by_run[planned.planned_run_id]] + if run_record.outcome != "success": + value, resolution = False, "automatic_non_success" + elif planned.planned_run_id in adjudication_by_run: + value = adjudication_by_run[planned.planned_run_id].critical_error_free + resolution = "adjudicated" + elif values and len(set(values)) == 1: + value, resolution = values[0], "unanimous_raw" + elif values: + value, resolution = False, "unresolved_disagreement" + else: + value, resolution = False, "ungraded_success" + resolved.append( + ResolvedOutcome( + planned_run_id=planned.planned_run_id, + critical_error_free=value, + resolution=resolution, + ) + ) + + resolved_by_id = {item.planned_run_id: item for item in resolved} + condition_metrics = [] + task_condition_values: dict[ConditionId, dict[str, list[bool]]] = defaultdict( + lambda: defaultdict(list) + ) + for condition_id in dict.fromkeys(run.condition_id for run in manifest.planned_runs): + planned = [run for run in manifest.planned_runs if run.condition_id == condition_id] + successes = sum(resolved_by_id[run.planned_run_id].critical_error_free for run in planned) + distribution = dict( + sorted(Counter(run_by_id[run.planned_run_id].outcome for run in planned).items()) + ) + condition_metrics.append( + ConditionMetric( + condition_id=condition_id, + planned_runs=len(planned), + critical_error_free_count=successes, + critical_error_free_rate=successes / len(planned), + wilson_95_interval=wilson_interval(successes=successes, trials=len(planned)), + outcome_distribution=distribution, + ) + ) + for run in planned: + task_condition_values[condition_id][run.task_id].append( + resolved_by_id[run.planned_run_id].critical_error_free + ) + + means = { + condition_id: { + task_id: sum(values) / len(values) for task_id, values in task_values.items() + } + for condition_id, task_values in task_condition_values.items() + } + paired = tuple( + paired_bootstrap_difference( + means["elite_full"], + means[condition_id], + seed=bootstrap_seed, + samples=bootstrap_samples, + baseline_condition_id=condition_id, + ) + for condition_id in means + if condition_id != "elite_full" + ) + return StudyScoreReport( + study_id=manifest.study_id, + total_planned_runs=len(manifest.planned_runs), + run_outcome_distribution=dict( + sorted(Counter(item.outcome for item in study_run.records).items()) + ), + condition_metrics=tuple(condition_metrics), + paired_differences=paired, + reliability=_reliability(judgments_by_run, set(adjudication_by_run)), + raw_judgments=tuple(raw_judgments), + adjudications=tuple(adjudications), + resolved_outcomes=tuple(resolved), + ) diff --git a/tests/evals/test_reporting.py b/tests/evals/test_reporting.py new file mode 100644 index 0000000..c548fd2 --- /dev/null +++ b/tests/evals/test_reporting.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import json + +from conclave.evals.models import PublicTask, RunRecord, StudyRun +from conclave.evals.protocols import CONDITION_IDS, build_study_manifest +from conclave.evals.reporting import render_markdown_report, write_report_bundle +from conclave.evals.scoring import GraderJudgment, score_study + + +def _synthetic_report(): + tasks = [PublicTask(task_id="task-a", prompt="Choose.")] + manifest = build_study_manifest( + study_id="synthetic-report", + tasks=tasks, + replicates=1, + seed=7, + output_token_budgets={condition_id: 100 for condition_id in CONDITION_IDS}, + ) + records = tuple( + RunRecord( + planned_run_id=planned.planned_run_id, + outcome="success", + output="synthetic answer", + ) + for planned in manifest.planned_runs + ) + study_run = StudyRun( + study_id=manifest.study_id, + records=records, + total_planned_runs=len(records), + outcome_counts={"success": len(records)}, + total_completion_tokens=0, + total_latency_ms=0, + ) + judgments = tuple( + GraderJudgment( + judgment_id=f"judgment-{index}", + planned_run_id=record.planned_run_id, + grader_id="synthetic-grader", + critical_error_free=index % 2 == 0, + ) + for index, record in enumerate(records) + ) + return score_study( + manifest=manifest, + study_run=study_run, + raw_judgments=judgments, + bootstrap_seed=11, + bootstrap_samples=100, + ) + + +def test_markdown_report_is_explicitly_exploratory_and_complete() -> None: + markdown = render_markdown_report(_synthetic_report()) + + assert "SYNTHETIC / EXPLORATORY" in markdown + assert "not yet eligible" in markdown.lower() + assert "Outcome distribution" in markdown + assert "95% Wilson CI" in markdown + assert "Paired bootstrap" in markdown + assert "Raw grader provenance" in markdown + assert "Go / redesign / kill" in markdown + assert "confirmatory" in markdown.lower() + + +def test_report_bundle_writes_machine_json_and_human_markdown(tmp_path) -> None: + report = _synthetic_report() + json_path = tmp_path / "nested" / "report.json" + markdown_path = tmp_path / "nested" / "report.md" + + write_report_bundle(report, json_path=json_path, markdown_path=markdown_path) + + payload = json.loads(json_path.read_text()) + assert payload["study_id"] == "synthetic-report" + assert payload["raw_judgments"] + assert markdown_path.read_text().startswith("# Conclave Elite Evaluation") diff --git a/tests/evals/test_scoring.py b/tests/evals/test_scoring.py new file mode 100644 index 0000000..21a7fd7 --- /dev/null +++ b/tests/evals/test_scoring.py @@ -0,0 +1,242 @@ +from __future__ import annotations + +import json + +import pytest +from pydantic import ValidationError + +from conclave.evals.blinding import BlindMap, BlindMapEntry +from conclave.evals.models import PublicTask, RunRecord, StudyRun +from conclave.evals.protocols import CONDITION_IDS, build_study_manifest +from conclave.evals.scoring import ( + AdjudicationRecord, + GraderJudgment, + cohen_kappa, + paired_bootstrap_difference, + score_study, + wilson_interval, +) + + +def _study(*, task_count: int = 2): + tasks = [PublicTask(task_id=f"task-{index}", prompt="Choose.") for index in range(task_count)] + manifest = build_study_manifest( + study_id="synthetic-study", + tasks=tasks, + replicates=1, + seed=17, + output_token_budgets={condition_id: 100 for condition_id in CONDITION_IDS}, + ) + records = tuple( + RunRecord(planned_run_id=planned.planned_run_id, outcome="success", output="answer") + for planned in manifest.planned_runs + ) + run = StudyRun( + study_id=manifest.study_id, + records=records, + total_planned_runs=len(records), + outcome_counts={"success": len(records)}, + total_completion_tokens=0, + total_latency_ms=0, + ) + return manifest, run + + +def test_atomic_judgments_accept_exactly_one_opaque_or_planned_target() -> None: + planned = GraderJudgment( + judgment_id="judgment-planned", + planned_run_id="run_" + "a" * 24, + grader_id="grader-a", + critical_error_free=True, + ) + opaque = GraderJudgment( + judgment_id="judgment-opaque", + opaque_output_id="output_" + "b" * 24, + grader_id="grader-b", + critical_error_free=False, + ) + + assert planned.planned_run_id and planned.opaque_output_id is None + assert opaque.opaque_output_id and opaque.planned_run_id is None + with pytest.raises(ValidationError, match="exactly one"): + GraderJudgment( + judgment_id="bad", + planned_run_id="run_" + "a" * 24, + opaque_output_id="output_" + "b" * 24, + grader_id="grader-a", + critical_error_free=True, + ) + with pytest.raises(ValidationError, match="exactly one"): + GraderJudgment( + judgment_id="bad", + grader_id="grader-a", + critical_error_free=True, + ) + + +def test_scoring_preserves_raw_records_and_keeps_adjudication_separate() -> None: + manifest, run = _study(task_count=1) + planned_id = manifest.planned_runs[0].planned_run_id + opaque_id = "output_" + "c" * 24 + raw = ( + GraderJudgment( + judgment_id="judgment-a", + opaque_output_id=opaque_id, + grader_id="grader-a", + critical_error_free=True, + notes="raw-a", + ), + GraderJudgment( + judgment_id="judgment-b", + opaque_output_id=opaque_id, + grader_id="grader-b", + critical_error_free=False, + notes="raw-b", + ), + ) + adjudication = AdjudicationRecord( + adjudication_id="adjudication-1", + planned_run_id=planned_id, + critical_error_free=True, + source_judgment_ids=("judgment-a", "judgment-b"), + adjudicator_id="adjudicator", + rationale="Resolved from rubric evidence.", + ) + blind_map = BlindMap( + entries=(BlindMapEntry(opaque_output_id=opaque_id, planned_run_id=planned_id),) + ) + + report = score_study( + manifest=manifest, + study_run=run, + raw_judgments=raw, + adjudications=(adjudication,), + blind_map=blind_map, + bootstrap_seed=9, + bootstrap_samples=100, + ) + + assert report.raw_judgments == raw + assert report.adjudications == (adjudication,) + assert report.raw_judgments[0].critical_error_free is True + assert report.raw_judgments[1].critical_error_free is False + assert report.resolved_outcomes[0].critical_error_free is True + assert report.reliability.disagreements == 1 + assert report.reliability.adjudicated_disagreements == 1 + assert report.reliability.adjudication_rate == 1.0 + + +def test_critical_error_free_rate_keeps_all_planned_runs_in_denominator() -> None: + manifest, successful = _study(task_count=1) + first_condition = manifest.planned_runs[0].condition_id + failed_id = manifest.planned_runs[0].planned_run_id + records = tuple( + RunRecord( + planned_run_id=record.planned_run_id, + outcome="failed" if record.planned_run_id == failed_id else record.outcome, + output=None if record.planned_run_id == failed_id else record.output, + ) + for record in successful.records + ) + run = StudyRun( + study_id=manifest.study_id, + records=records, + total_planned_runs=len(records), + outcome_counts={"failed": 1, "success": len(records) - 1}, + total_completion_tokens=0, + total_latency_ms=0, + ) + judgments = tuple( + GraderJudgment( + judgment_id=f"judgment-{index}", + planned_run_id=record.planned_run_id, + grader_id="grader-a", + critical_error_free=True, + ) + for index, record in enumerate(records) + if record.outcome == "success" + ) + + report = score_study( + manifest=manifest, + study_run=run, + raw_judgments=judgments, + bootstrap_seed=3, + bootstrap_samples=100, + ) + metric = next(item for item in report.condition_metrics if item.condition_id == first_condition) + + assert metric.planned_runs == 1 + assert metric.critical_error_free_count == 0 + assert metric.critical_error_free_rate == 0.0 + assert report.resolved_outcomes[0].critical_error_free is False + assert report.resolved_outcomes[0].resolution == "automatic_non_success" + + +def test_wilson_interval_has_expected_95_percent_bounds() -> None: + interval = wilson_interval(successes=5, trials=10) + + assert interval.confidence_level == 0.95 + assert interval.lower == pytest.approx(0.2366, abs=0.0001) + assert interval.upper == pytest.approx(0.7634, abs=0.0001) + assert wilson_interval(successes=0, trials=0).lower == 0.0 + assert wilson_interval(successes=0, trials=0).upper == 1.0 + + +def test_seeded_paired_bootstrap_is_deterministic_and_task_paired() -> None: + elite = {"a": 1.0, "b": 1.0, "c": 0.0, "d": 1.0} + baseline = {"a": 0.0, "b": 1.0, "c": 0.0, "d": 0.0} + + first = paired_bootstrap_difference( + elite, baseline, seed=42, samples=1000, confidence_level=0.95 + ) + second = paired_bootstrap_difference( + elite, baseline, seed=42, samples=1000, confidence_level=0.95 + ) + + assert first == second + assert first.task_count == 4 + assert first.estimate == 0.5 + assert first.lower <= first.estimate <= first.upper + with pytest.raises(ValueError, match="same task IDs"): + paired_bootstrap_difference({"a": 1.0}, {"b": 0.0}, seed=1, samples=10) + + +def test_cohen_kappa_and_adjudication_rate_are_reported() -> None: + assert cohen_kappa([True, True, False, False], [True, False, False, False]) == pytest.approx( + 0.5 + ) + assert cohen_kappa([True], [True, False]) is None + assert cohen_kappa([], []) is None + + +def test_report_contract_serializes_raw_provenance_without_mutation() -> None: + manifest, run = _study(task_count=1) + judgments = tuple( + GraderJudgment( + judgment_id=f"judgment-{index}", + planned_run_id=record.planned_run_id, + grader_id="grader-a", + critical_error_free=True, + notes="synthetic fixture judgment", + ) + for index, record in enumerate(run.records) + ) + + report = score_study( + manifest=manifest, + study_run=run, + raw_judgments=judgments, + bootstrap_seed=5, + bootstrap_samples=100, + ) + payload = json.loads(report.model_dump_json()) + + assert payload["evidence_classification"] == "synthetic_exploratory" + assert payload["decision_eligibility"] == "not_yet_eligible" + assert payload["raw_judgments"][0]["notes"] == "synthetic fixture judgment" + assert payload["decision_gates"] == { + "go": "not_yet_eligible", + "redesign": "not_yet_eligible", + "kill": "not_yet_eligible", + } From f1c61efc0b43d7dcc24df5a3e5d550fe9c32b660 Mon Sep 17 00:00:00 2001 From: ernestprovo23 Date: Fri, 17 Jul 2026 17:19:17 -0400 Subject: [PATCH 6/7] feat(evals): expose offline evaluation workflow --- DOCUMENTATION_INDEX.md | 6 + README.md | 13 +++ SYSTEM_CONTEXT_DIAGRAM.md | 8 ++ src/conclave/cli.py | 2 + src/conclave/eval_cli.py | 208 +++++++++++++++++++++++++++++++++++ tests/evals/test_eval_cli.py | 190 ++++++++++++++++++++++++++++++++ 6 files changed, 427 insertions(+) create mode 100644 src/conclave/eval_cli.py create mode 100644 tests/evals/test_eval_cli.py diff --git a/DOCUMENTATION_INDEX.md b/DOCUMENTATION_INDEX.md index 82e7cb0..667943a 100644 --- a/DOCUMENTATION_INDEX.md +++ b/DOCUMENTATION_INDEX.md @@ -30,6 +30,8 @@ the canonical authority spec on top of those. | Doc | Path | Purpose | |-----|------|---------| | **Decision Quality Roadmap** | [`docs/plans/2026-07-17-decision-quality-roadmap.md`](docs/plans/2026-07-17-decision-quality-roadmap.md) | H0-H4 prove-it roadmap: Elite correctness, randomized ablations, source-grounded Decision Briefs, buyer pull, outcome learning, old-backlog disposition, demand gates, and kill criteria. | +| **H1 Evaluation Design** | [`docs/plans/2026-07-17-h1-budget-matched-evaluation-design.md`](docs/plans/2026-07-17-h1-budget-matched-evaluation-design.md) | DSE-708 experimental boundary, six frozen conditions, replay, blinding, and failure-inclusive analysis. | +| **H1 Evaluation Plan** | [`docs/plans/2026-07-17-h1-budget-matched-evaluation.md`](docs/plans/2026-07-17-h1-budget-matched-evaluation.md) | TDD delivery plan for the offline, budget-matched evaluation substrate. | --- @@ -60,6 +62,8 @@ Package root: `src/conclave/` (installed as the `conclave` package; console scri | Config | [`src/conclave/config.py`](src/conclave/config.py) | Loads/merges `~/.conclave/config.yml` over defaults; resolves model ids and named/CSV councils; parses the `endpoints:` section (custom OpenAI-compatible providers). | | Models | [`src/conclave/models.py`](src/conclave/models.py) | Stable Pydantic contract, including `EliteResult` phase artifacts and backward-compatible `CouncilResult.elite`. | | CLI | [`src/conclave/cli.py`](src/conclave/cli.py) | Six-mode `conclave ask` plus `providers`; Elite adds a phase summary, full JSON, incomplete exit 1, and explicit stream rejection. | +| Experimental evals | [`src/conclave/evals/`](src/conclave/evals/) | Versioned DSE-708 planning, strict replay, failure-inclusive running, blinding, scoring, and exploratory reports; no product-quality claim. | +| Eval CLI | [`src/conclave/eval_cli.py`](src/conclave/eval_cli.py) | Offline-only `conclave eval plan/run/blind/report`; `run` validates replay artifacts and cannot call providers. | | Logging | [`src/conclave/logging.py`](src/conclave/logging.py) | Logger factory; stderr; verbosity via `CONCLAVE_LOG_LEVEL` (default `WARNING`). | ## Tests @@ -74,6 +78,7 @@ Package root: `src/conclave/` (installed as the `conclave` package; console scri | Provider highway tests | [`tests/test_providers.py`](tests/test_providers.py) | `resolve_adapter` (built-in prefixes, per-provider URLs, custom endpoints, unknown-prefix raise), end-to-end `call_model`, and `redact()` (bearer/`sk-`/env-var-value/`x-api-key` scrubbing; pre-redacted provider errors). | | Registry/config tests | [`tests/test_registry_config.py`](tests/test_registry_config.py) | Name resolution, key-presence logic, config merge. | | CLI tests | [`tests/test_cli.py`](tests/test_cli.py) | Typer `CliRunner`: exit-code contract (0 success / 1 zero-usable-answers / 2 usage error), `--json` payload + exit code, human renderers per mode, `providers` table never prints secrets, aclose lifecycle. | +| Eval tests | [`tests/evals/`](tests/evals/) | Frozen matrix, replay fail-closed behavior, runner denominators, blinding, scoring, reporting, and offline CLI. | | Transport tests | [`tests/test_transport.py`](tests/test_transport.py) | `post_json` via httpx `MockTransport`: success/error-status/non-JSON fallback, timeout & connect/HTTP errors → `TransportError` (key never leaks), client reuse/pooling, aclose idempotency. | | Streaming tests | [`tests/test_streaming.py`](tests/test_streaming.py) | Per-adapter SSE via `MockTransport` (openai-compat/anthropic/gemini): incremental chunks + assembled answer == concatenation == buffered result; mid-stream malformed-frame/connection-drop/non-2xx → error set with partial text preserved (never raises); key redaction in stream errors; buffered `ask()` never opens a stream; `Council.ask_stream` interleaving + terminal `done` shape; CLI `--stream` smoke + exit-code contract + debate rejection; `--stream` + cache one-shot replay. | | Logging tests | [`tests/test_logging.py`](tests/test_logging.py) | `CONCLAVE_LOG_LEVEL` resolution (default `WARNING`, case-insensitive, unknown → `WARNING`), factory contract, one-shot configuration. | @@ -107,6 +112,7 @@ Run: `pytest` (config in `pyproject.toml`, `asyncio_mode = "auto"`). | Date | Change | |------|--------| +| 2026-07-17 | Added the experimental DSE-708 offline evaluation substrate and `conclave eval` artifact workflow; no live study or quality claim. | | 2026-07-17 | Reframed the product roadmap around empirically proven decision quality: narrowed current claims to execution traceability, identified answer IDs as internal provenance rather than external evidence, gated Elite merge on H0 correctness, and added H1-H4 evidence, buyer, and outcome gates. | | 2026-07-17 | Documented the implemented-but-unreleased Elite Decision Protocol: fixed three-success gates, initial/claim-audit/revision artifacts, existing final verdict, phased receipts, failure semantics, cost/latency tradeoff, and no streaming. | | 2026-07-13 | **Manifest-on-every-result invariant fix + doc reconciliation.** `ModelHarnessManifest` now attaches on all five modes (was `None` for `debate`/`adversarial`/`vote`) via a single-site fix at `Council._cached_run` → new `_ensure_manifest`; regression tests in `tests/test_manifest_all_modes.py`. Docs corrected so the manifest-on-every-result claim is accurate and the `vote` mode is documented as **shipped** (CAC-09 / #3), not "absorbed" — across README, `SYSTEM_CONTEXT_DIAGRAM.md`, PDD (§1/§3/§4a/§8/§9/§12), and `CHANGELOG.md` (`[Unreleased]`). Verdict scope unchanged (still `synthesize`/`ask`-only; not layered on `adversarial`). | diff --git a/README.md b/README.md index 77bc29b..f275334 100644 --- a/README.md +++ b/README.md @@ -149,6 +149,19 @@ a future release explicitly includes it. of a council defined in your config (see below). The built-in `default` council is all known providers. +### Experimental offline evaluation (DSE-708) + +The source branch also exposes an H1 evidence harness. It makes no decision-quality claim and +never calls providers: `run` only validates an existing replay artifact against its frozen +manifest. Grader output and the restricted blind map must remain separate. + +```bash +conclave eval plan tasks.json manifest.json --study-id pilot --replicates 2 --seed 19 --max-output-tokens 1200 +conclave eval run manifest.json validated-run.json --replay-artifact recorded-run.json +conclave eval blind validated-run.json grader.json restricted-map.json --seed 23 +conclave eval report manifest.json validated-run.json judgments.json report.json report.md --bootstrap-seed 29 +``` + Add `--stream` to render member (and synthesizer) tokens live as they arrive (`synthesize`/`raw` modes only): diff --git a/SYSTEM_CONTEXT_DIAGRAM.md b/SYSTEM_CONTEXT_DIAGRAM.md index b3a7159..f390891 100644 --- a/SYSTEM_CONTEXT_DIAGRAM.md +++ b/SYSTEM_CONTEXT_DIAGRAM.md @@ -32,6 +32,8 @@ flowchart TB council["Council orchestrator
fan_out · synthesize_blocks · skip-no-key (council.py)"] modes["Deliberation modes
debate · adversarial · vote (modes.py + prompts.py)"] elite["Elite (UNRELEASED)
initial → claim audit → revision
fixed 3-success gate each phase"] + evalcli["Experimental eval CLI (DSE-708)
plan · replay validation · blind · report
OFFLINE ONLY"] + evalartifacts["Versioned eval artifacts
manifest · run · grader set + separate map
exploratory report"] registry["Registry · name to model-id
key PRESENCE only, never values (registry.py)"] config["Config loader · custom endpoints (config.py)"] models["Result contract · CouncilResult v2
answers · verdict · consensus · manifest (models.py)"] @@ -65,6 +67,8 @@ flowchart TB end user -->|"prompt + council + mode"| cli + user -->|"public tasks + offline artifacts"| evalcli + evalcli --> evalartifacts user -->|"import"| lib warden -.->|"imports at DEV time only · NOT a runtime dep"| lib @@ -123,6 +127,10 @@ flowchart TB - **Two entry points, one core.** The CLI (`cli.py`) and the library API (`from conclave import Council`) are both thin drivers over the same `Council` orchestrator. There is no behavior in the CLI that the library can't reach. +- **The DSE-708 eval lane is experimental and offline.** `conclave eval` freezes manifests, + validates pre-recorded run artifacts, blinds outputs into a grader set plus separate identity + map, and writes exploratory reports. It has no edge to the provider highway: live execution is + fail-closed, and these artifacts support measurement rather than a decision-quality claim. - **mcp-warden is dashed and dev-time.** The dotted edge from `mcp-warden` to the library is deliberate: warden imports conclave **only at design/eval time**. conclave is stochastic and must never sit in warden's deterministic runtime decision path. See PDD diff --git a/src/conclave/cli.py b/src/conclave/cli.py index ec5b8eb..169c27a 100644 --- a/src/conclave/cli.py +++ b/src/conclave/cli.py @@ -21,6 +21,7 @@ from . import __version__ from .config import load_config from .council import Council +from .eval_cli import app as eval_app from .models import CouncilResult, StreamEvent from .registry import DEFAULT_MODELS, key_present, key_source @@ -28,6 +29,7 @@ add_completion=False, help="Bring-your-own-keys multi-model council. Fan a prompt to N models.", ) +app.add_typer(eval_app, name="eval") console = Console() err_console = Console(stderr=True) diff --git a/src/conclave/eval_cli.py b/src/conclave/eval_cli.py new file mode 100644 index 0000000..9ecc240 --- /dev/null +++ b/src/conclave/eval_cli.py @@ -0,0 +1,208 @@ +"""Experimental, offline-only command surface for the H1 evaluation harness.""" + +from __future__ import annotations + +import json +import os +import tempfile +from collections import Counter +from pathlib import Path +from typing import Annotated, Any + +import typer +from pydantic import ValidationError + +from .evals.blinding import BlindMap, blind_run_records +from .evals.dataset import load_public_tasks +from .evals.models import EVAL_CONDITION_IDS, StudyManifest, StudyRun +from .evals.protocols import build_study_manifest +from .evals.reporting import write_report_bundle +from .evals.runner import validate_run_records +from .evals.scoring import AdjudicationRecord, GraderJudgment, score_study + +app = typer.Typer( + add_completion=False, + help=( + "Experimental H1 evaluation tools (DSE-708). Offline artifacts only; " + "these commands never call providers." + ), +) + + +def _read_json(path: Path) -> Any: + with path.open(encoding="utf-8") as handle: + return json.load(handle) + + +def _atomic_json(path: Path, value: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + content = json.dumps(value, indent=2, sort_keys=True) + "\n" + with tempfile.NamedTemporaryFile( + "w", dir=path.parent, delete=False, encoding="utf-8" + ) as handle: + handle.write(content) + temporary_path = Path(handle.name) + os.replace(temporary_path, path) + + +def _abort(message: str) -> None: + typer.echo(f"Error: {message}", err=True) + raise typer.Exit(code=2) + + +def _validated(model_type, path: Path): + try: + return model_type.model_validate(_read_json(path)) + except (OSError, json.JSONDecodeError, ValidationError, ValueError) as exc: + _abort(f"invalid {path}: {exc}") + + +def _validate_run_summary(study_run: StudyRun) -> None: + records = study_run.records + expected_counts = dict(sorted(Counter(record.outcome for record in records).items())) + expected_tokens = sum(record.completion_tokens or 0 for record in records) + expected_latency = sum(record.latency_ms or 0.0 for record in records) + if ( + study_run.total_planned_runs != len(records) + or study_run.outcome_counts != expected_counts + or study_run.total_completion_tokens != expected_tokens + or abs(study_run.total_latency_ms - expected_latency) > 1e-9 + ): + _abort("run artifact summary does not match its immutable records") + + +@app.command("plan") +def plan_command( + public_tasks: Annotated[Path, typer.Argument(exists=True, readable=True)], + output: Annotated[Path, typer.Argument()], + study_id: Annotated[str, typer.Option("--study-id")], + replicates: Annotated[int, typer.Option("--replicates", min=1)] = 1, + seed: Annotated[int, typer.Option("--seed")] = 0, + max_output_tokens: Annotated[int, typer.Option("--max-output-tokens", min=1)] = 1000, +) -> None: + """Freeze a complete, equal-budget study manifest from public tasks.""" + + try: + tasks = load_public_tasks(public_tasks) + manifest = build_study_manifest( + study_id=study_id, + tasks=tasks, + replicates=replicates, + seed=seed, + output_token_budgets={ + condition_id: max_output_tokens for condition_id in EVAL_CONDITION_IDS + }, + ) + except (OSError, json.JSONDecodeError, ValidationError, ValueError) as exc: + _abort(f"could not build manifest: {exc}") + _atomic_json(output, manifest.model_dump(mode="json")) + typer.echo(f"Wrote {len(manifest.planned_runs)} predeclared runs to {output}") + + +@app.command("run") +def run_command( + manifest_path: Annotated[Path, typer.Argument(exists=True, readable=True)], + output: Annotated[Path, typer.Argument()], + replay_artifact: Annotated[ + Path | None, + typer.Option( + "--replay-artifact", + exists=True, + readable=True, + help="Existing offline StudyRun artifact to validate. Live execution is unavailable.", + ), + ] = None, +) -> None: + """Validate an offline replay artifact; live provider execution is disabled.""" + + if replay_artifact is None: + _abort( + "an offline replay artifact is required (--replay-artifact); " + "live provider execution is disabled in DSE-708" + ) + manifest = _validated(StudyManifest, manifest_path) + study_run = _validated(StudyRun, replay_artifact) + _validate_run_summary(study_run) + if study_run.study_id != manifest.study_id: + _abort("replay artifact study_id does not match the manifest") + try: + validate_run_records(manifest, study_run.records) + except ValueError as exc: + _abort(f"replay artifact does not cover the frozen manifest: {exc}") + if study_run.total_planned_runs != len(manifest.planned_runs): + _abort("replay artifact total_planned_runs does not match the manifest") + _atomic_json(output, study_run.model_dump(mode="json")) + typer.echo(f"Validated offline replay artifact and wrote {output}") + + +@app.command("blind") +def blind_command( + run_path: Annotated[Path, typer.Argument(exists=True, readable=True)], + grader_output: Annotated[Path, typer.Argument()], + blind_map_output: Annotated[Path, typer.Argument()], + seed: Annotated[int, typer.Option("--seed")], +) -> None: + """Write grader-visible outputs and a physically separate restricted map.""" + + if grader_output.resolve() == blind_map_output.resolve(): + _abort("grader output and blind map must use different paths") + study_run = _validated(StudyRun, run_path) + _validate_run_summary(study_run) + try: + blinded, blind_map = blind_run_records(study_run.records, seed=seed) + except ValueError as exc: + _abort(f"could not blind run artifact: {exc}") + _atomic_json(grader_output, blinded.model_dump(mode="json")) + _atomic_json(blind_map_output, blind_map.model_dump(mode="json")) + typer.echo(f"Wrote {len(blinded.outputs)} blinded outputs and separate map") + + +def _records(path: Path | None, key: str, model_type) -> tuple: + if path is None: + return () + try: + payload = _read_json(path) + values = payload[key] + return tuple(model_type.model_validate(value) for value in values) + except (OSError, KeyError, TypeError, json.JSONDecodeError, ValidationError, ValueError) as exc: + _abort(f"invalid {key} artifact {path}: {exc}") + + +@app.command("report") +def report_command( + manifest_path: Annotated[Path, typer.Argument(exists=True, readable=True)], + run_path: Annotated[Path, typer.Argument(exists=True, readable=True)], + judgments_path: Annotated[Path, typer.Argument(exists=True, readable=True)], + json_output: Annotated[Path, typer.Argument()], + markdown_output: Annotated[Path, typer.Argument()], + bootstrap_seed: Annotated[int, typer.Option("--bootstrap-seed")], + adjudications_path: Annotated[ + Path | None, typer.Option("--adjudications", exists=True, readable=True) + ] = None, + blind_map_path: Annotated[ + Path | None, typer.Option("--blind-map", exists=True, readable=True) + ] = None, + bootstrap_samples: Annotated[int, typer.Option("--bootstrap-samples", min=1)] = 1000, +) -> None: + """Score frozen artifacts and write exploratory JSON plus Markdown reports.""" + + manifest = _validated(StudyManifest, manifest_path) + study_run = _validated(StudyRun, run_path) + _validate_run_summary(study_run) + judgments = _records(judgments_path, "judgments", GraderJudgment) + adjudications = _records(adjudications_path, "adjudications", AdjudicationRecord) + blind_map = _validated(BlindMap, blind_map_path) if blind_map_path is not None else None + try: + report = score_study( + manifest=manifest, + study_run=study_run, + raw_judgments=judgments, + adjudications=adjudications, + blind_map=blind_map, + bootstrap_seed=bootstrap_seed, + bootstrap_samples=bootstrap_samples, + ) + except (ValidationError, ValueError) as exc: + _abort(f"could not score study: {exc}") + write_report_bundle(report, json_path=json_output, markdown_path=markdown_output) + typer.echo(f"Wrote exploratory report to {json_output} and {markdown_output}") diff --git a/tests/evals/test_eval_cli.py b/tests/evals/test_eval_cli.py new file mode 100644 index 0000000..2c894ce --- /dev/null +++ b/tests/evals/test_eval_cli.py @@ -0,0 +1,190 @@ +from __future__ import annotations + +import json + +from typer.testing import CliRunner + +from conclave.cli import app +from conclave.evals.models import PublicTask, RunRecord, StudyRun +from conclave.evals.protocols import CONDITION_IDS, build_study_manifest + +runner = CliRunner() + + +def _write_json(path, payload) -> None: + path.write_text(json.dumps(payload), encoding="utf-8") + + +def _study_artifacts(tmp_path): + tasks = [PublicTask(task_id="task-a", prompt="Choose one.")] + tasks_path = tmp_path / "tasks.json" + _write_json( + tasks_path, + { + "schema_version": "conclave_eval_v1", + "tasks": [task.model_dump(mode="json") for task in tasks], + }, + ) + manifest = build_study_manifest( + study_id="offline-cli", + tasks=tasks, + replicates=1, + seed=7, + output_token_budgets={condition_id: 100 for condition_id in CONDITION_IDS}, + ) + manifest_path = tmp_path / "manifest.json" + _write_json(manifest_path, manifest.model_dump(mode="json")) + records = tuple( + RunRecord(planned_run_id=item.planned_run_id, outcome="success", output="fixture") + for item in manifest.planned_runs + ) + study_run = StudyRun( + study_id=manifest.study_id, + records=records, + total_planned_runs=len(records), + outcome_counts={"success": len(records)}, + total_completion_tokens=0, + total_latency_ms=0, + ) + run_path = tmp_path / "run.json" + _write_json(run_path, study_run.model_dump(mode="json")) + return tasks_path, manifest_path, run_path, manifest + + +def test_eval_plan_writes_complete_manifest_atomically(tmp_path) -> None: + tasks_path, _, _, _ = _study_artifacts(tmp_path) + output = tmp_path / "nested" / "manifest.json" + + result = runner.invoke( + app, + [ + "eval", + "plan", + str(tasks_path), + str(output), + "--study-id", + "cli-plan", + "--replicates", + "2", + "--seed", + "19", + "--max-output-tokens", + "120", + ], + ) + + assert result.exit_code == 0, result.output + payload = json.loads(output.read_text()) + assert payload["study_id"] == "cli-plan" + assert payload["replicates"] == 2 + assert len(payload["planned_runs"]) == 12 + assert {item["max_output_tokens"] for item in payload["planned_runs"]} == {120} + + +def test_eval_run_rejects_live_and_validates_offline_replay_artifact(tmp_path) -> None: + _, manifest_path, run_path, _ = _study_artifacts(tmp_path) + output = tmp_path / "validated-run.json" + + rejected = runner.invoke(app, ["eval", "run", str(manifest_path), str(output)]) + assert rejected.exit_code == 2 + assert "offline replay artifact" in rejected.output.lower() + + accepted = runner.invoke( + app, + [ + "eval", + "run", + str(manifest_path), + str(output), + "--replay-artifact", + str(run_path), + ], + ) + assert accepted.exit_code == 0, accepted.output + assert json.loads(output.read_text())["total_planned_runs"] == 6 + + tampered = json.loads(run_path.read_text()) + tampered["outcome_counts"] = {"success": 5} + tampered_path = tmp_path / "tampered-run.json" + _write_json(tampered_path, tampered) + rejected_tamper = runner.invoke( + app, + [ + "eval", + "run", + str(manifest_path), + str(output), + "--replay-artifact", + str(tampered_path), + ], + ) + assert rejected_tamper.exit_code == 2 + assert "summary" in rejected_tamper.output.lower() + + +def test_eval_blind_writes_separate_grader_and_identity_artifacts(tmp_path) -> None: + _, _, run_path, _ = _study_artifacts(tmp_path) + grader_path = tmp_path / "grader.json" + map_path = tmp_path / "restricted" / "blind-map.json" + + result = runner.invoke( + app, + [ + "eval", + "blind", + str(run_path), + str(grader_path), + str(map_path), + "--seed", + "23", + ], + ) + + assert result.exit_code == 0, result.output + grader_payload = json.loads(grader_path.read_text()) + map_payload = json.loads(map_path.read_text()) + assert "planned_run_id" not in grader_path.read_text() + assert len(grader_payload["outputs"]) == len(map_payload["entries"]) == 6 + + +def test_eval_report_writes_json_and_markdown_from_atomic_inputs(tmp_path) -> None: + _, manifest_path, run_path, manifest = _study_artifacts(tmp_path) + judgments_path = tmp_path / "judgments.json" + _write_json( + judgments_path, + { + "judgments": [ + { + "schema_version": "conclave_eval_v1", + "judgment_id": f"j-{index}", + "planned_run_id": item.planned_run_id, + "grader_id": "human-a", + "critical_error_free": True, + } + for index, item in enumerate(manifest.planned_runs) + ] + }, + ) + json_output = tmp_path / "report.json" + markdown_output = tmp_path / "report.md" + + result = runner.invoke( + app, + [ + "eval", + "report", + str(manifest_path), + str(run_path), + str(judgments_path), + str(json_output), + str(markdown_output), + "--bootstrap-seed", + "29", + "--bootstrap-samples", + "50", + ], + ) + + assert result.exit_code == 0, result.output + assert json.loads(json_output.read_text())["decision_eligibility"] == "not_yet_eligible" + assert "SYNTHETIC / EXPLORATORY" in markdown_output.read_text() From 4365c288b48c38bc995e17bc39804406c34a3031 Mon Sep 17 00:00:00 2001 From: ernestprovo23 Date: Fri, 17 Jul 2026 17:36:00 -0400 Subject: [PATCH 7/7] fix(evals): enforce artifact integrity boundaries --- src/conclave/evals/replay.py | 119 ++++++++++++++++++++++++++++------ src/conclave/evals/runner.py | 17 ++++- src/conclave/evals/scoring.py | 10 ++- tests/evals/test_eval_cli.py | 36 ++++++++++ tests/evals/test_replay.py | 97 +++++++++++++++++++++++++++ tests/evals/test_runner.py | 2 + tests/evals/test_scoring.py | 42 ++++++++++++ 7 files changed, 299 insertions(+), 24 deletions(-) diff --git a/src/conclave/evals/replay.py b/src/conclave/evals/replay.py index c7e0764..78f66e6 100644 --- a/src/conclave/evals/replay.py +++ b/src/conclave/evals/replay.py @@ -9,24 +9,23 @@ from typing import Any, Literal from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, model_validator from ..adapters.base import redact from .models import Sha256Digest REPLAY_SCHEMA_VERSION = "conclave_replay_v1" -_SENSITIVE_NAMES = { +_CREDENTIAL_NAMES = { "api_key", "apikey", "authorization", - "key", "password", "secret", "signature", - "token", "x-api-key", "x-goog-api-key", } +_AMBIGUOUS_CREDENTIAL_NAMES = {"key", "token"} PostJson = Callable[[str, dict[str, str], dict, float], Awaitable[tuple[int, object]]] @@ -62,45 +61,119 @@ class ReplayArtifact(BaseModel): base_manifest_hash: Sha256Digest records: tuple[ReplayRecord, ...] + @model_validator(mode="after") + def validate_integrity(self) -> ReplayArtifact: + occurrences: dict[str, list[int]] = {} + for record in self.records: + sanitized_request = _sanitize_stored_request(record.request) + if sanitized_request != record.request: + raise ValueError("replay request must be sanitized") + if _sanitize(record.response, ()) != record.response: + raise ValueError("replay response must be sanitized") + if _hash_request(record.request) != record.request_hash: + raise ValueError("replay request hash does not match stored request") + occurrences.setdefault(record.request_hash, []).append(record.occurrence_index) + if any(sorted(indexes) != list(range(len(indexes))) for indexes in occurrences.values()): + raise ValueError("replay occurrence indexes must be contiguous from zero") + return self + def _is_sensitive_name(name: str) -> bool: lowered = name.lower().replace("-", "_") - return lowered in {item.replace("-", "_") for item in _SENSITIVE_NAMES} or any( + return lowered in {item.replace("-", "_") for item in _CREDENTIAL_NAMES} or any( marker in lowered for marker in ("secret", "password", "authorization") ) -def _sanitize(value: Any) -> Any: +def _redact_exact(text: str, credentials: tuple[str, ...]) -> str: + cleaned = text + for credential in sorted(credentials, key=len, reverse=True): + if credential: + cleaned = cleaned.replace(credential, "[REDACTED]") + return redact(cleaned) + + +def _sanitize(value: Any, credentials: tuple[str, ...], *, body_root: bool = False) -> Any: if isinstance(value, Mapping): return { - str(key): _sanitize(item) + str(key): _sanitize(item, credentials) for key, item in sorted(value.items(), key=lambda pair: str(pair[0])) if not _is_sensitive_name(str(key)) + and not ( + body_root and str(key).lower().replace("-", "_") in _AMBIGUOUS_CREDENTIAL_NAMES + ) } if isinstance(value, (list, tuple)): - return [_sanitize(item) for item in value] + return [_sanitize(item, credentials) for item in value] if isinstance(value, str): - return redact(value) + return _redact_exact(value, credentials) if value is None or isinstance(value, (bool, int, float)): return value - return redact(str(value)) + return _redact_exact(str(value), credentials) -def _sanitize_url(url: str) -> str: +def _sanitize_url(url: str, credentials: tuple[str, ...] = ()) -> str: parts = urlsplit(url) safe_query = [ - (name, redact(value)) + (name, _redact_exact(value, credentials)) for name, value in parse_qsl(parts.query, keep_blank_values=True) if not _is_sensitive_name(name) + and name.lower().replace("-", "_") not in _AMBIGUOUS_CREDENTIAL_NAMES ] return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(safe_query), "")) -def _request(url: str, body: dict) -> tuple[dict[str, Any], str]: - safe = {"url": _sanitize_url(url), "body": _sanitize(body)} - canonical = json.dumps(safe, ensure_ascii=False, separators=(",", ":"), sort_keys=True) - digest = f"sha256:{hashlib.sha256(canonical.encode('utf-8')).hexdigest()}" - return safe, digest +def _string_values(value: Any) -> list[str]: + if isinstance(value, str): + return [value] + if isinstance(value, Mapping): + return [item for nested in value.values() for item in _string_values(nested)] + if isinstance(value, (list, tuple)): + return [item for nested in value for item in _string_values(nested)] + return [] + + +def _credentials(url: str, headers: Mapping[str, str], body: Mapping[str, Any]) -> tuple[str, ...]: + values: list[str] = [] + for name, value in parse_qsl(urlsplit(url).query, keep_blank_values=True): + normalized = name.lower().replace("-", "_") + if _is_sensitive_name(name) or normalized in _AMBIGUOUS_CREDENTIAL_NAMES: + values.extend(_string_values(value)) + for name, value in headers.items(): + if _is_sensitive_name(name): + values.append(value) + if value.lower().startswith("bearer "): + values.append(value[7:]) + for name, value in body.items(): + normalized = str(name).lower().replace("-", "_") + if _is_sensitive_name(str(name)) or normalized in _AMBIGUOUS_CREDENTIAL_NAMES: + values.extend(_string_values(value)) + return tuple(dict.fromkeys(item for item in values if item)) + + +def _hash_request(request: Mapping[str, Any]) -> str: + canonical = json.dumps(request, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + return f"sha256:{hashlib.sha256(canonical.encode('utf-8')).hexdigest()}" + + +def _sanitize_stored_request(request: Mapping[str, Any]) -> dict[str, Any]: + if set(request) != {"url", "body"} or not isinstance(request.get("url"), str): + raise ValueError("sanitized replay request must contain only string url and object body") + body = request.get("body") + if not isinstance(body, Mapping): + raise ValueError("sanitized replay request body must be an object") + return {"url": _sanitize_url(request["url"]), "body": _sanitize(body, (), body_root=True)} + + +def _request( + url: str, headers: Mapping[str, str], body: dict +) -> tuple[dict[str, Any], str, tuple[str, ...]]: + credentials = _credentials(url, headers, body) + safe = { + "url": _sanitize_url(url, credentials), + "body": _sanitize(body, credentials, body_root=True), + } + return safe, _hash_request(safe), credentials class RecordingPostJson: @@ -115,7 +188,7 @@ def __init__(self, delegate: PostJson, *, base_manifest_hash: str) -> None: async def __call__( self, url: str, headers: dict[str, str], json_body: dict, timeout: float ) -> tuple[int, object]: - safe_request, request_hash = _request(url, json_body) + safe_request, request_hash, credentials = _request(url, headers, json_body) occurrence = self._counts[request_hash] self._counts[request_hash] += 1 status, response = await self._delegate(url, headers, json_body, timeout) @@ -125,7 +198,7 @@ async def __call__( occurrence_index=occurrence, request=safe_request, status=status, - response=_sanitize(response), + response=_sanitize(response, credentials), ) ) return status, response @@ -147,6 +220,10 @@ def __init__(self, artifact: ReplayArtifact, *, base_manifest_hash: str) -> None ) if artifact.base_manifest_hash != base_manifest_hash: raise ReplayCompatibilityError("replay base manifest hash mismatch") + try: + artifact = ReplayArtifact.model_validate(artifact.model_dump(mode="python")) + except ValueError as exc: + raise ReplayCompatibilityError("replay artifact integrity validation failed") from exc self._records = { (record.request_hash, record.occurrence_index): record for record in artifact.records } @@ -158,8 +235,8 @@ def __init__(self, artifact: ReplayArtifact, *, base_manifest_hash: str) -> None async def __call__( self, url: str, headers: dict[str, str], json_body: dict, timeout: float ) -> tuple[int, object]: - del headers, timeout - _safe_request, request_hash = _request(url, json_body) + del timeout + _safe_request, request_hash, _credentials_found = _request(url, headers, json_body) occurrence = self._counts[request_hash] self._counts[request_hash] += 1 key = (request_hash, occurrence) diff --git a/src/conclave/evals/runner.py b/src/conclave/evals/runner.py index ffadd44..7ca0ff9 100644 --- a/src/conclave/evals/runner.py +++ b/src/conclave/evals/runner.py @@ -32,12 +32,27 @@ class RunValidationError(ValueError): def validate_run_records( manifest: StudyManifest, records: Sequence[RunRecord] ) -> tuple[RunRecord, ...]: - """Require exactly one result for every planned cell and no other results.""" + """Require complete coverage and enforce each cell's execution invariants.""" planned_ids = [planned.planned_run_id for planned in manifest.planned_runs] actual_ids = [record.planned_run_id for record in records] if Counter(actual_ids) != Counter(planned_ids): raise RunValidationError("run records must cover every planned_run_id exactly once") + planned_by_id = {planned.planned_run_id: planned for planned in manifest.planned_runs} + for record in records: + planned = planned_by_id[record.planned_run_id] + if ( + record.completion_tokens is not None + and record.completion_tokens > planned.max_output_tokens + and not ( + record.outcome == "malformed" and record.error_category == "output_budget_exceeded" + ) + ): + raise RunValidationError( + f"run {record.planned_run_id} exceeds its planned output budget" + ) + if record.outcome == "success" and record.output is None: + raise RunValidationError(f"run {record.planned_run_id} is missing its success output") return tuple(records) diff --git a/src/conclave/evals/scoring.py b/src/conclave/evals/scoring.py index 297ba8f..f43037d 100644 --- a/src/conclave/evals/scoring.py +++ b/src/conclave/evals/scoring.py @@ -17,6 +17,7 @@ StudyManifest, StudyRun, ) +from .runner import validate_run_records class GraderJudgment(EvalModel): @@ -288,6 +289,7 @@ def score_study( if study_run.study_id != manifest.study_id: raise ValueError("study run and manifest study_id values must match") + validate_run_records(manifest, study_run.records) planned_by_id = {run.planned_run_id: run for run in manifest.planned_runs} run_by_id = {record.planned_run_id: record for record in study_run.records} if set(run_by_id) != set(planned_by_id) or len(run_by_id) != len(study_run.records): @@ -321,8 +323,12 @@ def score_study( if not set(adjudication.source_judgment_ids).issubset(judgment_ids): raise ValueError("adjudication references an unknown source judgment") expected_source_ids = {item.judgment_id for item in judgments_by_run[run_id]} - if not set(adjudication.source_judgment_ids).issubset(expected_source_ids): - raise ValueError("adjudication source judgments must target the same run") + source_ids = set(adjudication.source_judgment_ids) + if source_ids != expected_source_ids: + raise ValueError("adjudication must cite exactly all same-run source judgments") + values = {item.critical_error_free for item in judgments_by_run[run_id]} + if len(expected_source_ids) < 2 or len(values) < 2: + raise ValueError("adjudication requires a genuine multi-grader disagreement") adjudication_by_run[run_id] = adjudication resolved: list[ResolvedOutcome] = [] diff --git a/tests/evals/test_eval_cli.py b/tests/evals/test_eval_cli.py index 2c894ce..9440e3a 100644 --- a/tests/evals/test_eval_cli.py +++ b/tests/evals/test_eval_cli.py @@ -122,6 +122,42 @@ def test_eval_run_rejects_live_and_validates_offline_replay_artifact(tmp_path) - assert "summary" in rejected_tamper.output.lower() +def test_eval_run_rejects_records_that_bypass_manifest_execution_invariants(tmp_path) -> None: + _, manifest_path, run_path, manifest = _study_artifacts(tmp_path) + output = tmp_path / "validated-run.json" + + for name, replacement, expected in ( + ( + "over-budget", + {"completion_tokens": manifest.planned_runs[0].max_output_tokens + 1}, + "output budget", + ), + ("missing-output", {"output": None}, "success output"), + ): + payload = json.loads(run_path.read_text()) + payload["records"][0].update(replacement) + payload["total_completion_tokens"] = sum( + item.get("completion_tokens") or 0 for item in payload["records"] + ) + tampered_path = tmp_path / f"{name}.json" + _write_json(tampered_path, payload) + + result = runner.invoke( + app, + [ + "eval", + "run", + str(manifest_path), + str(output), + "--replay-artifact", + str(tampered_path), + ], + ) + + assert result.exit_code == 2 + assert expected in result.output.lower() + + def test_eval_blind_writes_separate_grader_and_identity_artifacts(tmp_path) -> None: _, _, run_path, _ = _study_artifacts(tmp_path) grader_path = tmp_path / "grader.json" diff --git a/tests/evals/test_replay.py b/tests/evals/test_replay.py index cb150a5..5409c7a 100644 --- a/tests/evals/test_replay.py +++ b/tests/evals/test_replay.py @@ -13,6 +13,7 @@ ReplayCompatibilityError, ReplayingPostJson, ReplayMismatchError, + ReplayRecord, ) BASE_HASH = "sha256:" + "a" * 64 @@ -68,6 +69,55 @@ async def network(url, headers, body, timeout): assert "alt=json" in encoded +async def test_recording_preserves_legitimate_nested_key_token_and_clean_response_strings(): + async def network(url, headers, body, timeout): + return 200, {"message": "token budgeting uses a key/value table"} + + recorder = RecordingPostJson(network, base_manifest_hash=BASE_HASH) + await recorder( + "https://example.test/v1", + {}, + { + "metadata": {"key": "topic", "token": "budget-unit"}, + "prompt": "Explain token budgeting.", + }, + 10, + ) + + record = recorder.artifact().records[0] + assert record.request["body"]["metadata"] == {"key": "topic", "token": "budget-unit"} + assert record.response == {"message": "token budgeting uses a key/value table"} + + +async def test_recording_redacts_exact_credentials_from_request_and_response(monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "opaque-env-credential") + + async def network(url, headers, body, timeout): + return 200, { + "header_echo": "Bearer opaque-header-credential", + "body_echo": "opaque-body-credential", + "url_echo": "opaque-url-credential", + "env_echo": "opaque-env-credential", + } + + recorder = RecordingPostJson(network, base_manifest_hash=BASE_HASH) + await recorder( + "https://example.test/v1?api_key=opaque-url-credential&alt=json", + {"Authorization": "Bearer opaque-header-credential"}, + {"api_key": "opaque-body-credential", "prompt": "safe"}, + 10, + ) + + encoded = json.dumps(recorder.artifact().model_dump(mode="json"), sort_keys=True) + for credential in ( + "opaque-header-credential", + "opaque-body-credential", + "opaque-url-credential", + "opaque-env-credential", + ): + assert credential not in encoded + + def test_replay_rejects_schema_or_base_manifest_hash_drift(): artifact = ReplayArtifact(base_manifest_hash=BASE_HASH, records=()) with pytest.raises(ReplayCompatibilityError, match="base manifest hash"): @@ -99,3 +149,50 @@ async def network(url, headers, body, timeout): await replay("https://example.test/v1", {}, {"model": "m", "prompt": "one"}, 10) with pytest.raises(ReplayMismatchError, match="unmatched request"): await replay("https://example.test/v1", {}, {"model": "m", "prompt": "one"}, 10) + + +def test_replay_artifact_rejects_stored_request_hash_mismatch(): + request = {"url": "https://example.test/v1", "body": {"prompt": "safe"}} + record = ReplayRecord( + request_hash="sha256:" + "b" * 64, + occurrence_index=0, + request=request, + status=200, + response={"ok": True}, + ) + + with pytest.raises(ValueError, match="request hash"): + ReplayArtifact(base_manifest_hash=BASE_HASH, records=(record,)) + + +async def test_replay_artifact_rejects_noncontiguous_occurrence_indexes(): + async def network(url, headers, body, timeout): + return 200, {"ok": True} + + recorder = RecordingPostJson(network, base_manifest_hash=BASE_HASH) + await recorder("https://example.test/v1", {}, {"prompt": "safe"}, 10) + record = recorder.artifact().records[0].model_copy(update={"occurrence_index": 1}) + + with pytest.raises(ValueError, match="contiguous"): + ReplayArtifact(base_manifest_hash=BASE_HASH, records=(record,)) + + +async def test_replay_artifact_rejects_secret_bearing_or_tampered_payloads(): + async def network(url, headers, body, timeout): + return 200, {"ok": True} + + recorder = RecordingPostJson(network, base_manifest_hash=BASE_HASH) + await recorder("https://example.test/v1", {}, {"prompt": "safe"}, 10) + valid = recorder.artifact().records[0] + + secret_request = valid.model_copy( + update={"request": {**valid.request, "authorization": "Bearer opaque-secret"}} + ) + secret_response = valid.model_copy(update={"response": "Bearer sk-secret-value"}) + for record in (secret_request, secret_response): + with pytest.raises(ValueError, match="sanitized"): + ReplayArtifact(base_manifest_hash=BASE_HASH, records=(record,)) + + bypassed_validation = recorder.artifact().model_copy(update={"records": (secret_response,)}) + with pytest.raises(ReplayCompatibilityError, match="integrity"): + ReplayingPostJson(bypassed_validation, base_manifest_hash=BASE_HASH) diff --git a/tests/evals/test_runner.py b/tests/evals/test_runner.py index cf58f93..b57f59e 100644 --- a/tests/evals/test_runner.py +++ b/tests/evals/test_runner.py @@ -139,10 +139,12 @@ async def executor(*, task, planned_run, max_output_tokens): assert by_condition["single_frontier"].outcome == "malformed" assert by_condition["single_frontier"].error_category == "output_budget_exceeded" + assert by_condition["single_frontier"].completion_tokens == 101 assert by_condition["self_refine"].outcome == "malformed" assert by_condition["self_refine"].error_category == "missing_success_output" assert study_run.total_planned_runs == 6 assert sum(study_run.outcome_counts.values()) == 6 + assert study_run.total_completion_tokens == 121 @pytest.mark.asyncio diff --git a/tests/evals/test_scoring.py b/tests/evals/test_scoring.py index 21a7fd7..7cb3ed2 100644 --- a/tests/evals/test_scoring.py +++ b/tests/evals/test_scoring.py @@ -126,6 +126,48 @@ def test_scoring_preserves_raw_records_and_keeps_adjudication_separate() -> None assert report.reliability.adjudication_rate == 1.0 +@pytest.mark.parametrize( + ("raw_values", "source_indexes", "expected"), + ( + ((True,), (0,), "disagreement"), + ((True, True), (0, 1), "disagreement"), + ((True, False), (0,), "exactly all"), + ), +) +def test_adjudication_only_resolves_complete_multi_grader_disagreement( + raw_values, source_indexes, expected +) -> None: + manifest, run = _study(task_count=1) + planned_id = manifest.planned_runs[0].planned_run_id + raw = tuple( + GraderJudgment( + judgment_id=f"judgment-{index}", + planned_run_id=planned_id, + grader_id=f"grader-{index}", + critical_error_free=value, + ) + for index, value in enumerate(raw_values) + ) + adjudication = AdjudicationRecord( + adjudication_id="adjudication-invalid", + planned_run_id=planned_id, + critical_error_free=True, + source_judgment_ids=tuple(raw[index].judgment_id for index in source_indexes), + adjudicator_id="adjudicator", + rationale="Attempted override.", + ) + + with pytest.raises(ValueError, match=expected): + score_study( + manifest=manifest, + study_run=run, + raw_judgments=raw, + adjudications=(adjudication,), + bootstrap_seed=9, + bootstrap_samples=10, + ) + + def test_critical_error_free_rate_keeps_all_planned_runs_in_denominator() -> None: manifest, successful = _study(task_count=1) first_condition = manifest.planned_runs[0].condition_id