feat: 优化运行时判定

tests/test_policy.py

@@ -0,0 +1,102 @@
+from laya_router.policy import decide_post_tool, decide_pre_tool, decide_stop
+
+
+def test_post_tool_dedupes_and_escalates_when_stuck():
+    signals = {"consecutive_failures": 0, "repeated_command": 1}
+    scores = {"is_stuck": 0.1, "should_replan": 0.1, "needs_more_context": 0.1}
+    first = decide_post_tool(enabled=True, signals=signals, scores=scores, last_hint="")
+    assert first.action == "inject"
+    again = decide_post_tool(enabled=True, signals=signals, scores=scores, last_hint=first.context)
+    assert again.action == "silent"
+    stuck = decide_post_tool(
+        enabled=True,
+        signals={"consecutive_failures": 2, "repeated_command": 2},
+        scores={"is_stuck": 0.9},
+        last_hint="",
+    )
+    assert "replan" in stuck.context
+
+
+def test_gate_never_allows_and_codex_does_not_ask():
+    quiet = decide_pre_tool(
+        trajectory_enabled=True,
+        gate_enabled=True,
+        client="claude",
+        tool_name="Bash",
+        signals={"is_edit": False, "target_unread": False},
+        edit_scores={},
+        gate_scores={"destructive": 0.1, "irreversible": 0.1, "touches_secrets_or_prod": 0.1},
+    )
+    assert quiet.action == "silent"
+    medium = decide_pre_tool(
+        trajectory_enabled=False,
+        gate_enabled=True,
+        client="claude",
+        tool_name="Bash",
+        signals={},
+        edit_scores={},
+        gate_scores={"destructive": 0.6, "irreversible": 0.5, "touches_secrets_or_prod": 0.4},
+    )
+    assert medium.action == "ask"
+    codex = decide_pre_tool(
+        trajectory_enabled=False,
+        gate_enabled=True,
+        client="codex",
+        tool_name="Bash",
+        signals={},
+        edit_scores={},
+        gate_scores={"destructive": 0.6, "irreversible": 0.5, "touches_secrets_or_prod": 0.4},
+    )
+    assert codex.action == "inject"
+    denied = decide_pre_tool(
+        trajectory_enabled=False,
+        gate_enabled=True,
+        client="codex",
+        tool_name="Bash",
+        signals={},
+        edit_scores={},
+        gate_scores={"destructive": 0.95, "irreversible": 0.2, "touches_secrets_or_prod": 0.1},
+    )
+    assert denied.action == "deny"
+    assert denied.action != "allow"
+
+
+def test_unread_edit_injects_hint():
+    decision = decide_pre_tool(
+        trajectory_enabled=True,
+        gate_enabled=False,
+        client="claude",
+        tool_name="Edit",
+        signals={"is_edit": True, "target_unread": True},
+        edit_scores={"edit_premature": 0.2},
+        gate_scores={},
+    )
+    assert decision.action == "inject"
+    assert "read" in decision.context
+
+
+def test_stop_blocks_once_and_respects_the_loop_guard():
+    scores = {"claims_done_without_verification": 0.8}
+    signals = {"edited_without_verification": True}
+    blocked = decide_stop(
+        enabled=True,
+        stop_hook_active=False,
+        already_blocked=False,
+        signals=signals,
+        scores=scores,
+    )
+    assert blocked.action == "block"
+    assert decide_stop(
+        enabled=True,
+        stop_hook_active=True,
+        already_blocked=False,
+        signals=signals,
+        scores=scores,
+    ).action == "silent"
+    assert decide_stop(
+        enabled=True,
+        stop_hook_active=False,
+        already_blocked=True,
+        signals=signals,
+        scores=scores,
+    ).action == "silent"