61 lines
1.9 KiB
Python
61 lines
1.9 KiB
Python
from laya_router.config import Settings
|
|
from laya_router.evaluate import evaluate_event
|
|
from laya_router.mcp import handle_message
|
|
from laya_router.trajectory import SessionStore
|
|
|
|
|
|
class _Agent:
|
|
def predict(self, state, questions):
|
|
answers = {}
|
|
for name, spec in questions.items():
|
|
if spec["type"] == "noul":
|
|
value = 0.9 if name in {"is_stuck", "destructive", "claims_done_without_verification"} else 0.1
|
|
answers[name] = {"noul": value}
|
|
return {"answers": answers}
|
|
|
|
|
|
def test_stop_blocks_only_once_per_turn(tmp_path):
|
|
settings = Settings(home=tmp_path, skills=False)
|
|
store = SessionStore()
|
|
body = {
|
|
"event": "PostToolUse",
|
|
"session_id": "s",
|
|
"tool_name": "Edit",
|
|
"tool_input": {"file_path": "src/app.py"},
|
|
"client": "claude",
|
|
}
|
|
evaluate_event(_Agent(), settings, store, body)
|
|
stop = {
|
|
"event": "Stop",
|
|
"session_id": "s",
|
|
"turn_id": "t1",
|
|
"client": "codex",
|
|
"stop_hook_active": False,
|
|
}
|
|
first = evaluate_event(_Agent(), settings, store, stop)
|
|
assert first["action"] == "block"
|
|
second = evaluate_event(_Agent(), settings, store, stop)
|
|
assert second["action"] == "silent"
|
|
|
|
|
|
def test_read_is_recorded_without_a_hint(tmp_path):
|
|
settings = Settings(home=tmp_path)
|
|
outcome = evaluate_event(
|
|
_Agent(),
|
|
settings,
|
|
SessionStore(),
|
|
{
|
|
"event": "PostToolUse",
|
|
"session_id": "s",
|
|
"tool_name": "Read",
|
|
"tool_input": {"file_path": "src/app.py"},
|
|
},
|
|
)
|
|
assert outcome["action"] == "silent"
|
|
|
|
|
|
def test_mcp_lists_the_judge_tool():
|
|
listed = handle_message({"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
|
|
assert listed["result"]["tools"][0]["name"] == "laya_judge"
|
|
assert handle_message({"jsonrpc": "2.0", "method": "notifications/initialized"}) is None
|