File size: 3,999 Bytes
302b72d | 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 | from osint_core.scheduler import (
INVARIANT_CHECKS,
DecisionPacket,
ScheduleReason,
ShortcutDebt,
SystemState,
fits_deadline,
safe_utility,
schedule_decision,
total_required_time_ms,
)
def make_packet(**overrides):
data = {
"intent_id": "intent_123",
"action": "enrich_indicator",
"risk_label": "low",
"confidence": 0.90,
"reversibility": 0.90,
"deadline_ms": 1000,
"verification_cost_ms": 100,
"execution_cost_ms": 100,
"rollback_cost_ms": 100,
"expected_utility_decay": 0.10,
"required_checks": INVARIANT_CHECKS,
"rollback_plan": "observe_only",
"uncertainty_notes": (),
}
data.update(overrides)
return DecisionPacket(**data)
def test_total_required_time_and_deadline_fit():
packet = make_packet()
assert total_required_time_ms(packet) == 300
assert fits_deadline(packet) is True
def test_fast_path_for_low_risk_reversible_action():
decision = schedule_decision(make_packet(risk_label="low"))
assert decision.route == "FAST"
assert decision.allowed is True
assert decision.reason == ScheduleReason.FAST_PATH_AVAILABLE
def test_deliberative_path_for_high_risk_action_that_fits_deadline():
packet = make_packet(
risk_label="high",
confidence=0.90,
reversibility=0.70,
rollback_plan="sandbox",
rollback_cost_ms=200,
deadline_ms=1000,
)
decision = schedule_decision(packet)
assert decision.route == "DELIBERATIVE"
assert decision.allowed is True
def test_invariant_checks_cannot_be_skipped():
packet = make_packet(required_checks=("scope_validated",))
decision = schedule_decision(packet)
assert decision.route == "FAIL_CLOSED"
assert decision.allowed is False
assert decision.reason == ScheduleReason.INVARIANT_VIOLATION
assert "hash_salt_present" in decision.skipped_checks
def test_high_risk_without_rollback_fails_closed():
packet = make_packet(
risk_label="critical",
rollback_plan="",
rollback_cost_ms=0,
)
decision = schedule_decision(packet)
assert decision.route == "FAIL_CLOSED"
assert decision.reason == ScheduleReason.MISSING_ROLLBACK
def test_deadline_too_tight_routes_to_containment_when_reversible():
packet = make_packet(
deadline_ms=100,
verification_cost_ms=100,
execution_cost_ms=100,
rollback_cost_ms=100,
reversibility=0.90,
)
decision = schedule_decision(packet)
assert decision.route == "CONTAINMENT"
assert decision.reason == ScheduleReason.DEADLINE_TOO_TIGHT
assert decision.authority_scale == 0.25
def test_deadline_too_tight_and_not_reversible_fails_closed():
packet = make_packet(
deadline_ms=100,
reversibility=0.20,
)
decision = schedule_decision(packet)
assert decision.route == "FAIL_CLOSED"
assert decision.reason == ScheduleReason.NO_SAFE_ACTION_FITS
def test_shortcut_debt_forces_containment():
state = SystemState(shortcut_debt=ShortcutDebt(emergency_overrides=2), shortcut_debt_limit=0.70)
decision = schedule_decision(make_packet(), state)
assert decision.route == "CONTAINMENT"
assert decision.reason == ScheduleReason.SHORTCUT_DEBT_TOO_HIGH
def test_contested_trust_state_forces_containment():
state = SystemState(trust_state="contested")
decision = schedule_decision(make_packet(), state)
assert decision.route == "CONTAINMENT"
assert decision.reason == ScheduleReason.TRUST_STATE_DEGRADED
def test_low_confidence_high_risk_forces_containment():
packet = make_packet(risk_label="high", confidence=0.20, rollback_plan="sandbox")
decision = schedule_decision(packet)
assert decision.route == "CONTAINMENT"
assert decision.reason == ScheduleReason.LOW_CONFIDENCE
def test_safe_utility_is_bounded():
score = safe_utility(make_packet())
assert 0.0 <= score <= 1.0
|