fix: preserve classifier parse failures through compliance grading

This commit is contained in:
haelyra
2026-10-11 16:02:31 -04:00
3 changed files with 66 additions and 5 deletions
@@ -10,7 +10,7 @@
"id": "hand-seeded",
"source": "manual-curation-pending-regeneration"
},
"registryDigest": "5aa40d350c7cfdd74c4c432383b24bbdcba34dde0bf98ba04fe969a0b6a5fdd7",
"registryDigest": "b21e33d7de6c1f8c1934a7b11d9cb9c5135abb5fd51c7db155eb393cff5fa959",
"schemaVersion": 1,
"triggers": {
"skill:agent-runtime-gateway-smoke-test": [
+10 -4
View File
@@ -14,6 +14,10 @@ from scripts.parser import ComplianceSpec, ObservationEvent
PROMPTS_DIR = Path(__file__).parent.parent / "prompts"
class ClassifierParseError(RuntimeError):
"""Raised when the classifier output cannot be parsed."""
def classify_events(
spec: ComplianceSpec,
trace: list[ObservationEvent],
@@ -73,13 +77,15 @@ def _parse_classification(text: str) -> dict[str, list[int]]:
try:
parsed = json.loads(cleaned)
if not isinstance(parsed, dict):
logger.warning("Classifier returned non-dict JSON: %s", type(parsed).__name__)
return {}
raise ClassifierParseError(
f"classifier returned non-dict JSON: {type(parsed).__name__}"
)
return {
k: [int(i) for i in v]
for k, v in parsed.items()
if isinstance(v, list)
}
except (json.JSONDecodeError, ValueError, TypeError) as e:
logger.warning("Failed to parse classification output: %s", e)
return {}
raise ClassifierParseError(
f"failed to parse classification output: {e}"
) from e
@@ -0,0 +1,55 @@
"""Tests for scripts.classifier._parse_classification."""
import subprocess
from unittest.mock import patch
import pytest
from scripts import classifier
from scripts.classifier import _parse_classification
from scripts.grader import grade
from scripts.parser import ComplianceSpec, Detector, ObservationEvent, Step
class TestParseClassification:
def test_plain_json(self):
assert _parse_classification('{"a": [0, 2]}') == {"a": [0, 2]}
def test_fenced_json(self):
assert _parse_classification('```json\n{"a": [1]}\n```') == {"a": [1]}
def test_dict_without_list_values_is_empty(self):
assert _parse_classification('{"a": "x"}') == {}
def test_empty_object_is_valid(self):
assert _parse_classification("{}") == {}
def test_garbage_raises(self):
with pytest.raises(classifier.ClassifierParseError, match="failed to parse"):
_parse_classification("sorry, I cannot classify these")
def test_non_dict_json_raises(self):
with pytest.raises(classifier.ClassifierParseError, match="non-dict"):
_parse_classification("[1, 2, 3]")
def test_non_integer_indices_raise(self):
with pytest.raises(classifier.ClassifierParseError, match="failed to parse"):
_parse_classification('{"a": ["x"]}')
def test_is_runtime_error(self):
assert issubclass(classifier.ClassifierParseError, RuntimeError)
@pytest.mark.parametrize("response", ["not JSON", "[]", '{"a": ["x"]}'])
def test_grader_propagates_classifier_parse_failure(response):
"""An invalid successful subprocess response cannot become a zero score."""
spec = ComplianceSpec(
id="fixture", name="Fixture", source_rule="fixture", version="1",
steps=(Step("a", "Read a file", True, Detector("Read a file")),),
threshold_promote_to_hook=0.6,
)
trace = [ObservationEvent("T0000", "tool_complete", "Read", "fixture", "", "")]
result = subprocess.CompletedProcess(["claude"], 0, stdout=response, stderr="")
with patch("scripts.classifier.subprocess.run", return_value=result) as execute:
with pytest.raises(classifier.ClassifierParseError):
grade(spec, trace)
execute.assert_called_once()