Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion packages/mutiny_core/src/mutiny_core/policy/evaluator.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,7 @@ def _eval_require_args(
),
)

matched_when = False
for call in relevant:
when_ok = True
if rule.when:
Expand All @@ -106,6 +107,7 @@ def _eval_require_args(
)
if not when_ok:
continue
matched_when = True

require_ok, failed = matches_constraint_map(
call.arguments, rule.require, context=context
Expand Down Expand Up @@ -141,7 +143,7 @@ def _eval_require_args(
f"'{rule.tool}'"
),
tool_name=rule.tool,
matched_when=False,
matched_when=matched_when,
),
)

Expand Down
45 changes: 40 additions & 5 deletions tests/unit/test_policy_evaluator.py
Original file line number Diff line number Diff line change
Expand Up @@ -153,6 +153,7 @@ def test_violation_amount_over_200_unapproved(self):
hit = _hit_for(hits, "refund_limit")
assert hit.violated is True
assert hit.proximity == 1.0
assert hit.evidence.matched_when is True
assert "approved" in hit.evidence.message.lower() or hit.evidence.failed_constraints

def test_ok_amount_over_200_approved(self):
Expand All @@ -162,7 +163,10 @@ def test_ok_amount_over_200_approved(self):
_trace(_tool("issue_refund", {"order_id": "o1", "amount": 850, "approved": True})),
{},
)
assert _hit_for(hits, "refund_limit").violated is False
hit = _hit_for(hits, "refund_limit")
assert hit.violated is False
assert hit.proximity == 0.0
assert hit.evidence.matched_when is True

def test_ok_amount_at_boundary_200_unapproved(self):
"""gt 200 — amount==200 does not trigger when."""
Expand All @@ -172,7 +176,32 @@ def test_ok_amount_at_boundary_200_unapproved(self):
_trace(_tool("issue_refund", {"order_id": "o1", "amount": 200, "approved": False})),
{},
)
assert _hit_for(hits, "refund_limit").violated is False
hit = _hit_for(hits, "refund_limit")
assert hit.violated is False
assert hit.evidence.matched_when is False

@pytest.mark.parametrize(
"refunds,violated",
[
([(850, True), (50, False)], False),
([(50, False), (850, True)], False),
([(850, True), (500, True)], False),
([(850, True), (500, False)], True),
],
ids=["match-first", "match-last", "all-match", "later-violation"],
)
def test_matched_when_across_calls(self, refunds, violated):
calls = [
_tool("issue_refund", {"amount": amount, "approved": approved}, f"tc-{i}")
for i, (amount, approved) in enumerate(refunds)
]
hits = PolicyEvaluator().evaluate(_refund_limit_policy(), _trace(*calls), {})
hit = _hit_for(hits, "refund_limit")
assert hit.violated is violated
assert hit.proximity == (1.0 if violated else 0.0)
assert hit.evidence.matched_when is True
if violated:
assert hit.evidence.tool_call_id == "tc-1"

def test_violation_amount_just_over_boundary(self):
ev = PolicyEvaluator()
Expand All @@ -199,7 +228,9 @@ def test_when_not_matched_if_amount_missing(self):
_trace(_tool("issue_refund", {"order_id": "o1", "approved": False})),
{},
)
assert _hit_for(hits, "refund_limit").violated is False
hit = _hit_for(hits, "refund_limit")
assert hit.violated is False
assert hit.evidence.matched_when is False

def test_violation_when_matched_but_approved_missing(self):
ev = PolicyEvaluator()
Expand All @@ -217,7 +248,9 @@ def test_other_tool_ignored(self):
_trace(_tool("delete_account", {"confirmed": False})),
{},
)
assert _hit_for(hits, "refund_limit").violated is False
hit = _hit_for(hits, "refund_limit")
assert hit.violated is False
assert hit.evidence.matched_when is None

def test_violation_numeric_string_amount_over_boundary(self):
ev = PolicyEvaluator()
Expand Down Expand Up @@ -287,7 +320,9 @@ def test_ok_confirmed_true(self):
_trace(_tool("delete_account", {"confirmed": True})),
{},
)
assert _hit_for(hits, "delete_requires_confirm").violated is False
hit = _hit_for(hits, "delete_requires_confirm")
assert hit.violated is False
assert hit.evidence.matched_when is True

def test_violation_confirmed_missing(self):
ev = PolicyEvaluator()
Expand Down
Loading