Spaces:
Running
Running
File size: 4,085 Bytes
8835ca1 | 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 | """Focused P0 tests for EngineeringState's safety and rollout contract."""
from __future__ import annotations
import os
import sys
import unittest
from unittest.mock import patch
_BACKEND = os.path.join(os.path.dirname(__file__), "..")
if _BACKEND not in sys.path:
sys.path.insert(0, _BACKEND)
from agents.engineering_state import ( # noqa: E402
EngineeringState,
EngineeringStateConfig,
EngineeringStateMode,
SCHEMA_VERSION,
redact_text,
)
class TestEngineeringState(unittest.TestCase):
def test_default_rollout_is_authoritative_and_invalid_mode_fails_closed(self) -> None:
with patch.dict(os.environ, {}, clear=True):
self.assertEqual(EngineeringStateConfig.from_env().mode, EngineeringStateMode.AUTHORITATIVE)
with patch.dict(os.environ, {"ENGINEERING_STATE_MODE": "unsafe"}, clear=False):
self.assertEqual(EngineeringStateConfig.from_env().mode, EngineeringStateMode.OFF)
def test_redaction_removes_common_credentials(self) -> None:
value = "Authorization: Bearer abcdefghijkl token=ghp_1234567890abcdef hf_1234567890"
result = redact_text(value)
self.assertNotIn("abcdefghijkl", result)
self.assertNotIn("ghp_1234567890abcdef", result)
self.assertNotIn("hf_1234567890", result)
self.assertIn("[REDACTED]", result)
def test_transitions_are_validated_and_idempotent(self) -> None:
state = EngineeringState.start("build a safe agent", run_id="run-1", now_ms=100)
self.assertTrue(state.transition("CLASSIFYING", now_ms=101))
self.assertFalse(state.transition("CLASSIFYING", now_ms=102))
with self.assertRaises(ValueError):
state.transition("IDLE", now_ms=103)
self.assertEqual(state.revision, 1)
self.assertEqual(state.sequence, 1)
def test_round_trip_is_bounded_and_does_not_store_raw_goal(self) -> None:
goal = "use token=super-secret-value to build this agent"
state = EngineeringState.start(goal, run_id="run-2", session_id="session-2", now_ms=100)
for target in ("CLASSIFYING", "THINKING", "COMPLETED"):
state.transition(target, now_ms=101)
snapshot = state.snapshot()
restored = EngineeringState.from_snapshot(snapshot)
self.assertEqual(restored.snapshot(), snapshot)
self.assertEqual(snapshot["schema_version"], SCHEMA_VERSION)
self.assertNotIn("super-secret-value", str(snapshot))
self.assertLessEqual(len(snapshot["history"]), 64)
def test_corrupt_schema_and_revision_are_rejected(self) -> None:
state = EngineeringState.start("goal", run_id="run-3")
snapshot = state.snapshot()
snapshot["schema_version"] = 999
with self.assertRaises(ValueError):
EngineeringState.from_snapshot(snapshot)
snapshot = state.snapshot()
snapshot["revision"] = -1
with self.assertRaises(ValueError):
EngineeringState.from_snapshot(snapshot)
def test_canary_selection_is_deterministic_and_requires_session(self) -> None:
config = EngineeringStateConfig(EngineeringStateMode.CANARY, 0.5)
self.assertFalse(config.selects_canary("run", ""))
self.assertEqual(
config.selects_canary("run", "session"),
config.selects_canary("run", "session"),
)
def test_resume_normalizes_terminal_state_and_preserves_history(self) -> None:
state = EngineeringState.start("resume this task", run_id="run-4", session_id="session-4")
for target in ("CLASSIFYING", "THINKING", "COMPLETED"):
state.transition(target)
history_before_resume = list(state.history)
state.prepare_for_resume()
self.assertEqual(state.current_state, "IDLE")
self.assertEqual(state.status, "active")
self.assertEqual(state.history[:len(history_before_resume)], history_before_resume)
self.assertTrue(any("resume normalized state to IDLE" in item for item in state.diagnostics))
if __name__ == "__main__":
unittest.main(verbosity=2)
|