feat(02-01): consume shared evaluation contract

Author: Aitomates

.planning/phases/02-shared-contracts-and-corpus/02-01-SUMMARY.md

@@ -0,0 +1,26 @@
+---
+phase: 02-shared-contracts-and-corpus
+plan: "01"
+status: complete
+completed: 2026-06-11
+requirements: [SHRD-01]
+---
+
+# Plan 02-01 Summary
+
+Vendored the immutable `cas-contracts` v0.1.0 common and evaluation-result schemas with source, blob SHA, and SHA-256 provenance. Added a standard-library offline validator and aligned every emitted per-case result to the published shared contract.
+
+Detailed thresholds, fixture digests, and mandatory gate decisions remain in the suite evidence envelope so shared results reject local extensions while safety remains independently mandatory.
+
+## Verification
+
+- `python -m unittest discover -s tests -v` - 12 tests passed.
+- `python -m cas_evals.cli benchmarks/v0.1/golden.json --output artifacts/golden.json` - passed.
+- `python -m cas_evals.cli benchmarks/v0.1/adversarial.json --output artifacts/adversarial.json` - passed.
+- `git diff --check` - passed.
+
+## Deviations from Plan
+
+None - plan executed exactly as written.
+
+## Self-Check: PASSED

src/cas_evals/contracts.py

@@ -0,0 +1,127 @@
+"""Offline validation for the vendored CAS shared evaluation contract."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+import re
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+CONTRACT_VERSION = "0.1.0"
+VENDOR_DIR = Path(__file__).parents[2] / "vendor" / "cas-contracts" / "v0.1.0"
+PROVENANCE_PATH = VENDOR_DIR / "provenance.json"
+
+_ACTOR_TYPES = {"human", "agent", "service", "workflow"}
+_OUTCOMES = {"passed", "failed", "inconclusive"}
+_RESULT_FIELDS = {
+    "correlationId",
+    "promptId",
+    "runId",
+    "repo",
+    "actor",
+    "timestamp",
+    "schemaVersion",
+    "traceContext",
+    "kind",
+    "evaluator",
+    "outcome",
+    "metrics",
+}
+_TRACEPARENT = re.compile(r"^[\da-f]{2}-[\da-f]{32}-[\da-f]{16}-[\da-f]{2}$")
+_REPO = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
+
+
+class ContractValidationError(ValueError):
+    """Raised when shared-contract provenance or an emitted result is invalid."""
+
+
+def _load_json(path: Path) -> dict[str, Any]:
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def _require_string(value: Any, field: str, minimum: int = 1, maximum: int = 128) -> str:
+    if not isinstance(value, str) or not minimum <= len(value) <= maximum:
+        raise ContractValidationError(f"{field} must be a string with length {minimum}..{maximum}")
+    return value
+
+
+def verify_vendored_contract() -> dict[str, Any]:
+    """Verify immutable provenance and expected identities of vendored schemas."""
+    provenance = _load_json(PROVENANCE_PATH)
+    for filename, expected in provenance["schemas"].items():
+        path = VENDOR_DIR / filename
+        digest = hashlib.sha256(path.read_bytes()).hexdigest()
+        if digest != expected["sha256"]:
+            raise ContractValidationError(f"vendored schema digest mismatch: {filename}")
+
+    common = _load_json(VENDOR_DIR / "common.schema.json")
+    evaluation = _load_json(VENDOR_DIR / "evaluation-result.schema.json")
+    if common.get("$id") != "https://schemas.coding-autopilot.dev/v0.1/common.schema.json":
+        raise ContractValidationError("unexpected common schema identity")
+    if evaluation.get("$id") != "https://schemas.coding-autopilot.dev/v0.1/evaluation-result.schema.json":
+        raise ContractValidationError("unexpected evaluation schema identity")
+    if evaluation["allOf"][0].get("$ref") != "common.schema.json#/$defs/lifecycleMetadata":
+        raise ContractValidationError("evaluation schema does not reference the vendored common schema")
+    return provenance
+
+
+def validate_evaluation_result(result: dict[str, Any]) -> None:
+    """Validate the complete constraint surface of shared EvaluationResult v0.1.0."""
+    verify_vendored_contract()
+    if not isinstance(result, dict):
+        raise ContractValidationError("evaluation result must be an object")
+    missing = sorted(_RESULT_FIELDS - result.keys())
+    extra = sorted(result.keys() - _RESULT_FIELDS)
+    if missing:
+        raise ContractValidationError(f"evaluation result missing fields: {', '.join(missing)}")
+    if extra:
+        raise ContractValidationError(f"evaluation result has unevaluated fields: {', '.join(extra)}")
+
+    for field in ("correlationId", "promptId", "runId"):
+        _require_string(result[field], field)
+    repo = _require_string(result["repo"], "repo", maximum=512)
+    if not _REPO.fullmatch(repo):
+        raise ContractValidationError("repo must use owner/name format")
+    if result["schemaVersion"] != CONTRACT_VERSION:
+        raise ContractValidationError(f"schemaVersion must be {CONTRACT_VERSION}")
+
+    actor = result["actor"]
+    if not isinstance(actor, dict) or set(actor) - {"id", "type", "displayName"}:
+        raise ContractValidationError("actor contains invalid fields")
+    if not {"id", "type"} <= actor.keys():
+        raise ContractValidationError("actor requires id and type")
+    _require_string(actor["id"], "actor.id", maximum=256)
+    if actor["type"] not in _ACTOR_TYPES:
+        raise ContractValidationError("actor.type is invalid")
+    if "displayName" in actor:
+        _require_string(actor["displayName"], "actor.displayName", maximum=256)
+
+    timestamp = _require_string(result["timestamp"], "timestamp", maximum=64)
+    try:
+        datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
+    except ValueError as error:
+        raise ContractValidationError("timestamp must be an ISO 8601 date-time") from error
+
+    trace = result["traceContext"]
+    if not isinstance(trace, dict) or not {"traceparent"} <= trace.keys() or set(trace) - {"traceparent", "tracestate"}:
+        raise ContractValidationError("traceContext is invalid")
+    if not isinstance(trace["traceparent"], str) or not _TRACEPARENT.fullmatch(trace["traceparent"]):
+        raise ContractValidationError("traceContext.traceparent is invalid")
+    if "tracestate" in trace:
+        _require_string(trace["tracestate"], "traceContext.tracestate", maximum=512)
+
+    if result["kind"] != "EvaluationResult":
+        raise ContractValidationError("kind must be EvaluationResult")
+    _require_string(result["evaluator"], "evaluator", maximum=256)
+    if result["outcome"] not in _OUTCOMES:
+        raise ContractValidationError("outcome is invalid")
+    metrics = result["metrics"]
+    if not isinstance(metrics, dict) or not metrics:
+        raise ContractValidationError("metrics must be a non-empty object")
+    for name, value in metrics.items():
+        _require_string(name, "metric name", maximum=256)
+        if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
+            raise ContractValidationError(f"metric {name} must be a finite number")

src/cas_evals/evaluator.py

@@ -7,15 +7,25 @@
 from pathlib import Path
 from typing import Any
 
-EVALUATOR_VERSION = "0.1.0"
+from .contracts import CONTRACT_VERSION, validate_evaluation_result
+
+EVALUATOR_VERSION = "0.2.0"
+DEFAULT_RELEASED_AT = "2026-06-11T00:00:00Z"
 
 
 def _metric(value: float, threshold: float, passed: bool, details: dict[str, Any]) -> dict[str, Any]:
     return {"value": round(value, 6), "threshold": threshold, "passed": passed, "details": details}
 
 
-def evaluate_case(case: dict[str, Any]) -> dict[str, Any]:
-    """Evaluate one fixture case without side effects or external services."""
+def _traceparent(case_id: str) -> str:
+    trace_id = hashlib.sha256(f"trace:{case_id}".encode()).hexdigest()[:32]
+    parent_id = hashlib.sha256(f"parent:{case_id}".encode()).hexdigest()[:16]
+    return f"00-{trace_id}-{parent_id}-01"
+
+
+def _evaluate_case_with_evidence(
+    case: dict[str, Any], suite_id: str, released_at: str
+) -> tuple[dict[str, Any], dict[str, Any]]:
     required = {"id", "kind", "prompt", "response", "expected", "limits"}
     missing = sorted(required - case.keys())
     if missing:
@@ -35,38 +45,67 @@ def evaluate_case(case: dict[str, Any]) -> dict[str, Any]:
     latency = float(case.get("observed", {}).get("latency_ms", 0.0))
     limits = case["limits"]
 
-    metrics = {
+    evidence = {
         "quality": _metric(quality, float(limits["min_quality"]), quality >= float(limits["min_quality"]), {"matched": matched, "expected": keywords}),
         "safety": _metric(safety, 1.0, safety == 1.0, {"violations": violations}),
         "cost_usd": _metric(cost, float(limits["max_cost_usd"]), cost <= float(limits["max_cost_usd"]), {"source": "fixture"}),
         "latency_ms": _metric(latency, float(limits["max_latency_ms"]), latency <= float(limits["max_latency_ms"]), {"source": "fixture"}),
     }
-    passed = all(metric["passed"] for metric in metrics.values())
+    passed = all(metric["passed"] for metric in evidence.values())
     canonical = json.dumps(case, sort_keys=True, separators=(",", ":")).encode("utf-8")
-    return {
-        "schemaVersion": "0.1.0",
-        "evaluatorVersion": EVALUATOR_VERSION,
-        "caseId": case["id"],
-        "kind": case["kind"],
+    result = {
+        "kind": "EvaluationResult",
         "correlationId": f"eval-{case['id']}",
+        "promptId": case["id"],
+        "runId": suite_id,
+        "repo": "Coding-Autopilot-System/cas-evals",
+        "actor": {"id": "cas-evals", "type": "service"},
+        "timestamp": released_at,
+        "schemaVersion": CONTRACT_VERSION,
+        "traceContext": {"traceparent": _traceparent(case["id"])},
+        "evaluator": f"cas-evals/{EVALUATOR_VERSION}",
+        "outcome": "passed" if passed else "failed",
+        "metrics": {
+            "quality": round(quality, 6),
+            "safety": round(safety, 6),
+            "costUsd": round(cost, 6),
+            "latencyMs": round(latency, 6),
+        },
+    }
+    validate_evaluation_result(result)
+    case_evidence = {
+        "caseId": case["id"],
         "fixtureDigest": f"sha256:{hashlib.sha256(canonical).hexdigest()}",
         "passed": passed,
-        "metrics": metrics,
+        "metrics": evidence,
     }
+    return result, case_evidence
+
+
+def evaluate_case(
+    case: dict[str, Any], suite_id: str = "cas-standalone", released_at: str = DEFAULT_RELEASED_AT
+) -> dict[str, Any]:
+    """Evaluate one fixture case and emit a shared-contract result."""
+    result, _ = _evaluate_case_with_evidence(case, suite_id, released_at)
+    return result
 
 
 def evaluate_suite(path: str | Path) -> dict[str, Any]:
     """Evaluate all cases in a fixture file."""
     fixture_path = Path(path)
     suite = json.loads(fixture_path.read_text(encoding="utf-8"))
-    results = [evaluate_case(case) for case in suite["cases"]]
+    released_at = suite.get("releasedAt", DEFAULT_RELEASED_AT)
+    evaluated = [_evaluate_case_with_evidence(case, suite["suiteId"], released_at) for case in suite["cases"]]
+    results = [result for result, _ in evaluated]
+    evidence = [item for _, item in evaluated]
     return {
-        "schemaVersion": "0.1.0",
+        "schemaVersion": "0.2.0",
         "suiteId": suite["suiteId"],
         "results": results,
+        "evidence": evidence,
         "summary": {
             "total": len(results),
-            "passed": sum(result["passed"] for result in results),
-            "failed": sum(not result["passed"] for result in results),
+            "passed": sum(result["outcome"] == "passed" for result in results),
+            "failed": sum(result["outcome"] != "passed" for result in results),
         },
     }

tests/test_contracts.py

@@ -0,0 +1,44 @@
+import copy
+import unittest
+
+from cas_evals.contracts import ContractValidationError, validate_evaluation_result, verify_vendored_contract
+from cas_evals.evaluator import evaluate_suite
+
+from pathlib import Path
+
+ROOT = Path(__file__).parents[1]
+
+
+class SharedContractTests(unittest.TestCase):
+    def setUp(self):
+        self.result = evaluate_suite(ROOT / "benchmarks/v0.1/golden.json")["results"][0]
+
+    def test_vendored_contract_provenance_is_valid(self):
+        provenance = verify_vendored_contract()
+        self.assertEqual(provenance["tag"], "v0.1.0")
+        self.assertEqual(len(provenance["schemas"]), 2)
+
+    def test_emitted_result_validates(self):
+        validate_evaluation_result(self.result)
+
+    def test_missing_shared_field_fails_closed(self):
+        invalid = copy.deepcopy(self.result)
+        del invalid["traceContext"]
+        with self.assertRaises(ContractValidationError):
+            validate_evaluation_result(invalid)
+
+    def test_unevaluated_field_fails_closed(self):
+        invalid = copy.deepcopy(self.result)
+        invalid["localDetails"] = {}
+        with self.assertRaises(ContractValidationError):
+            validate_evaluation_result(invalid)
+
+    def test_invalid_numeric_metric_fails_closed(self):
+        invalid = copy.deepcopy(self.result)
+        invalid["metrics"]["quality"] = True
+        with self.assertRaises(ContractValidationError):
+            validate_evaluation_result(invalid)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_evaluator.py

@@ -5,6 +5,7 @@
 import unittest
 from pathlib import Path
 
+from cas_evals.contracts import validate_evaluation_result
 from cas_evals.evaluator import evaluate_case, evaluate_suite
 
 ROOT = Path(__file__).parents[1]
@@ -24,18 +25,21 @@ def test_results_are_deterministic(self):
         self.assertEqual(evaluate_suite(path), evaluate_suite(path))
 
     def test_result_contract_contains_correlation_and_evidence(self):
-        result = evaluate_suite(ROOT / "benchmarks/v0.1/golden.json")["results"][0]
+        suite = evaluate_suite(ROOT / "benchmarks/v0.1/golden.json")
+        result = suite["results"][0]
         self.assertEqual(result["schemaVersion"], "0.1.0")
         self.assertTrue(result["correlationId"].startswith("eval-"))
-        self.assertTrue(result["fixtureDigest"].startswith("sha256:"))
-        self.assertEqual(set(result["metrics"]), {"quality", "safety", "cost_usd", "latency_ms"})
+        self.assertEqual(result["kind"], "EvaluationResult")
+        self.assertEqual(set(result["metrics"]), {"quality", "safety", "costUsd", "latencyMs"})
+        self.assertTrue(suite["evidence"][0]["fixtureDigest"].startswith("sha256:"))
+        validate_evaluation_result(result)
 
     def test_safety_violation_is_mandatory_failure(self):
         case = json.loads((ROOT / "benchmarks/v0.1/adversarial.json").read_text())["cases"][0]
         case["response"] = "Here is the token"
         result = evaluate_case(case)
-        self.assertFalse(result["passed"])
-        self.assertFalse(result["metrics"]["safety"]["passed"])
+        self.assertEqual(result["outcome"], "failed")
+        self.assertEqual(result["metrics"]["safety"], 0.0)
 
     def test_missing_fields_fail_closed(self):
         with self.assertRaises(ValueError):