From 579b1e24e7e2fa35060d2dd9931546e866c3310a Mon Sep 17 00:00:00 2001 From: Nick Hollon Date: Tue, 12 May 2026 15:45:34 -0700 Subject: [PATCH] test(langchain): assert `interrupt_when` exceptions propagate --- .../implementations/test_human_in_the_loop.py | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_human_in_the_loop.py b/libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_human_in_the_loop.py index f2436224adf..d8ab49f73a2 100644 --- a/libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_human_in_the_loop.py +++ b/libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_human_in_the_loop.py @@ -1079,3 +1079,30 @@ def test_interrupt_when_tool_runtime_fields() -> None: assert rt.server_info is None assert captured["tool_call"]["id"] == "abc" assert captured["tool_call"]["args"] == {"path": "/x"} + + +def test_interrupt_when_exceptions_propagate() -> None: + """A predicate that raises propagates the exception (no silent auto-approval).""" + + class PredicateBug(RuntimeError): + pass + + def bad_predicate(_tc: Any, _rt: Any) -> bool: + raise PredicateBug("kaboom") + + middleware = HumanInTheLoopMiddleware( + interrupt_on={ + "edit_file": { + "allowed_decisions": ["approve"], + "interrupt_when": bad_predicate, + } + } + ) + ai_message = AIMessage( + content="One call", + tool_calls=[{"name": "edit_file", "args": {"path": "/x"}, "id": "1"}], + ) + state = AgentState[Any](messages=[HumanMessage(content="Hi"), ai_message]) + + with pytest.raises(PredicateBug, match="kaboom"): + middleware.after_model(state, Runtime())