Spaces:
Sleeping
Sleeping
File size: 7,215 Bytes
d9c2197 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 | """Tests for JCRSafetyGate.
Covers:
- Risk score computation across the role / candidate / shuffle / reuse axes
- INV-15: Critic with risk > threshold ALWAYS uses dense prefill
- Non-judge roles never trigger dense fallback
- gate_decision logging + summary stats
- Edge case: invalid args
"""
from __future__ import annotations
import pytest
from apohara_context_forge.safety.jcr_gate import (
JCRDecision,
JCRSafetyGate,
)
class TestJCRSafetyGateDefaults:
def test_default_threshold(self):
gate = JCRSafetyGate()
assert gate.jcr_threshold == 0.7
def test_invalid_threshold_rejected(self):
with pytest.raises(ValueError, match="must be in"):
JCRSafetyGate(jcr_threshold=1.5)
with pytest.raises(ValueError, match="must be in"):
JCRSafetyGate(jcr_threshold=-0.1)
class TestJCRRiskComputation:
def test_critic_base_risk(self):
gate = JCRSafetyGate()
risk = gate.compute_jcr_risk(
agent_role="critic",
candidate_count=2,
reuse_rate=0.5,
layout_shuffled=False,
)
assert risk == pytest.approx(0.6)
def test_non_critic_base_risk(self):
gate = JCRSafetyGate()
risk = gate.compute_jcr_risk(
agent_role="retriever",
candidate_count=2,
reuse_rate=0.5,
layout_shuffled=False,
)
assert risk == pytest.approx(0.1)
def test_extra_candidates_increase_risk(self):
gate = JCRSafetyGate()
baseline = gate.compute_jcr_risk("critic", 2, 0.0, False)
five = gate.compute_jcr_risk("critic", 5, 0.0, False)
assert five == pytest.approx(baseline + 0.3)
def test_layout_shuffled_increases_risk(self):
gate = JCRSafetyGate()
plain = gate.compute_jcr_risk("critic", 2, 0.0, False)
shuffled = gate.compute_jcr_risk("critic", 2, 0.0, True)
assert shuffled == pytest.approx(plain + 0.2)
def test_high_reuse_rate_increases_risk(self):
gate = JCRSafetyGate()
low = gate.compute_jcr_risk("critic", 2, 0.5, False)
high = gate.compute_jcr_risk("critic", 2, 0.95, False)
assert high == pytest.approx(low + 0.15)
def test_risk_clamped_to_one(self):
gate = JCRSafetyGate()
risk = gate.compute_jcr_risk(
agent_role="critic",
candidate_count=20,
reuse_rate=1.0,
layout_shuffled=True,
)
assert 0.0 <= risk <= 1.0
assert risk == pytest.approx(1.0)
def test_invalid_candidate_count_rejected(self):
gate = JCRSafetyGate()
with pytest.raises(ValueError, match="non-negative"):
gate.compute_jcr_risk("critic", -1, 0.5, False)
def test_invalid_reuse_rate_rejected(self):
gate = JCRSafetyGate()
with pytest.raises(ValueError, match="reuse_rate must be"):
gate.compute_jcr_risk("critic", 2, 1.5, False)
class TestINV15CriticAlwaysDense:
"""INV-15: Critic with risk > threshold ALWAYS returns use_dense=True."""
def test_critic_5_candidates_shuffle_uses_dense(self):
gate = JCRSafetyGate()
# Risk = 0.6 + 0.3 + 0.2 = 1.1 → clamped to 1.0 → > 0.7
assert gate.should_use_dense_prefill(
agent_role="critic",
candidate_count=5,
reuse_rate=0.5,
layout_shuffled=True,
) is True
def test_retriever_2_candidates_no_dense(self):
gate = JCRSafetyGate()
assert gate.should_use_dense_prefill(
agent_role="retriever",
candidate_count=2,
reuse_rate=0.5,
layout_shuffled=False,
) is False
def test_non_critic_never_uses_dense_even_with_high_risk(self):
"""Non-judge roles aren't protected by INV-15."""
gate = JCRSafetyGate()
# Even with all risk knobs cranked up, a retriever passes through.
assert gate.should_use_dense_prefill(
agent_role="retriever",
candidate_count=10,
reuse_rate=1.0,
layout_shuffled=True,
) is False
@pytest.mark.parametrize("candidates,shuffle,reuse", [
(5, True, 0.9),
(4, True, 0.85),
(8, False, 0.85),
(10, True, 0.5),
])
def test_critic_above_threshold_always_dense(self, candidates, shuffle, reuse):
"""Comprehensive sweep: Critic above threshold always dense (INV-15)."""
gate = JCRSafetyGate()
decision = gate.gate_decision(
agent_role="critic",
candidate_count=candidates,
reuse_rate=reuse,
layout_shuffled=shuffle,
)
if decision.risk_score > gate.jcr_threshold:
assert decision.use_dense is True, (
f"INV-15 violated: critic with risk {decision.risk_score} "
f"> threshold {gate.jcr_threshold} did not get dense prefill"
)
def test_critic_exactly_at_threshold_uses_reuse(self):
"""Threshold is strict: > threshold triggers dense, not >=."""
gate = JCRSafetyGate(jcr_threshold=0.6)
# Critic, 2 candidates, no shuffle, low reuse → exactly 0.6
decision = gate.gate_decision(
agent_role="critic",
candidate_count=2,
reuse_rate=0.5,
layout_shuffled=False,
)
assert decision.risk_score == pytest.approx(0.6)
assert decision.use_dense is False
class TestGateDecisionLogging:
def test_gate_decision_returns_structured_record(self):
gate = JCRSafetyGate()
decision = gate.gate_decision("critic", 5, 0.9, True)
assert isinstance(decision, JCRDecision)
assert decision.agent_role == "critic"
assert decision.use_dense is True
assert "INV-15" in decision.reason
assert decision.timestamp > 0
def test_log_accumulates(self):
gate = JCRSafetyGate()
for _ in range(3):
gate.gate_decision("critic", 5, 0.9, True)
gate.gate_decision("retriever", 2, 0.1, False)
assert len(gate.gate_log) == 4
def test_summary_aggregates(self):
gate = JCRSafetyGate()
gate.gate_decision("critic", 5, 0.9, True) # dense
gate.gate_decision("critic", 2, 0.1, False) # reuse
gate.gate_decision("retriever", 2, 0.1, False) # reuse
s = gate.summary()
assert s["total_decisions"] == 3
assert s["dense_fallback_count"] == 1
# 2 critic decisions, 1 dense → 0.5
assert s["critic_dense_rate"] == pytest.approx(0.5)
assert 0.0 <= s["avg_risk_score"] <= 1.0
def test_summary_empty_safe(self):
gate = JCRSafetyGate()
s = gate.summary()
assert s["total_decisions"] == 0
assert s["dense_fallback_count"] == 0
assert s["avg_risk_score"] == 0.0
assert s["critic_dense_rate"] == 0.0
def test_role_case_insensitive(self):
gate = JCRSafetyGate()
# Upper-case role still resolves to "critic".
decision = gate.gate_decision("CRITIC", 5, 0.9, True)
assert decision.agent_role == "critic"
assert decision.use_dense is True
|