From 8b951d3bf63b5f3dcdac3f255c8a15c23fbf35f3 Mon Sep 17 00:00:00 2001 From: Lukas <103962359+L4XB@users.noreply.github.com> Date: Mon, 21 Sep 2026 20:39:20 +0200 Subject: [PATCH] fix(skill-comply): stop a failed step supplying evidence downstream (#3109) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(skill-comply): stop a failed step supplying evidence downstream `_check_temporal_order` fell back to the raw classifier output whenever the referenced step was absent from `resolved`. A step only enters `resolved` once it passes, so "failed" and "not graded yet" were the same thing to that lookup, and a dependant could pass on an event belonging to a prerequisite that had failed its own ordering check. With three steps C, A (before C) and B (after A) and events C@T0, A@T1, B@T2, A fails and B passed on A's classified event: 2/3 instead of 1/3. It compounds down a chain, so one failed prerequisite could leave a five-step workflow reading 4/5. The grader now tracks which steps have been graded at all. A referenced step that was graded and is missing from `resolved` failed, and its events are refused with a reason that says so. A step not graded yet is a forward reference to a step declared later, and the fallback stays as it was: that is what makes an out-of-order declaration work, and the existing regression for it goes red if the fallback is removed instead. `before_step` deliberately keeps the old fallback. The two fail in opposite directions: an `after_step` fallback can only turn a failure into a pass, a `before_step` one can only turn a pass into a failure, so dropping it would relax a constraint because some other step failed. * fix(skill-comply): revoke a pass that rested on a later-failing prerequisite Review of #3109 found the mirror image of the case that PR fixes. `graded` only catches a prerequisite that had already failed when its dependant was graded. A step declared *before* its `after_step` is graded against the classifier's raw events for a step that has not run yet — the fallback that makes an out-of-order declaration work — and nothing revisited it once that step went on to fail its own checks. Add a pass after grading that demotes any detected step whose `after_step` ended up failing, repeated to a fixed point: one demotion can invalidate whatever depended on it, in either declaration order. Demotion only removes passes, so it terminates. `compliance_rate` is computed from the demoted results. Also from review: build `graded` and the chain test's step list as new objects rather than mutating (AGENTS.md immutability rule), and annotate the injected mocks in the tests this PR owns. --- skills/skill-comply/scripts/grader.py | 88 +++++++- skills/skill-comply/tests/test_grader.py | 249 ++++++++++++++++++++++- 2 files changed, 326 insertions(+), 11 deletions(-) diff --git a/skills/skill-comply/scripts/grader.py b/skills/skill-comply/scripts/grader.py index 516beb18b..1209d042a 100644 --- a/skills/skill-comply/scripts/grader.py +++ b/skills/skill-comply/scripts/grader.py @@ -2,7 +2,7 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, replace from scripts.classifier import classify_events from scripts.parser import ComplianceSpec, ObservationEvent, Step @@ -30,23 +30,38 @@ def _check_temporal_order( event: ObservationEvent, resolved: dict[str, list[ObservationEvent]], classified: dict[str, list[ObservationEvent]], + graded: set[str], ) -> str | None: """Check before_step/after_step constraints. Returns failure reason or None.""" if step.detector.after_step is not None: - after_events = resolved.get(step.detector.after_step) + after_step = step.detector.after_step + after_events = resolved.get(after_step) if after_events is None: - after_events = classified.get(step.detector.after_step, []) + if after_step in graded: + # Graded and missing from `resolved` means it failed its own checks. + # Its classified events are not evidence for anything: reusing them + # here let a dependant pass on the strength of a failed prerequisite, + # and the overstatement carried down the whole chain. + return f"after_step '{after_step}' did not pass its own checks" + # Not graded yet, so this is a forward reference to a step declared later. + # The classifier's own output is the only thing available, and using it is + # what makes an out-of-order declaration work. + after_events = classified.get(after_step, []) if not after_events: - return f"after_step '{step.detector.after_step}' not yet detected" + return f"after_step '{after_step}' not yet detected" latest_after = max(e.timestamp for e in after_events) if event.timestamp <= latest_after: return ( - f"must occur after '{step.detector.after_step}' " + f"must occur after '{after_step}' " f"(last at {latest_after}), but found at {event.timestamp}" ) if step.detector.before_step is not None: - # Look ahead using LLM classification results + # Look ahead using LLM classification results. A failed reference step is NOT + # excluded here the way it is above: the two fall in opposite directions. An + # `after_step` fallback can only turn a failure into a pass, while a + # `before_step` one can only turn a pass into a failure, so dropping it would + # relax a constraint because some other step failed. before_events = resolved.get(step.detector.before_step) if before_events is None: before_events = classified.get(step.detector.before_step, []) @@ -61,6 +76,46 @@ def _check_temporal_order( return None +def _demote_steps_resting_on_failures( + step_results: tuple[StepResult, ...], + after_steps: dict[str, str], +) -> tuple[StepResult, ...]: + """Undo passes that rest on an `after_step` which ended up failing. + + A step declared before its prerequisite is graded against the classifier's raw + events for that step, because `resolved` has nothing for it yet. That fallback is + what makes an out-of-order declaration work, but the prerequisite can go on to + fail its own checks, and then the dependant is left passing on evidence that + never held. Repeat until nothing changes: demoting one step can invalidate + whatever depended on it, whichever order the two were declared in. Demotion only + ever removes passes, so the loop terminates. + """ + results = step_results + while True: + failed = {result.step_id for result in results if not result.detected} + demote = { + result.step_id + for result in results + if result.detected and after_steps.get(result.step_id) in failed + } + if not demote: + return results + results = tuple( + replace( + result, + detected=False, + evidence=(), + failure_reason=( + f"after_step '{after_steps[result.step_id]}' " + "did not pass its own checks" + ), + ) + if result.step_id in demote + else result + for result in results + ) + + def grade( spec: ComplianceSpec, trace: list[ObservationEvent], @@ -80,6 +135,9 @@ def grade( # Step 2: Check temporal ordering (deterministic) resolved: dict[str, list[ObservationEvent]] = {} + # Steps already graded, pass or fail. `resolved` alone cannot tell "failed" from + # "not reached yet", and those need opposite answers in `_check_temporal_order`. + graded: set[str] = set() step_results: list[StepResult] = [] for step in spec.steps: @@ -88,7 +146,7 @@ def grade( failure_reason: str | None = None for event in candidates: - temporal_fail = _check_temporal_order(step, event, resolved, classified) + temporal_fail = _check_temporal_order(step, event, resolved, classified, graded) if temporal_fail is None: matched.append(event) break @@ -101,6 +159,7 @@ def grade( elif failure_reason is None: failure_reason = f"no matching event classified for step '{step.id}'" + graded = graded | {step.id} step_results.append(StepResult( step_id=step.id, detected=detected, @@ -108,8 +167,19 @@ def grade( failure_reason=failure_reason if not detected else None, )) + # `graded` catches a prerequisite that had already failed when its dependant was + # graded. The other direction needs a second pass: a dependant declared first is + # graded against the classifier's raw events for a prerequisite that has not run + # yet, and only later does that prerequisite fail. + after_steps = { + step.id: step.detector.after_step + for step in spec.steps + if step.detector.after_step is not None + } + resolved_results = _demote_steps_resting_on_failures(tuple(step_results), after_steps) + required_ids = {s.id for s in spec.steps if s.required} - required_steps = [s for s in step_results if s.step_id in required_ids] + required_steps = [s for s in resolved_results if s.step_id in required_ids] detected_required = sum(1 for s in required_steps if s.detected) total_required = len(required_steps) @@ -117,7 +187,7 @@ def grade( return ComplianceResult( spec_id=spec.id, - steps=tuple(step_results), + steps=resolved_results, compliance_rate=compliance_rate, recommend_hook_promotion=compliance_rate < spec.threshold_promote_to_hook, classification=classification, diff --git a/skills/skill-comply/tests/test_grader.py b/skills/skill-comply/tests/test_grader.py index a95825266..8ca056452 100644 --- a/skills/skill-comply/tests/test_grader.py +++ b/skills/skill-comply/tests/test_grader.py @@ -1,7 +1,7 @@ """Tests for grader module — compliance scoring with LLM classification.""" from pathlib import Path -from unittest.mock import patch +from unittest.mock import MagicMock, patch import pytest @@ -137,7 +137,7 @@ class TestGradeEdgeCases: assert result.spec_id == "tdd-workflow" @patch("scripts.grader.classify_events") - def test_after_step_can_reference_later_declared_spec_step(self, mock_cls) -> None: + def test_after_step_can_reference_later_declared_spec_step(self, mock_cls: MagicMock) -> None: spec = ComplianceSpec( id="out-of-order-after-step", name="Out of order after_step", @@ -195,3 +195,248 @@ class TestGradeEdgeCases: assert step_a.failure_reason is None assert step_b.detected is True assert result.compliance_rate == 1.0 + + @patch("scripts.grader.classify_events") + def test_after_step_does_not_reuse_a_step_that_failed_its_own_constraint( + self, mock_cls: MagicMock + ) -> None: + """A step that failed may not supply evidence to a step that depends on it (#3108). + + `resolved` only receives a step once it passes, so a dependant fell back to the + raw classifier output and could pass on an event belonging to a failed parent. + The fallback exists for forward references, which the case above covers; a + parent that has already been graded and failed is a different thing. + """ + spec = ComplianceSpec( + id="failed-parent-evidence", + name="Failed parent evidence", + source_rule="rules/common/testing.md", + version="1.0", + steps=( + Step( + id="C", + description="Reference step with no constraint of its own", + required=True, + detector=Detector(description="Event C"), + ), + Step( + id="A", + description="Must occur before C, and does not", + required=True, + detector=Detector(description="Event A", before_step="C"), + ), + Step( + id="B", + description="Depends on A", + required=True, + detector=Detector(description="Event B", after_step="A"), + ), + ), + threshold_promote_to_hook=0.5, + ) + trace = [ + ObservationEvent( + timestamp=f"2026-03-20T10:00:0{index}Z", + event="tool_complete", + tool="Write", + session="sess-evidence", + input=f'{{"file_path":"src/{name}.py"}}', + output=f"step {name}", + ) + for index, name in enumerate(("c", "a", "b")) + ] + mock_cls.return_value = {"C": [0], "A": [1], "B": [2]} + + result = grade(spec, trace) + + detected = {step.step_id: step.detected for step in result.steps} + assert detected["C"] is True + assert detected["A"] is False + assert detected["B"] is False + assert result.compliance_rate == pytest.approx(1 / 3) + step_b = next(step for step in result.steps if step.step_id == "B") + assert "A" in (step_b.failure_reason or "") + + @patch("scripts.grader.classify_events") + def test_a_failed_prerequisite_does_not_carry_a_chain_of_passes( + self, mock_cls: MagicMock + ) -> None: + """The overstatement compounds: B on A, C on B, D on C (#3108). + + Each link used to pass on the classifier's raw output for the link above it, so + one failed prerequisite could leave a four-step workflow reading 3/4 compliant. + """ + steps = ( + Step( + id="Z", + description="Reference step with no constraint of its own", + required=True, + detector=Detector(description="Event Z"), + ), + Step( + id="A", + description="Must occur before Z, and does not", + required=True, + detector=Detector(description="Event A", before_step="Z"), + ), + *( + Step( + id=later, + description=f"Depends on {earlier}", + required=True, + detector=Detector(description=f"Event {later}", after_step=earlier), + ) + for earlier, later in (("A", "B"), ("B", "C"), ("C", "D")) + ), + ) + spec = ComplianceSpec( + id="failed-prerequisite-chain", + name="Failed prerequisite chain", + source_rule="rules/common/testing.md", + version="1.0", + steps=steps, + threshold_promote_to_hook=0.5, + ) + names = ("z", "a", "b", "c", "d") + trace = [ + ObservationEvent( + timestamp=f"2026-03-20T10:00:0{index}Z", + event="tool_complete", + tool="Write", + session="sess-chain", + input=f'{{"file_path":"src/{name}.py"}}', + output=f"step {name}", + ) + for index, name in enumerate(names) + ] + mock_cls.return_value = {"Z": [0], "A": [1], "B": [2], "C": [3], "D": [4]} + + result = grade(spec, trace) + + detected = {step.step_id: step.detected for step in result.steps} + assert detected == {"Z": True, "A": False, "B": False, "C": False, "D": False} + assert result.compliance_rate == pytest.approx(1 / 5) + + @patch("scripts.grader.classify_events") + def test_forward_reference_is_revoked_when_the_prerequisite_later_fails( + self, mock_cls: MagicMock + ) -> None: + """The mirror image of the case above, raised in review of #3109. + + `graded` only catches a prerequisite that had already failed. A step declared + *before* its prerequisite is graded against the classifier's raw events for a + step that has not run yet — the fallback that makes an out-of-order + declaration work — and nothing revisited it once that step failed. + """ + spec = ComplianceSpec( + id="forward-reference-revoked", + name="Forward reference revoked", + source_rule="rules/common/testing.md", + version="1.0", + steps=( + Step( + id="Z", + description="Reference step with no constraint of its own", + required=True, + detector=Detector(description="Event Z"), + ), + Step( + id="B", + description="Depends on A, which is declared after it", + required=True, + detector=Detector(description="Event B", after_step="A"), + ), + Step( + id="A", + description="Must occur before Z, and does not", + required=True, + detector=Detector(description="Event A", before_step="Z"), + ), + ), + threshold_promote_to_hook=0.5, + ) + trace = [ + ObservationEvent( + timestamp=f"2026-03-20T10:00:0{index}Z", + event="tool_complete", + tool="Write", + session="sess-forward", + input=f'{{"file_path":"src/{name}.py"}}', + output=f"step {name}", + ) + for index, name in enumerate(("z", "a", "b")) + ] + mock_cls.return_value = {"Z": [0], "A": [1], "B": [2]} + + result = grade(spec, trace) + + detected = {step.step_id: step.detected for step in result.steps} + assert detected == {"Z": True, "A": False, "B": False} + assert result.compliance_rate == pytest.approx(1 / 3) + step_b = next(step for step in result.steps if step.step_id == "B") + assert step_b.evidence == () + assert "A" in (step_b.failure_reason or "") + + @patch("scripts.grader.classify_events") + def test_revocation_reaches_a_step_that_referenced_the_dependant_backwards( + self, mock_cls: MagicMock + ) -> None: + """One demotion invalidates the next, in either declaration order. + + C is declared after B and passes on `resolved`, the ordinary backward + reference — B was still detected at the time. B is the forward-reference case + above and comes down with A, so C has to follow, which takes a second pass. + """ + spec = ComplianceSpec( + id="revocation-propagates", + name="Revocation propagates", + source_rule="rules/common/testing.md", + version="1.0", + steps=( + Step( + id="Z", + description="Reference step with no constraint of its own", + required=True, + detector=Detector(description="Event Z"), + ), + Step( + id="B", + description="Depends on A, which is declared after it", + required=True, + detector=Detector(description="Event B", after_step="A"), + ), + Step( + id="A", + description="Must occur before Z, and does not", + required=True, + detector=Detector(description="Event A", before_step="Z"), + ), + Step( + id="C", + description="Depends on B, which is declared before it", + required=True, + detector=Detector(description="Event C", after_step="B"), + ), + ), + threshold_promote_to_hook=0.5, + ) + trace = [ + ObservationEvent( + timestamp=f"2026-03-20T10:00:0{index}Z", + event="tool_complete", + tool="Write", + session="sess-propagate", + input=f'{{"file_path":"src/{name}.py"}}', + output=f"step {name}", + ) + for index, name in enumerate(("z", "a", "b", "c")) + ] + mock_cls.return_value = {"Z": [0], "A": [1], "B": [2], "C": [3]} + + result = grade(spec, trace) + + detected = {step.step_id: step.detected for step in result.steps} + assert detected == {"Z": True, "A": False, "B": False, "C": False} + assert result.compliance_rate == pytest.approx(1 / 4) + step_c = next(step for step in result.steps if step.step_id == "C") + assert "B" in (step_c.failure_reason or "")