feat: add HITL approval for drafted directives

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:

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,