fix(skill-comply): stop a failed step supplying evidence downstream (#3109)

* 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.
This commit is contained in:
Lukas
2026-09-21 14:39:20 -04:00
committed by GitHub
parent 7b7dfc6412
commit 8b951d3bf6
2 changed files with 326 additions and 11 deletions
+79 -9
View File
@@ -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,
+247 -2
View File
@@ -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 "")