105 lines
3.5 KiB
Python
105 lines
3.5 KiB
Python
|
import json
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from scripts.run_semantic_review import build_packet, consensus, packet_batches
|
||
|
|
|
||
|
|
|
||
|
|
def _judge(verdict: str, confidence: float = 0.9):
|
||
|
|
return {
|
||
|
|
"assessments": [{"claim_id": "claim-one", "verdict": verdict, "confidence": confidence}]
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def test_semantic_consensus_is_advisory_and_cannot_accept():
|
||
|
|
result = consensus(_judge("supported"), _judge("supported"), model_ids=("judge-a", "judge-b"))
|
||
|
|
|
||
|
|
assert result["assessments"][0]["outcome"] == "advisory_support_consensus"
|
||
|
|
assert result["owner_decisions_written"] == 0
|
||
|
|
assert result["canonical_records_written"] == 0
|
||
|
|
|
||
|
|
|
||
|
|
def test_semantic_disagreement_and_low_confidence_defer():
|
||
|
|
disagreement = consensus(
|
||
|
|
_judge("supported"), _judge("partially_supported"), model_ids=("judge-a", "judge-b")
|
||
|
|
)
|
||
|
|
low = consensus(
|
||
|
|
_judge("supported", 0.79), _judge("supported"), model_ids=("judge-a", "judge-b")
|
||
|
|
)
|
||
|
|
|
||
|
|
assert (
|
||
|
|
disagreement["assessments"][0]["outcome"] == "defer_semantic_disagreement_or_low_confidence"
|
||
|
|
)
|
||
|
|
assert low["assessments"][0]["outcome"] == "defer_semantic_disagreement_or_low_confidence"
|
||
|
|
|
||
|
|
|
||
|
|
def test_semantic_consensus_requires_identical_claim_sets():
|
||
|
|
with pytest.raises(ValueError, match="different claim ID sets"):
|
||
|
|
consensus(
|
||
|
|
_judge("supported"),
|
||
|
|
{"assessments": [{"claim_id": "claim-two", "verdict": "supported", "confidence": 0.9}]},
|
||
|
|
model_ids=("judge-a", "judge-b"),
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_semantic_packet_keeps_evidence_and_policy_status_without_authority():
|
||
|
|
items = [
|
||
|
|
{
|
||
|
|
"source_stage": "review",
|
||
|
|
"evaluation": {"status": "needs-review"},
|
||
|
|
"proposal": {
|
||
|
|
"proposal_id": "proposal-1",
|
||
|
|
"revision": 2,
|
||
|
|
"operations": [
|
||
|
|
{
|
||
|
|
"node": {
|
||
|
|
"node_id": "node-1",
|
||
|
|
"title": "Finding",
|
||
|
|
"claims": [
|
||
|
|
{
|
||
|
|
"claim_id": "claim-one",
|
||
|
|
"text": "The claim.",
|
||
|
|
"evidence": [{"quote": "The evidence."}],
|
||
|
|
}
|
||
|
|
],
|
||
|
|
}
|
||
|
|
}
|
||
|
|
],
|
||
|
|
},
|
||
|
|
}
|
||
|
|
]
|
||
|
|
|
||
|
|
packet = build_packet(items, input_sha256="f" * 64)
|
||
|
|
|
||
|
|
assert packet["claims"][0]["policy_status"] == "needs-review"
|
||
|
|
assert packet["claims"][0]["evidence"][0]["quote"] == "The evidence."
|
||
|
|
assert any("untrusted data" in line for line in packet["review_instructions"])
|
||
|
|
|
||
|
|
|
||
|
|
def test_semantic_batches_stay_under_cli_argument_limit():
|
||
|
|
items = [
|
||
|
|
{
|
||
|
|
"proposal": {
|
||
|
|
"proposal_id": f"proposal-{index}",
|
||
|
|
"operations": [
|
||
|
|
{
|
||
|
|
"node": {
|
||
|
|
"claims": [
|
||
|
|
{
|
||
|
|
"claim_id": f"claim-{index}",
|
||
|
|
"text": "A" * 10_000,
|
||
|
|
"evidence": [{"quote": "B" * 10_000}],
|
||
|
|
}
|
||
|
|
]
|
||
|
|
}
|
||
|
|
}
|
||
|
|
],
|
||
|
|
}
|
||
|
|
}
|
||
|
|
for index in range(10)
|
||
|
|
]
|
||
|
|
packet = build_packet(items, input_sha256="a" * 64)
|
||
|
|
|
||
|
|
batches = packet_batches(packet)
|
||
|
|
|
||
|
|
assert len(batches) > 1
|
||
|
|
assert all(len(json.dumps(batch, ensure_ascii=False).encode()) < 131_000 for batch in batches)
|