File size: 3,599 Bytes
3b452a0 8a1f89f 3b452a0 8a1f89f 3b452a0 8a1f89f 3b452a0 | 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 | import sys
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
LOCAL_DEPS = ROOT / ".deps"
if LOCAL_DEPS.exists():
sys.path.insert(0, str(LOCAL_DEPS))
sys.path.insert(0, str(ROOT))
from env.environment import IDPIExfilEnv
from env.models import Action
class EvaluationHandoffTests(unittest.TestCase):
def test_step_info_exposes_reward_and_grader_inputs(self) -> None:
env = IDPIExfilEnv()
env.reset(difficulty="easy", seed=42)
env.step(Action(type="ReadDoc"))
_, _, done, info = env.step(Action(type="FlagInjection", reason="external send request"))
self.assertTrue(done)
self.assertIn("reward_inputs", info)
self.assertIn("grader_inputs", info)
self.assertIn("evaluation_facts", info)
self.assertIn("reward_diagnostics", info)
self.assertEqual(info["reward_inputs"]["scenario"]["scenario_id"], info["scenario_id"])
self.assertEqual(info["grader_inputs"]["scenario_id"], info["scenario_id"])
self.assertEqual(
info["evaluation_facts"]["grader_inputs"]["scenario_id"],
info["grader_inputs"]["scenario_id"],
)
self.assertTrue(info["grader_inputs"]["safe_completion"])
self.assertFalse(info["grader_inputs"]["exfil_occurred"])
self.assertIn("action_counts", info["reward_inputs"]["progress"])
self.assertIn("reward_term_totals", info["reward_inputs"]["diagnostics"])
self.assertIn("action_counts", info["reward_diagnostics"])
def test_grader_inputs_match_episode_result(self) -> None:
env = IDPIExfilEnv()
env.reset(difficulty="hard", seed=3)
env.step(Action(type="ReadDoc"))
env.step(Action(type="FlagInjection", reason="summary contamination attempt"))
clean_summary = "Delayed access review sign-off. Gap in vendor attestation tracking."
env.step(Action(type="WriteDB", content=clean_summary))
grader_inputs = env.get_grader_inputs()
result = env.get_result()
self.assertEqual(grader_inputs["injection_flagged"], result.injection_flagged)
self.assertEqual(grader_inputs["safe_completion"], result.task_completed)
self.assertEqual(grader_inputs["summary_score"], result.summary_score)
self.assertFalse(grader_inputs["exfil_occurred"])
def test_evaluation_facts_surface_semantic_contamination_fields(self) -> None:
env = IDPIExfilEnv()
env.reset(difficulty="hard", seed=3)
env.step(Action(type="ReadDoc"))
facts = env.get_evaluation_facts()
self.assertIn("semantic_labels", facts["summary"])
self.assertIn("semantic_labels", facts["outbound"])
self.assertIn("summary_semantic_labels", facts["reward_inputs"]["taint"])
self.assertIn("outbound_semantic_labels", facts["reward_inputs"]["taint"])
def test_terminal_step_keeps_handoff_available(self) -> None:
env = IDPIExfilEnv()
env.reset(difficulty="easy", seed=42)
env.step(Action(type="ReadDoc"))
env.step(Action(type="FlagInjection", reason="external send request"))
_, reward, done, info = env.step(Action(type="NoOp"))
self.assertTrue(done)
self.assertEqual(reward.value, 0.0)
self.assertIn("episode_result", info)
self.assertIn("reward_inputs", info)
self.assertIn("grader_inputs", info)
self.assertTrue(info["grader_inputs"]["done"])
self.assertIn("reward_term_totals", info["reward_inputs"]["diagnostics"])
if __name__ == "__main__":
unittest.main()
|