From 522f6eccea777bf6a699b3875592850bdc114b96 Mon Sep 17 00:00:00 2001 From: Robert Lippmann Date: Mon, 10 Aug 2026 02:52:48 -0400 Subject: [PATCH] feat: add HITL approval for drafted directives --- .../litellm/with_directive_drafter.py | 28 ++++++- .../test_litellm_with_directive_drafter.py | 76 ++++++++++++++++++- 2 files changed, 100 insertions(+), 4 deletions(-) diff --git a/python/examples/prompt_construction/litellm/with_directive_drafter.py b/python/examples/prompt_construction/litellm/with_directive_drafter.py index 7156ba1..58bbc8b 100644 --- a/python/examples/prompt_construction/litellm/with_directive_drafter.py +++ b/python/examples/prompt_construction/litellm/with_directive_drafter.py @@ -56,6 +56,9 @@ class _LiteLLMCallKwargs(TypedDict, total=False): api_base: str +ApprovalHandler = Callable[[str], bool] + + def _extract_response_content(response: object) -> str | None: if isinstance(response, Mapping): choices = response.get("choices") @@ -293,7 +296,18 @@ def _append_trace( return f"{response_text}\n\n{trace_text}" -def handle_turn(user_input: str, engine: Engine) -> str: +def _default_approval_handler(directive_text: str) -> bool: + print("This is what I think the directive is:") + print(directive_text) + response = input("Apply it? (y/n)\n") + return response.strip().lower() == "y" + + +def handle_turn( + user_input: str, + engine: Engine, + approval_handler: ApprovalHandler = _default_approval_handler, +) -> str: state_before = (engine.premise, dict(engine.policies)) preprocessd = _preprocess_user_input(user_input) if preprocessd is None: @@ -312,6 +326,18 @@ def handle_turn(user_input: str, engine: Engine) -> str: compile_input = preprocessd logger.debug("preprocessor: engine_input=directive") + approved = approval_handler(compile_input) + if not approved: + return _append_trace( + "Directive rejected. No state change applied.", + original_input=user_input, + compiler_input=compile_input, + preprocessor_output=preprocessd, + decision={"kind": DecisionKind.NO_DIRECTIVE.value, "message": None}, + state_before=state_before, + state_after=(engine.premise, dict(engine.policies)), + llm_called=False, + ) decision = engine.step(compile_input) if decision["kind"] == DecisionKind.ERROR: diff --git a/python/tests/test_litellm_with_directive_drafter.py b/python/tests/test_litellm_with_directive_drafter.py index 0467fe1..695e153 100644 --- a/python/tests/test_litellm_with_directive_drafter.py +++ b/python/tests/test_litellm_with_directive_drafter.py @@ -39,13 +39,51 @@ def step_with_capture(user_input: str): DirectiveDrafter(fallback=lambda _message: "use docker"), ) - result = module.handle_turn("please use docker", engine) + result = module.handle_turn( + "please use docker", engine, approval_handler=lambda _directive: True + ) assert result == "State updated." assert compile_inputs == ["use docker"] -def test_unknown_or_unsafe_drafting_falls_back_to_raw_input(monkeypatch) -> None: +def test_rejected_canonical_directive_does_not_call_engine_step_or_mutate_state( + monkeypatch, +) -> None: + engine = create_engine() + compile_inputs: list[str] = [] + real_step = engine.step + + def step_with_capture(user_input: str): + compile_inputs.append(user_input) + return real_step(user_input) + + monkeypatch.setattr(engine, "step", step_with_capture) + monkeypatch.setattr( + module, + "_DIRECTIVE_DRAFTER", + DirectiveDrafter(fallback=lambda _message: "use docker"), + ) + llm_calls: list[list[dict[str, str]]] = [] + + def should_not_call(messages: list[dict[str, str]]) -> str: + llm_calls.append(messages) + return "should not be called" + + monkeypatch.setattr(module, "_call_litellm", should_not_call) + + result = module.handle_turn( + "please use docker", engine, approval_handler=lambda _directive: False + ) + + assert compile_inputs == [] + assert result == "Directive rejected. No state change applied." + assert llm_calls == [] + assert engine.policies == {} + assert engine.premise is None + + +def test_no_directive_keeps_normal_flow(monkeypatch) -> None: engine = create_engine() compile_inputs: list[str] = [] llm_calls: list[list[dict[str, str]]] = [] @@ -75,6 +113,36 @@ def downstream(messages: list[dict[str, str]]) -> str: assert len(llm_calls) == 1 +def test_unknown_directive_keeps_normal_flow(monkeypatch) -> None: + engine = create_engine() + compile_inputs: list[str] = [] + llm_calls: list[list[dict[str, str]]] = [] + real_step = engine.step + + def step_with_capture(user_input: str): + compile_inputs.append(user_input) + return real_step(user_input) + + monkeypatch.setattr(engine, "step", step_with_capture) + monkeypatch.setattr( + module, + "_DIRECTIVE_DRAFTER", + DirectiveDrafter(fallback=lambda _message: "use docker and prohibit peanuts"), + ) + + def downstream(messages: list[dict[str, str]]) -> str: + llm_calls.append(messages) + return "stubbed reply" + + monkeypatch.setattr(module, "_call_litellm", downstream) + + result = module.handle_turn("please use docker", engine) + + assert compile_inputs == [] + assert result == "stubbed reply" + assert len(llm_calls) == 1 + + def test_extract_drafted_text_observes_draft_result_behavior() -> None: drafter = DirectiveDrafter(fallback=lambda _message: "use docker") drafted_result = drafter.draft_directive("please use docker") @@ -124,7 +192,9 @@ def should_not_call(messages: list[dict[str, str]]) -> str: ) update_engine = create_engine() - update = module.handle_turn("please use docker", update_engine) + update = module.handle_turn( + "please use docker", update_engine, approval_handler=lambda _directive: True + ) monkeypatch.setattr( module,