Add deterministic degradation curve artifacts

ProfRandom92 · web-flow · commit e465d7c9c1b5 · 2026-05-19T03:51:15.000-07:00
Add deterministic degradation curve artifacts

- Add DegradationCurveGenerator with stable JSON and Markdown artifact output.
- Add expected/disallowed failure validation for fixture expectations.
- Read fixture_version from fixture metadata and require explicit curve_id generation.
- Add committed layered admissibility artifact and benchmark documentation.

Validation reported in PR: degradation curve generator tests, full pytest suite, and npm run check passed.
diff --git a/artifacts/layered_admissibility_results.json b/artifacts/layered_admissibility_results.json
@@ -0,0 +1,52 @@
+{
+  "curve_id": "coding_workflow_pr_review_curve_v1",
+  "generated_by": "DegradationCurveGenerator",
+  "points": [
+    {
+      "expected_admissible": true,
+      "failed_contracts": [],
+      "failure_labels": [],
+      "fixture_id": "coding_workflow_pr_review_v1",
+      "fixture_path": "fixtures/coding_workflow_pr_review_v1",
+      "fixture_version": "1.0.0",
+      "governance_score": 1.0,
+      "observed_admissible": true,
+      "operational_score": 1.0,
+      "overall_admissibility_score": 1.0,
+      "passed_contracts": [
+        "no_orphan_tool_calls",
+        "pre_merge_review",
+        "recovery_path_available",
+        "security_causal_block"
+      ],
+      "relational_score": 1.0,
+      "structural_score": 1.0
+    },
+    {
+      "expected_admissible": false,
+      "failed_contracts": [
+        "no_orphan_tool_calls",
+        "pre_merge_review",
+        "recovery_path_available",
+        "security_causal_block"
+      ],
+      "failure_labels": [
+        "CAUSAL_DEPENDENCY_LOSS",
+        "INVARIANT_VIOLATION",
+        "POLICY_ORDER_BROKEN",
+        "RECOVERY_PATH_INVALID"
+      ],
+      "fixture_id": "coding_workflow_pr_review_degraded_v1",
+      "fixture_path": "fixtures/coding_workflow_pr_review_degraded_v1",
+      "fixture_version": "1.0.0",
+      "governance_score": 1.0,
+      "observed_admissible": false,
+      "operational_score": 0.0,
+      "overall_admissibility_score": 0.5,
+      "passed_contracts": [],
+      "relational_score": 0.0,
+      "structural_score": 1.0
+    }
+  ],
+  "version": "1.0"
+}
diff --git a/docs/LAYERED_ADMISSIBILITY_SCORE_v1.md b/docs/LAYERED_ADMISSIBILITY_SCORE_v1.md
@@ -68,3 +68,9 @@ For each layer (`structural`, `relational`, `operational`, `governance`):
 
 - v1 uses unweighted averages only.
 - Future versions may add explicit configured weights, but not learned weights.
+
+
+## Generated artifacts
+
+- `artifacts/layered_admissibility_results.json`
+- `docs/benchmarks/layered_admissibility.md`
diff --git a/docs/benchmarks/layered_admissibility.md b/docs/benchmarks/layered_admissibility.md
@@ -0,0 +1,29 @@
+# Layered Admissibility Degradation Benchmark
+
+## Purpose
+
+Deterministically compare admissibility outcomes across fixture bundles using ContractValidator and AdmissibilityScorer.
+
+## Fixture results
+
+| fixture_id | expected_admissible | observed_admissible | structural_score | relational_score | operational_score | governance_score | overall_admissibility_score | failure_labels |
+| --- | --- | --- | --- | --- | --- | --- | --- | --- |
+| coding_workflow_pr_review_v1 | true | true | 1.000 | 1.000 | 1.000 | 1.000 | 1.000 | none |
+| coding_workflow_pr_review_degraded_v1 | false | false | 1.000 | 0.000 | 0.000 | 1.000 | 0.500 | CAUSAL_DEPENDENCY_LOSS, INVARIANT_VIOLATION, POLICY_ORDER_BROKEN, RECOVERY_PATH_INVALID |
+
+## Interpretation
+
+The positive fixture remains fully admissible while the degraded fixture shows deterministic score loss and explicit failure labels.
+
+## Non-goals
+
+- no LLM judges
+- no embeddings
+- no fuzzy matching
+- no semantic equivalence
+
+## Future
+
+- add more fixture families
+- add progressive degradation levels
+- add SVG curve visualization later
diff --git a/src/validation/degradation_curve_generator.py b/src/validation/degradation_curve_generator.py
@@ -0,0 +1,177 @@
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any
+
+from src.validation.admissibility_scorer import AdmissibilityScorer
+from src.validation.contract_validator import ContractValidator
+
+
+@dataclass(frozen=True, slots=True)
+class FixtureScorePoint:
+    fixture_id: str
+    fixture_version: str
+    fixture_path: str
+    expected_admissible: bool
+    observed_admissible: bool
+    structural_score: float
+    relational_score: float
+    operational_score: float
+    governance_score: float
+    overall_admissibility_score: float
+    passed_contracts: tuple[str, ...]
+    failed_contracts: tuple[str, ...]
+    failure_labels: tuple[str, ...]
+
+
+@dataclass(frozen=True, slots=True)
+class DegradationCurve:
+    curve_id: str
+    version: str
+    generated_by: str
+    points: tuple[FixtureScorePoint, ...]
+
+
+class DegradationCurveGenerator:
+    VERSION = "1.0"
+
+    def _load_json(self, path: Path) -> dict[str, Any]:
+        if not path.exists():
+            raise FileNotFoundError(f"missing required fixture file: {path}")
+        return json.loads(path.read_text(encoding="utf-8"))
+
+    def _fixture_version(self, fixture_path: Path, expected_admissibility: dict[str, Any]) -> str:
+        if "fixture_version" not in expected_admissibility:
+            raise ValueError(f"missing fixture_version in {fixture_path / 'expected/admissibility.json'}")
+        return str(expected_admissibility["fixture_version"])
+
+    def _validate_expected_failures(
+        self,
+        fixture_path: Path,
+        expected_failures_payload: dict[str, Any],
+        observed_failure_labels: tuple[str, ...],
+    ) -> None:
+        expected = set(expected_failures_payload.get("expected_failures", []))
+        disallowed = set(expected_failures_payload.get("disallowed_failures", []))
+        observed = set(observed_failure_labels)
+
+        missing_expected = sorted(expected - observed)
+        if missing_expected:
+            raise ValueError(f"missing expected failure labels for {fixture_path}: {missing_expected}")
+
+        emitted_disallowed = sorted(disallowed & observed)
+        if emitted_disallowed:
+            raise ValueError(f"emitted disallowed failure labels for {fixture_path}: {emitted_disallowed}")
+
+    def evaluate_fixture(self, fixture_path: Path) -> FixtureScorePoint:
+        original = {
+            **self._load_json(fixture_path / "original/trace.json"),
+            **self._load_json(fixture_path / "original/state.json"),
+            "dependency_graph": self._load_json(fixture_path / "original/dependency_graph.json"),
+        }
+        reconstructed = {
+            **self._load_json(fixture_path / "reconstructed/trace.json"),
+            **self._load_json(fixture_path / "reconstructed/state.json"),
+            "dependency_graph": self._load_json(fixture_path / "reconstructed/dependency_graph.json"),
+        }
+        contracts_dir = fixture_path / "original/contracts"
+        contracts = [self._load_json(contract_path) for contract_path in sorted(contracts_dir.glob("*.json"))]
+        if not contracts:
+            raise FileNotFoundError(f"no contract files found in fixture: {contracts_dir}")
+
+        expected_admissibility = self._load_json(fixture_path / "expected/admissibility.json")
+        expected_admissible = bool(expected_admissibility["expected_admissible"])
+        fixture_version = self._fixture_version(fixture_path, expected_admissibility)
+        expected_failures = self._load_json(fixture_path / "expected/failures.json")
+
+        results = ContractValidator().validate_contracts(original=original, reconstructed=reconstructed, contracts=contracts)
+        score = AdmissibilityScorer().score(results, expected_admissible=expected_admissible)
+        self._validate_expected_failures(fixture_path, expected_failures, score.failure_labels)
+
+        return FixtureScorePoint(
+            fixture_id=fixture_path.name,
+            fixture_version=fixture_version,
+            fixture_path=fixture_path.as_posix(),
+            expected_admissible=score.expected_admissible,
+            observed_admissible=score.observed_admissible,
+            structural_score=score.structural_score,
+            relational_score=score.relational_score,
+            operational_score=score.operational_score,
+            governance_score=score.governance_score,
+            overall_admissibility_score=score.overall_admissibility_score,
+            passed_contracts=tuple(sorted(score.passed_contracts)),
+            failed_contracts=tuple(sorted(score.failed_contracts)),
+            failure_labels=tuple(sorted(score.failure_labels)),
+        )
+
+    def generate(self, fixtures: list[Path], curve_id: str) -> DegradationCurve:
+        points = tuple(self.evaluate_fixture(path) for path in fixtures)
+        return DegradationCurve(curve_id=curve_id, version=self.VERSION, generated_by=self.__class__.__name__, points=points)
+
+    def to_dict(self, curve: DegradationCurve) -> dict[str, object]:
+        return {
+            "curve_id": curve.curve_id,
+            "version": curve.version,
+            "generated_by": curve.generated_by,
+            "points": [
+                {
+                    **asdict(point),
+                    "passed_contracts": list(point.passed_contracts),
+                    "failed_contracts": list(point.failed_contracts),
+                    "failure_labels": list(point.failure_labels),
+                }
+                for point in curve.points
+            ],
+        }
+
+    def write_json(self, curve: DegradationCurve, output_path: Path) -> None:
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        output_path.write_text(json.dumps(self.to_dict(curve), indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+    def write_markdown(self, curve: DegradationCurve, output_path: Path) -> None:
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        rows = []
+        for point in curve.points:
+            labels = ", ".join(point.failure_labels) if point.failure_labels else "none"
+            rows.append(
+                f"| {point.fixture_id} | {str(point.expected_admissible).lower()} | {str(point.observed_admissible).lower()} | "
+                f"{point.structural_score:.3f} | {point.relational_score:.3f} | {point.operational_score:.3f} | "
+                f"{point.governance_score:.3f} | {point.overall_admissibility_score:.3f} | {labels} |"
+            )
+
+        markdown = "\n".join(
+            [
+                "# Layered Admissibility Degradation Benchmark",
+                "",
+                "## Purpose",
+                "",
+                "Deterministically compare admissibility outcomes across fixture bundles using ContractValidator and AdmissibilityScorer.",
+                "",
+                "## Fixture results",
+                "",
+                "| fixture_id | expected_admissible | observed_admissible | structural_score | relational_score | operational_score | governance_score | overall_admissibility_score | failure_labels |",
+                "| --- | --- | --- | --- | --- | --- | --- | --- | --- |",
+                *rows,
+                "",
+                "## Interpretation",
+                "",
+                "The positive fixture remains fully admissible while the degraded fixture shows deterministic score loss and explicit failure labels.",
+                "",
+                "## Non-goals",
+                "",
+                "- no LLM judges",
+                "- no embeddings",
+                "- no fuzzy matching",
+                "- no semantic equivalence",
+                "",
+                "## Future",
+                "",
+                "- add more fixture families",
+                "- add progressive degradation levels",
+                "- add SVG curve visualization later",
+                "",
+            ]
+        )
+        output_path.write_text(markdown, encoding="utf-8")
diff --git a/tests/test_degradation_curve_generator.py b/tests/test_degradation_curve_generator.py
@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from src.validation.degradation_curve_generator import DegradationCurveGenerator
+
+
+POS_FIXTURE = Path("fixtures/coding_workflow_pr_review_v1")
+NEG_FIXTURE = Path("fixtures/coding_workflow_pr_review_degraded_v1")
+ARTIFACT_PATH = Path("artifacts/layered_admissibility_results.json")
+CURVE_ID = "coding_workflow_pr_review_curve_v1"
+
+
+def test_evaluate_positive_fixture_scores_one() -> None:
+    point = DegradationCurveGenerator().evaluate_fixture(POS_FIXTURE)
+    assert point.fixture_version == "1.0.0"
+    assert point.observed_admissible is True
+    assert point.overall_admissibility_score == 1.0
+    assert point.failed_contracts == ()
+    assert point.failure_labels == ()
+
+
+def test_evaluate_negative_fixture_detects_expected_failures() -> None:
+    point = DegradationCurveGenerator().evaluate_fixture(NEG_FIXTURE)
+    assert point.fixture_version == "1.0.0"
+    assert point.observed_admissible is False
+    assert point.overall_admissibility_score < 1.0
+    assert {
+        "POLICY_ORDER_BROKEN",
+        "RECOVERY_PATH_INVALID",
+        "CAUSAL_DEPENDENCY_LOSS",
+        "INVARIANT_VIOLATION",
+    }.issubset(set(point.failure_labels))
+
+
+def test_generate_curve_is_deterministic() -> None:
+    generator = DegradationCurveGenerator()
+    fixtures = [POS_FIXTURE, NEG_FIXTURE]
+    assert generator.to_dict(generator.generate(fixtures, curve_id=CURVE_ID)) == generator.to_dict(
+        generator.generate(fixtures, curve_id=CURVE_ID)
+    )
+
+
+def test_to_dict_is_json_compatible_and_sorted() -> None:
+    generator = DegradationCurveGenerator()
+    curve = generator.generate([POS_FIXTURE, NEG_FIXTURE], curve_id=CURVE_ID)
+    curve_dict = generator.to_dict(curve)
+    json.dumps(curve_dict, sort_keys=True)
+    assert [point["fixture_path"] for point in curve_dict["points"]] == [
+        POS_FIXTURE.as_posix(),
+        NEG_FIXTURE.as_posix(),
+    ]
+
+
+def test_write_json_matches_committed_artifact(tmp_path: Path) -> None:
+    generator = DegradationCurveGenerator()
+    curve = generator.generate([POS_FIXTURE, NEG_FIXTURE], curve_id=CURVE_ID)
+    generated_path = tmp_path / "layered_admissibility_results.json"
+    generator.write_json(curve, generated_path)
+
+    generated = json.loads(generated_path.read_text(encoding="utf-8"))
+    committed = json.loads(ARTIFACT_PATH.read_text(encoding="utf-8"))
+    assert generated == committed
+
+
+def test_write_markdown_contains_fixture_rows(tmp_path: Path) -> None:
+    generator = DegradationCurveGenerator()
+    curve = generator.generate([POS_FIXTURE, NEG_FIXTURE], curve_id=CURVE_ID)
+    markdown_path = tmp_path / "layered_admissibility.md"
+    generator.write_markdown(curve, markdown_path)
+
+    content = markdown_path.read_text(encoding="utf-8")
+    assert "coding_workflow_pr_review_v1" in content
+    assert "coding_workflow_pr_review_degraded_v1" in content
+    assert "POLICY_ORDER_BROKEN" in content
+    assert "RECOVERY_PATH_INVALID" in content
+
+
+def test_missing_fixture_file_raises_clear_error(tmp_path: Path) -> None:
+    incomplete = tmp_path / "fixture"
+    incomplete.mkdir(parents=True)
+    with pytest.raises(FileNotFoundError, match="missing required fixture file"):
+        DegradationCurveGenerator().evaluate_fixture(incomplete)
+
+
+def test_missing_expected_failure_label_raises_clear_error() -> None:
+    generator = DegradationCurveGenerator()
+
+    with pytest.raises(ValueError, match="missing expected failure labels"):
+        generator._validate_expected_failures(
+            Path("fixtures/example"),
+            {"expected_failures": ["MISSING_EXPECTED_FAILURE"], "disallowed_failures": []},
+            ("OBSERVED_FAILURE",),
+        )
+
+
+def test_disallowed_failure_label_raises_clear_error() -> None:
+    generator = DegradationCurveGenerator()
+
+    with pytest.raises(ValueError, match="emitted disallowed failure labels"):
+        generator._validate_expected_failures(
+            Path("fixtures/example"),
+            {"expected_failures": [], "disallowed_failures": ["DISALLOWED_FAILURE"]},
+            ("DISALLOWED_FAILURE",),
+        )
